impl Sink for try_stream combinators

Author: taiki-e

futures-util/src/try_stream/err_into.rs

@@ -2,6 +2,7 @@ use core::marker::PhantomData;
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream, TryStream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::unsafe_pinned;
 
 /// Stream for the [`err_into`](super::TryStreamExt::err_into) method.
@@ -43,3 +44,14 @@ where
             .map(|res| res.map(|some| some.map_err(Into::into)))
     }
 }
+
+// Forwarding impl of Sink from the underlying stream
+impl<S, E, Item> Sink<Item> for ErrInto<S, E>
+where
+    S: TryStream + Sink<Item>,
+    S::Error: Into<E>,
+{
+    type SinkError = S::SinkError;
+
+    delegate_sink!(stream, Item);
+}

futures-util/src/try_stream/into_stream.rs

@@ -1,6 +1,7 @@
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream, TryStream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::unsafe_pinned;
 
 /// Stream for the [`into_stream`](super::TryStreamExt::into_stream) method.
@@ -53,3 +54,10 @@ impl<St: TryStream> Stream for IntoStream<St> {
         self.stream().try_poll_next(waker)
     }
 }
+
+// Forwarding impl of Sink from the underlying stream
+impl<S: TryStream + Sink<Item>, Item> Sink<Item> for IntoStream<S> {
+    type SinkError = S::SinkError;
+
+    delegate_sink!(stream, Item);
+}

futures-util/src/try_stream/try_buffer_unordered.rs

@@ -4,6 +4,7 @@ use crate::try_stream::IntoStream;
 use futures_core::future::TryFuture;
 use futures_core::stream::{Stream, TryStream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 use core::pin::Pin;
 
@@ -96,3 +97,13 @@ impl<St> Stream for TryBufferUnordered<St>
         }
     }
 }
+
+// Forwarding impl of Sink from the underlying stream
+impl<S, Item> Sink<Item> for TryBufferUnordered<S>
+    where S: TryStream + Sink<Item>,
+          S::Ok: TryFuture<Error = S::Error>,
+{
+    type SinkError = S::SinkError;
+
+    delegate_sink!(stream, Item);
+}

futures-util/src/try_stream/try_filter_map.rs

@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::{TryFuture};
 use futures_core::stream::{Stream, TryStream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// Stream for the [`try_filter_map`](super::TryStreamExt::try_filter_map)
@@ -84,3 +85,13 @@ impl<St, Fut, F, T> Stream for TryFilterMap<St, Fut, F>
     }
 }
 
+// Forwarding impl of Sink from the underlying stream
+impl<S, Fut, F, T, Item> Sink<Item> for TryFilterMap<S, Fut, F>
+    where S: TryStream + Sink<Item>,
+          Fut: TryFuture<Ok = Option<T>, Error = S::Error>,
+          F: FnMut(S::Ok) -> Fut,
+{
+    type SinkError = S::SinkError;
+
+    delegate_sink!(stream, Item);
+}