diff --git a/tokio/src/sync/mpsc/bounded.rs b/tokio/src/sync/mpsc/bounded.rs index 5a2bfa6e8..ddded8ebb 100644 --- a/tokio/src/sync/mpsc/bounded.rs +++ b/tokio/src/sync/mpsc/bounded.rs @@ -563,6 +563,11 @@ impl Sender { /// [`close`]: Receiver::close /// [`Receiver`]: Receiver /// + /// # Panics + /// + /// This function panics if it is called outside the context of a Tokio + /// runtime [with time enabled](crate::runtime::Builder::enable_time). + /// /// # Examples /// /// In the following example, each call to `send_timeout` will block until the diff --git a/tokio/src/sync/mpsc/error.rs b/tokio/src/sync/mpsc/error.rs index b7b9cf7a2..3fe6bac5e 100644 --- a/tokio/src/sync/mpsc/error.rs +++ b/tokio/src/sync/mpsc/error.rs @@ -19,7 +19,7 @@ impl std::error::Error for SendError {} /// This enumeration is the list of the possible error outcomes for the /// [try_send](super::Sender::try_send) method. -#[derive(Debug)] +#[derive(Debug, Eq, PartialEq)] pub enum TrySendError { /// The data could not be sent on the channel because the channel is /// currently full and sending would require blocking. @@ -96,7 +96,7 @@ impl Error for RecvError {} cfg_time! { // ===== SendTimeoutError ===== - #[derive(Debug)] + #[derive(Debug, Eq, PartialEq)] /// Error returned by [`Sender::send_timeout`](super::Sender::send_timeout)]. pub enum SendTimeoutError { /// The data could not be sent on the channel because the channel is diff --git a/tokio/src/sync/mpsc/mod.rs b/tokio/src/sync/mpsc/mod.rs index 879e3dcfc..b1513a9da 100644 --- a/tokio/src/sync/mpsc/mod.rs +++ b/tokio/src/sync/mpsc/mod.rs @@ -58,6 +58,22 @@ //! [crossbeam][crossbeam-unbounded]. Similarly, for sending a message _from sync //! to async_, you should use an unbounded Tokio `mpsc` channel. //! +//! Please be aware that the above remarks were written with the `mpsc` channel +//! in mind, but they can also be generalized to other kinds of channels. In +//! general, any channel method that isn't marked async can be called anywhere, +//! including outside of the runtime. For example, sending a message on a +//! oneshot channel from outside the runtime is perfectly fine. +//! +//! # Multiple runtimes +//! +//! The mpsc channel does not care about which runtime you use it in, and can be +//! used to send messages from one runtime to another. It can also be used in +//! non-Tokio runtimes. +//! +//! There is one exception to the above: the [`send_timeout`] must be used from +//! within a Tokio runtime, however it is still not tied to one specific Tokio +//! runtime, and the sender may be moved from one Tokio runtime to another. +//! //! [`Sender`]: crate::sync::mpsc::Sender //! [`Receiver`]: crate::sync::mpsc::Receiver //! [bounded-send]: crate::sync::mpsc::Sender::send() @@ -69,6 +85,7 @@ //! [`Handle::block_on`]: crate::runtime::Handle::block_on() //! [std-unbounded]: std::sync::mpsc::channel //! [crossbeam-unbounded]: https://docs.rs/crossbeam/*/crossbeam/channel/fn.unbounded.html +//! [`send_timeout`]: crate::sync::mpsc::Sender::send_timeout pub(super) mod block; diff --git a/tokio/src/sync/oneshot.rs b/tokio/src/sync/oneshot.rs index 4fb22ecf6..08a2d9e49 100644 --- a/tokio/src/sync/oneshot.rs +++ b/tokio/src/sync/oneshot.rs @@ -9,6 +9,9 @@ //! //! Each handle can be used on separate tasks. //! +//! Since the `send` method is not async, it can be used anywhere. This includes +//! sending between two runtimes, and using it from non-async code. +//! //! # Examples //! //! ``` diff --git a/tokio/tests/sync_mpsc.rs b/tokio/tests/sync_mpsc.rs index 1947d2607..1a7cc7436 100644 --- a/tokio/tests/sync_mpsc.rs +++ b/tokio/tests/sync_mpsc.rs @@ -597,3 +597,37 @@ fn try_recv_close_while_empty_unbounded() { drop(tx); assert_eq!(Err(TryRecvError::Disconnected), rx.try_recv()); } + +#[tokio::test(start_paused = true)] +async fn recv_timeout() { + use tokio::sync::mpsc::error::SendTimeoutError::{Closed, Timeout}; + use tokio::time::Duration; + + let (tx, rx) = mpsc::channel(5); + + assert_eq!(tx.send_timeout(10, Duration::from_secs(1)).await, Ok(())); + assert_eq!(tx.send_timeout(20, Duration::from_secs(1)).await, Ok(())); + assert_eq!(tx.send_timeout(30, Duration::from_secs(1)).await, Ok(())); + assert_eq!(tx.send_timeout(40, Duration::from_secs(1)).await, Ok(())); + assert_eq!(tx.send_timeout(50, Duration::from_secs(1)).await, Ok(())); + assert_eq!( + tx.send_timeout(60, Duration::from_secs(1)).await, + Err(Timeout(60)) + ); + + drop(rx); + assert_eq!( + tx.send_timeout(70, Duration::from_secs(1)).await, + Err(Closed(70)) + ); +} + +#[test] +#[should_panic = "there is no reactor running, must be called from the context of a Tokio 1.x runtime"] +fn recv_timeout_panic() { + use futures::future::FutureExt; + use tokio::time::Duration; + + let (tx, _rx) = mpsc::channel(5); + tx.send_timeout(10, Duration::from_secs(1)).now_or_never(); +}