Refactor: Rebase & update

Cargo.toml

@@ -24,6 +24,7 @@ clap = { version = "4.1.11", features = ["derive", "env"] }
 derive_more = { version="0.99.9" }
 futures = "0.3"
 lazy_static = "1.4.0"
+local-sync = "0.1"
 maplit = "1.0.2"
 monoio = "0.2.2"
 pretty_assertions = "1.0.0"

openraft/Cargo.toml

@@ -27,6 +27,7 @@ serde           = { workspace = true, optional = true }
 serde_json      = { workspace = true, optional = true }
 tempfile        = { workspace = true, optional = true }
 monoio          = { workspace = true, optional = true }
+local-sync      = { workspace = true, optional = true }
 thiserror       = { workspace = true }
 tokio           = { workspace = true }
 tracing         = { workspace = true }
@@ -79,8 +80,9 @@ compat = []
 compat-07 = ["compat", "serde", "dep:or07", "compat-07-testing"]
 compat-07-testing = ["dep:tempfile", "anyhow", "dep:serde_json"]
 
-monoio = ["dep:monoio", "singlethreaded"]
-monoio-sync = ["dep:monoio", "monoio/sync"]
+# Turn on monoio Runtime with singlethreaded monoio tasks
+monoio = ["dep:monoio", "singlethreaded", "dep:local-sync"]
+# monoio-sync = ["dep:monoio", "monoio/sync"]
 
 # Allows an application to implement a custom the v2 storage API.
 # See `openraft::storage::v2` for more details.

openraft/src/async_runtime/mod.rs

@@ -21,7 +21,7 @@ use crate::OptionalSync;
 /// ## Note
 ///
 /// The default asynchronous runtime is `tokio`.
-pub trait AsyncRuntime: Debug + Default + OptionalSend + OptionalSync + 'static {
+pub trait AsyncRuntime: Debug + Default + PartialEq + Eq + OptionalSend + OptionalSync + 'static {
     /// The error type of [`Self::JoinHandle`].
     type JoinError: Debug + Display + OptionalSend;
 
@@ -47,6 +47,18 @@ pub trait AsyncRuntime: Debug + Default + OptionalSend + OptionalSync + 'static
     /// Type of a thread-local random number generator.
     type ThreadLocalRng: rand::Rng;
 
+    /// Type of a `oneshot` sender.
+    type OneshotSender<T: OptionalSend>: AsyncOneshotSendExt<T> + OptionalSend + OptionalSync + Debug + Sized;
+
+    /// Type of a `oneshot` receiver error.
+    type OneshotReceiverError: std::error::Error + OptionalSend;
+
+    /// Type of a `oneshot` receiver.
+    type OneshotReceiver<T: OptionalSend>: OptionalSend
+        + OptionalSync
+        + Future<Output = Result<T, Self::OneshotReceiverError>>
+        + Unpin;
+
     /// Spawn a new task.
     fn spawn<T>(future: T) -> Self::JoinHandle<T::Output>
     where
@@ -75,4 +87,26 @@ pub trait AsyncRuntime: Debug + Default + OptionalSend + OptionalSync + 'static
     /// This is a per-thread instance, which cannot be shared across threads or
     /// sent to another thread.
     fn thread_rng() -> Self::ThreadLocalRng;
+
+    /// Creates a new one-shot channel for sending single values.
+    ///
+    /// The function returns separate "send" and "receive" handles. The `Sender`
+    /// handle is used by the producer to send the value. The `Receiver` handle is
+    /// used by the consumer to receive the value.
+    ///
+    /// Each handle can be used on separate tasks.
+    fn oneshot<T>() -> (Self::OneshotSender<T>, Self::OneshotReceiver<T>)
+    where T: OptionalSend;
+}
+
+pub trait AsyncOneshotSendExt<T>: Unpin {
+    /// Attempts to send a value on this channel, returning it back if it could
+    /// not be sent.
+    ///
+    /// This method consumes `self` as only one value may ever be sent on a `oneshot`
+    /// channel. It is not marked async because sending a message to an `oneshot`
+    /// channel never requires any form of waiting.  Because of this, the `send`
+    /// method can be used in both synchronous and asynchronous code without
+    /// problems.
+    fn send(self, t: T) -> Result<(), T>;
 }