Merge pull request #915 from async-rs/feat-new-channels

feat: new channels

Author: yoshuawuyts

Cargo.toml

@@ -52,6 +52,7 @@ std = [
   "wasm-bindgen-futures",
   "futures-channel",
   "async-mutex",
+  "async-channel",
 ]
 alloc = [
   "futures-core/alloc",
@@ -74,10 +75,12 @@ once_cell = { version = "1.3.1", optional = true }
 pin-project-lite = { version = "0.2.0", optional = true }
 pin-utils = { version = "0.1.0-alpha.4", optional = true }
 slab = { version = "0.4.2", optional = true }
+async-channel = { version = "1.5.1", optional = true }
 
 # Devdepencency, but they are not allowed to be optional :/
 surf = { version = "2.0.0", optional = true }
 
+
 [target.'cfg(not(target_os = "unknown"))'.dependencies]
 async-global-executor = { version = "1.4.0", optional = true, features = ["async-io"] }
 async-io = { version = "1.0.1", optional = true }

src/channel.rs

@@ -0,0 +1,6 @@
+//! Channels
+
+#[cfg(feature = "unstable")]
+#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[doc(inline)]
+pub use async_channel::*;

src/lib.rs

@@ -106,15 +106,14 @@
 //! [`io`], [`fs`], and [`net`] modules.
 //!
 //! The [`task`] module contains `async-std`'s task abstractions. [`sync`]
-//! contains further primitive shared memory types, including [`channel`],
-//! which contains the channel types for message passing.
+//! contains further primitive shared memory types. [`channel`]  contains the channel types for message passing.
 //!
 //! [files]: fs/struct.File.html
 //! [TCP]: net/struct.TcpStream.html
 //! [UDP]: net/struct.UdpSocket.html
 //! [`io`]: fs/struct.File.html
 //! [`sync`]: sync/index.html
-//! [`channel`]: sync/fn.channel.html
+//! [`channel`]: channel/index.html
 //!
 //! ## Timeouts, intervals, and delays
 //!
@@ -300,6 +299,7 @@ cfg_std! {
     pub mod os;
     pub mod prelude;
     pub mod sync;
+    pub mod channel;
 }
 
 cfg_default! {

src/sync/channel.rs

@@ -1,3 +1,5 @@
+#![allow(deprecated)]
+
 use std::cell::UnsafeCell;
 use std::error::Error;
 use std::fmt::{self, Debug, Display};
@@ -32,6 +34,7 @@ use crate::sync::WakerSet;
 /// # Examples
 ///
 /// ```
+/// #![allow(deprecated)]
 /// # fn main() -> Result<(), async_std::sync::RecvError> {
 /// # async_std::task::block_on(async {
 /// #
@@ -60,6 +63,7 @@ use crate::sync::WakerSet;
 /// ```
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[deprecated = "new channel api at async_std::channel"]
 pub fn channel<T>(cap: usize) -> (Sender<T>, Receiver<T>) {
     let channel = Arc::new(Channel::with_capacity(cap));
     let s = Sender {
@@ -82,6 +86,7 @@ pub fn channel<T>(cap: usize) -> (Sender<T>, Receiver<T>) {
 /// # Examples
 ///
 /// ```
+/// #![allow(deprecated)]
 /// # async_std::task::block_on(async {
 /// #
 /// use async_std::sync::channel;
@@ -102,6 +107,7 @@ pub fn channel<T>(cap: usize) -> (Sender<T>, Receiver<T>) {
 /// ```
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[deprecated = "new channel api at async_std::channel"]
 pub struct Sender<T> {
     /// The inner channel.
     channel: Arc<Channel<T>>,
@@ -115,6 +121,7 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # fn main() -> Result<(), async_std::sync::RecvError> {
     /// # async_std::task::block_on(async {
     /// #
@@ -204,6 +211,7 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -223,6 +231,7 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// use async_std::sync::channel;
     ///
     /// let (s, _) = channel::<i32>(5);
@@ -237,6 +246,7 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -258,6 +268,7 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -279,6 +290,7 @@ impl<T> Sender<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -339,6 +351,7 @@ impl<T> fmt::Debug for Sender<T> {
 /// # Examples
 ///
 /// ```
+/// #![allow(deprecated)]
 /// # fn main() -> Result<(), async_std::sync::RecvError> {
 /// # async_std::task::block_on(async {
 /// #
@@ -363,6 +376,7 @@ impl<T> fmt::Debug for Sender<T> {
 /// ```
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
+#[deprecated = "new channel api at async_std::channel"]
 pub struct Receiver<T> {
     /// The inner channel.
     channel: Arc<Channel<T>>,
@@ -381,6 +395,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # fn main() -> Result<(), async_std::sync::RecvError> {
     /// # async_std::task::block_on(async {
     /// #
@@ -444,6 +459,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -466,6 +482,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// use async_std::sync::channel;
     ///
     /// let (_, r) = channel::<i32>(5);
@@ -480,6 +497,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -501,6 +519,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -522,6 +541,7 @@ impl<T> Receiver<T> {
     /// # Examples
     ///
     /// ```
+    /// #![allow(deprecated)]
     /// # async_std::task::block_on(async {
     /// #
     /// use async_std::sync::channel;
@@ -993,6 +1013,7 @@ impl<T> Drop for Channel<T> {
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
 #[derive(PartialEq, Eq)]
+#[deprecated = "new channel api at async_std::channel"]
 pub enum TrySendError<T> {
     /// The channel is full but not disconnected.
     Full(T),
@@ -1025,6 +1046,7 @@ impl<T> Display for TrySendError<T> {
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
 #[derive(Debug, PartialEq, Eq)]
+#[deprecated = "new channel api at async_std::channel"]
 pub enum TryRecvError {
     /// The channel is empty but not disconnected.
     Empty,
@@ -1048,6 +1070,7 @@ impl Display for TryRecvError {
 #[cfg(feature = "unstable")]
 #[cfg_attr(feature = "docs", doc(cfg(unstable)))]
 #[derive(Debug, PartialEq, Eq)]
+#[deprecated = "new channel api at async_std::channel"]
 pub struct RecvError;
 
 impl Error for RecvError {}

src/sync/mod.rs

@@ -185,6 +185,7 @@ mod rwlock;
 
 cfg_unstable! {
     pub use barrier::{Barrier, BarrierWaitResult};
+    #[allow(deprecated)]
     pub use channel::{channel, Sender, Receiver, RecvError, TryRecvError, TrySendError};
     pub use condvar::Condvar;
 

tests/channel.rs

@@ -1,4 +1,5 @@
 #![cfg(feature = "unstable")]
+#![allow(deprecated)]
 
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;

tests/stream.rs

@@ -5,9 +5,9 @@ use std::task::{Context, Poll};
 
 use pin_project_lite::pin_project;
 
+use async_std::channel::bounded as channel;
 use async_std::prelude::*;
 use async_std::stream;
-use async_std::sync::channel;
 use async_std::task;
 
 #[cfg(target_arch = "wasm32")]
@@ -36,7 +36,7 @@ fn merging_delayed_streams_work() {
 
     task::block_on(async move {
         task::sleep(std::time::Duration::from_millis(500)).await;
-        sender.send(92).await;
+        sender.send(92).await.unwrap();
         drop(sender);
         let xs = t.await;
         assert_eq!(xs, vec![92])
@@ -55,7 +55,7 @@ fn merging_delayed_streams_work() {
 
     task::block_on(async move {
         task::sleep(std::time::Duration::from_millis(500)).await;
-        sender.send(92).await;
+        sender.send(92).await.unwrap();
         drop(sender);
         let xs = t.await;
         assert_eq!(xs, vec![92])