austinjones · austinjones · Apr 14, 2022 · Apr 14, 2022
diff --git a/Cargo.toml b/Cargo.toml
@@ -38,6 +38,7 @@ parking_lot = "0.12"
 futures-test = "0.3"
 tokio = { version = "1.0", features = ["rt", "rt-multi-thread", "macros", "time", "sync"] }
 async-std = { version = "1.9", features = ["attributes"] }
+futures = { version = "0.3", default-features = false }
 criterion = "0.3"
 
 [[bench]]
@@ -71,3 +72,4 @@ harness = false
 [[bench]]
 name = "async_std_channel"
 harness = false
+
diff --git a/examples/futures.rs b/examples/futures.rs
@@ -0,0 +1,37 @@
+use futures::{SinkExt, StreamExt};
+use postage::{mpsc, sink::SendError};
+
+// ignore this.. build issue hack
+macro_rules! tokio_main_if {
+    ($feature:expr => async fn main() $body:block) => {
+        #[tokio::main]
+        async fn main() {
+            #[cfg(feature = $feature)]
+            inner().await;
+        }
+
+        #[cfg(feature = $feature)]
+        async fn inner() {
+            $body
+        }
+    };
+}
+
+tokio_main_if! ("futures-traits" => async fn main() {
+    let (mut tx, mut rx) = mpsc::channel(2);
+
+    tx.send(0usize).await.ok();
+    tx.send(1usize).await.ok();
+
+    println!("Sender says {:?}", rx.next().await);
+    println!("Sender says {:?}", rx.next().await);
+
+    // Let's deal with some errors.
+    tx.send(0usize).await.ok();
+    tx.send(1usize).await.ok();
+
+    // If the channel is closed, no value is returned.
+    // Unfortunately this is due to the design of futures::Sink
+    drop(rx);
+    assert_eq!(Err(SendError(0usize)), tx.send(0usize).await);
+});
diff --git a/src/channels/dispatch.rs b/src/channels/dispatch.rs
@@ -91,15 +91,15 @@ mod impl_futures {
     use std::task::Poll;
 
     impl<T> futures::sink::Sink<T> for super::Sender<T> {
-        type Error = SendError<()>;
+        type Error = SendError<T>;
 
         fn poll_ready(
             self: std::pin::Pin<&mut Self>,
             cx: &mut std::task::Context<'_>,
         ) -> Poll<Result<(), Self::Error>> {
             loop {
                 if self.shared.is_closed() {
-                    return Poll::Ready(Err(SendError(())));
+                    return Poll::Ready(Ok(()));
                 }
 
                 let queue = &self.shared.extension().queue;
@@ -121,12 +121,16 @@ mod impl_futures {
         }
 
         fn start_send(self: std::pin::Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+            if self.shared.is_closed() {
+                return Err(SendError(item));
+            }
+
             let result = self
                 .shared
                 .extension()
                 .queue
                 .push(item)
-                .map_err(|_t| SendError(()));
+                .map_err(|item| SendError(item));
 
             if result.is_ok() {
                 self.shared.notify_receivers();

diff --git a/src/channels/mpsc.rs b/src/channels/mpsc.rs
@@ -88,15 +88,15 @@ mod impl_futures {
     use std::task::Poll;
 
     impl<T> futures::sink::Sink<T> for super::Sender<T> {
-        type Error = SendError<()>;
+        type Error = SendError<T>;
 
         fn poll_ready(
             self: std::pin::Pin<&mut Self>,
             cx: &mut std::task::Context<'_>,
         ) -> Poll<Result<(), Self::Error>> {
             loop {
                 if self.shared.is_closed() {
-                    return Poll::Ready(Err(SendError(())));
+                    return Poll::Ready(Ok(()));
                 }
 
                 let queue = &self.shared.extension().queue;
@@ -118,12 +118,16 @@ mod impl_futures {
         }
 
         fn start_send(self: std::pin::Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
+            if self.shared.is_closed() {
+                return Err(SendError(item));
+            }
+
             let result = self
                 .shared
                 .extension()
                 .queue
                 .push(item)
-                .map_err(|_t| SendError(()));
+                .map_err(|item| SendError(item));
 
             if result.is_ok() {
                 self.shared.notify_receivers();

diff --git a/src/channels/watch.rs b/src/channels/watch.rs
@@ -120,22 +120,18 @@ mod impl_futures {
     use crate::sink::SendError;
 
     impl<T> futures::sink::Sink<T> for super::Sender<T> {
-        type Error = SendError<()>;
+        type Error = SendError<T>;
 
         fn poll_ready(
             self: std::pin::Pin<&mut Self>,
             _cx: &mut std::task::Context<'_>,
         ) -> Poll<Result<(), Self::Error>> {
-            if self.shared.is_closed() {
-                return Poll::Ready(Err(SendError(())));
-            }
-
             Poll::Ready(Ok(()))
         }
 
         fn start_send(self: std::pin::Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
             if self.shared.is_closed() {
-                return Err(SendError(()));
+                return Err(SendError(item));
             }
 
             self.shared.extension().push(item);

diff --git a/src/futures.rs b/src/futures.rs
@@ -103,10 +103,10 @@ mod sink_tests {
 
             drop(rx);
             assert_eq!(
-                Poll::Ready(Err(SendError(()))),
+                Poll::Ready(Ok(())),
                 Pin::new(&mut tx).poll_ready(&mut std_cx)
             );
-            assert_eq!(Err(SendError(())), Pin::new(&mut tx).start_send($val));
+            assert_eq!(Err(SendError($val)), Pin::new(&mut tx).start_send($val));
         };
     }