impl Sink for stream combinators

taiki-e · Nemo157 · commit d5a3976bd5bf · 2019-03-27T13:24:44.000+01:00
diff --git a/futures-util/src/stream/chunks.rs b/futures-util/src/stream/chunks.rs
@@ -1,6 +1,7 @@
 use crate::stream::Fuse;
 use futures_core::stream::Stream;
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 use core::mem;
 use core::pin::Pin;
@@ -100,14 +101,12 @@ impl<St: Stream> Stream for Chunks<St> {
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S> Sink for Chunks<S>
-    where S: Sink + Stream
+impl<S, Item> Sink<Item> for Chunks<S>
+where
+    S: Stream + Sink<Item>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/filter.rs b/futures-util/src/stream/filter.rs
@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator used to filter the results of a stream and only yield
@@ -113,17 +114,13 @@ impl<St, Fut, F> Stream for Filter<St, Fut, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, R, P> Sink for Filter<S, R, P>
-    where S: Stream,
-          P: FnMut(&S::Item) -> R,
-          R: Future<Item = bool>,
-          S: Sink,
+impl<S, Fut, F, Item> Sink<Item> for Filter<S, Fut, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(&S::Item) -> Fut,
+          Fut: Future<Output = bool>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/filter_map.rs b/futures-util/src/stream/filter_map.rs
@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A combinator used to filter the results of a stream and simultaneously map
@@ -103,16 +104,13 @@ impl<St, Fut, F, T> Stream for FilterMap<St, Fut, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, R, F> Sink for FilterMap<S, R, F>
-    where S: Stream + Sink,
-          F: FnMut(S::Item) -> R,
-          R: IntoFuture<Error=S::Error>,
+impl<S, Fut, F, Item> Sink<Item> for FilterMap<S, Fut, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(S::Item) -> Fut,
+          Fut: Future,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/flatten.rs b/futures-util/src/stream/flatten.rs
@@ -1,6 +1,7 @@
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::unsafe_pinned;
 
 /// A combinator used to flatten a stream-of-streams into one long stream of
@@ -89,14 +90,12 @@ impl<St> Stream for Flatten<St>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S> Sink for Flatten<S>
-    where S: Sink + Stream
+impl<S, Item> Sink<Item> for Flatten<S>
+    where S: Stream + Sink<Item>,
+          S::Item: Stream,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
- */
diff --git a/futures-util/src/stream/inspect.rs b/futures-util/src/stream/inspect.rs
@@ -1,6 +1,7 @@
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// Do something with the items of a stream, passing it on.
@@ -17,7 +18,7 @@ impl<St: Stream + Unpin, F> Unpin for Inspect<St, F> {}
 
 impl<St, F> Inspect<St, F>
     where St: Stream,
-          F: FnMut(&St::Item) -> (),
+          F: FnMut(&St::Item),
 {
     unsafe_pinned!(stream: St);
     unsafe_unpinned!(f: F);
@@ -74,14 +75,12 @@ impl<St, F> Stream for Inspect<St, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, F> Sink for Inspect<S, F>
-    where S: Sink + Stream
+impl<S, F, Item> Sink<Item> for Inspect<S, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(&S::Item),
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/map.rs b/futures-util/src/stream/map.rs
@@ -1,6 +1,7 @@
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator which will change the type of a stream from one
@@ -72,14 +73,12 @@ impl<St, F, T> Stream for Map<St, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, F> Sink for Map<S, F>
-    where S: Sink
+impl<S, F, T, Item> Sink<Item> for Map<S, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(S::Item) -> T,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/peek.rs b/futures-util/src/stream/peek.rs
@@ -2,6 +2,7 @@ use crate::stream::{StreamExt, Fuse};
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A `Stream` that implements a `peek` method.
@@ -72,14 +73,11 @@ impl<S: Stream> Stream for Peekable<S> {
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S> Sink for Peekable<S>
-    where S: Sink + Stream
+impl<S, Item> Sink<Item> for Peekable<S>
+    where S: Sink<Item> + Stream
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/skip.rs b/futures-util/src/stream/skip.rs
@@ -1,6 +1,7 @@
 use core::pin::Pin;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator which skips a number of elements before continuing.
@@ -74,14 +75,12 @@ impl<St: Stream> Stream for Skip<St> {
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S> Sink for Skip<S>
-    where S: Sink
+impl<S, Item> Sink<Item> for Skip<S>
+where
+    S: Stream + Sink<Item>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/skip_while.rs b/futures-util/src/stream/skip_while.rs
@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator which skips elements of a stream while a predicate
@@ -109,14 +110,13 @@ impl<St, Fut, F> Stream for SkipWhile<St, Fut, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, R, P> Sink for SkipWhile<S, R, P>
-    where S: Sink + Stream, R: IntoFuture
+impl<S, Fut, F, Item> Sink<Item> for SkipWhile<S, Fut, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(&S::Item) -> Fut,
+          Fut: Future<Output = bool>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/take.rs b/futures-util/src/stream/take.rs
@@ -1,6 +1,7 @@
 use core::pin::Pin;
 use futures_core::stream::Stream;
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator which returns a maximum number of elements.
@@ -72,14 +73,11 @@ impl<St> Stream for Take<St>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S> Sink for Take<S>
-    where S: Sink + Stream
+impl<S, Item> Sink<Item> for Take<S>
+    where S: Stream + Sink<Item>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/take_while.rs b/futures-util/src/stream/take_while.rs
@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::Stream;
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator which takes elements from a stream while a predicate
@@ -103,14 +104,13 @@ impl<St, Fut, F> Stream for TakeWhile<St, Fut, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, R, P> Sink for TakeWhile<S, R, P>
-    where S: Sink + Stream, R: IntoFuture
+impl<S, Fut, F, Item> Sink<Item> for TakeWhile<S, Fut, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(&S::Item) -> Fut,
+          Fut: Future<Output = bool>,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
-*/
diff --git a/futures-util/src/stream/then.rs b/futures-util/src/stream/then.rs
@@ -2,6 +2,7 @@ use core::pin::Pin;
 use futures_core::future::Future;
 use futures_core::stream::{FusedStream, Stream};
 use futures_core::task::{Waker, Poll};
+use futures_sink::Sink;
 use pin_utils::{unsafe_pinned, unsafe_unpinned};
 
 /// A stream combinator which chains a computation onto each item produced by a
@@ -67,14 +68,13 @@ impl<St, Fut, F> Stream for Then<St, Fut, F>
     }
 }
 
-/* TODO
 // Forwarding impl of Sink from the underlying stream
-impl<S, U, F> Sink for Then<S, U, F>
-    where S: Sink, U: IntoFuture,
+impl<S, Fut, F, Item> Sink<Item> for Then<S, Fut, F>
+    where S: Stream + Sink<Item>,
+          F: FnMut(S::Item) -> Fut,
+          Fut: Future,
 {
-    type SinkItem = S::SinkItem;
     type SinkError = S::SinkError;
 
-    delegate_sink!(stream);
+    delegate_sink!(stream, Item);
 }
- */
diff --git a/futures-util/src/try_stream/map_err.rs b/futures-util/src/try_stream/map_err.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, unsafe_unpinned};
 
 /// Stream for the [`map_err`](super::TryStreamExt::map_err) combinator.
@@ -48,3 +49,14 @@ where
         }
     }
 }
+
+// Forwarding impl of Sink from the underlying stream
+impl<S, F, E, Item> Sink<Item> for MapErr<S, F>
+where
+    S: TryStream + Sink<Item>,
+    F: FnMut(S::Error) -> E,
+{
+    type SinkError = S::SinkError;
+
+    delegate_sink!(stream, Item);
+}
diff --git a/futures-util/src/try_stream/map_ok.rs b/futures-util/src/try_stream/map_ok.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, unsafe_unpinned};
 
 /// Stream for the [`map_ok`](super::TryStreamExt::map_ok) combinator.
@@ -48,3 +49,14 @@ where
         }
     }
 }
+
+// Forwarding impl of Sink from the underlying stream
+impl<S, F, T, Item> Sink<Item> for MapOk<S, F>
+where
+    S: TryStream + Sink<Item>,
+    F: FnMut(S::Ok) -> T,
+{
+    type SinkError = S::SinkError;
+
+    delegate_sink!(stream, Item);
+}
diff --git a/futures-util/src/try_stream/try_skip_while.rs b/futures-util/src/try_stream/try_skip_while.rs

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`use crate::stream::Fuse;`
`2`	`2`	`use futures_core::stream::Stream;`
`3`	`3`	`use futures_core::task::{Waker, Poll};`
	`4`	`+use futures_sink::Sink;`
`4`	`5`	`use pin_utils::{unsafe_pinned, unsafe_unpinned};`
`5`	`6`	`use core::mem;`
`6`	`7`	`use core::pin::Pin;`
`@@ -100,14 +101,12 @@ impl<St: Stream> Stream for Chunks<St> {`
`100`	`101`	`}`
`101`	`102`	`}`
`102`	`103`
`103`		`-/* TODO`
`104`	`104`	`// Forwarding impl of Sink from the underlying stream`
`105`		`-impl<S> Sink for Chunks<S>`
`106`		`- where S: Sink + Stream`
	`105`	`+impl<S, Item> Sink<Item> for Chunks<S>`
	`106`	`+where`
	`107`	`+ S: Stream + Sink<Item>,`
`107`	`108`	`{`
`108`		`- type SinkItem = S::SinkItem;`
`109`	`109`	`type SinkError = S::SinkError;`
`110`	`110`
`111`		`- delegate_sink!(stream);`
	`111`	`+ delegate_sink!(stream, Item);`
`112`	`112`	`}`
`113`		`-*/`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`use core::pin::Pin;`
`2`	`2`	`use futures_core::stream::{FusedStream, Stream};`
`3`	`3`	`use futures_core::task::{Waker, Poll};`
	`4`	`+use futures_sink::Sink;`
`4`	`5`	`use pin_utils::unsafe_pinned;`
`5`	`6`
`6`	`7`	`/// A combinator used to flatten a stream-of-streams into one long stream of`
`@@ -89,14 +90,12 @@ impl<St> Stream for Flatten<St>`
`89`	`90`	`}`
`90`	`91`	`}`
`91`	`92`
`92`		`-/* TODO`
`93`	`93`	`// Forwarding impl of Sink from the underlying stream`
`94`		`-impl<S> Sink for Flatten<S>`
`95`		`- where S: Sink + Stream`
	`94`	`+impl<S, Item> Sink<Item> for Flatten<S>`
	`95`	`+ where S: Stream + Sink<Item>,`
	`96`	`+ S::Item: Stream,`
`96`	`97`	`{`
`97`		`- type SinkItem = S::SinkItem;`
`98`	`98`	`type SinkError = S::SinkError;`
`99`	`99`
`100`		`- delegate_sink!(stream);`
	`100`	`+ delegate_sink!(stream, Item);`
`101`	`101`	`}`
`102`		`- */`
Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@ use crate::stream::{StreamExt, Fuse};`
`2`	`2`	`use core::pin::Pin;`
`3`	`3`	`use futures_core::stream::{FusedStream, Stream};`
`4`	`4`	`use futures_core::task::{Waker, Poll};`
	`5`	`+use futures_sink::Sink;`
`5`	`6`	`use pin_utils::{unsafe_pinned, unsafe_unpinned};`
`6`	`7`
`7`	`8`	/// A `Stream` that implements a `peek` method.
`@@ -72,14 +73,11 @@ impl<S: Stream> Stream for Peekable<S> {`
`72`	`73`	`}`
`73`	`74`	`}`
`74`	`75`
`75`		`-/* TODO`
`76`	`76`	`// Forwarding impl of Sink from the underlying stream`
`77`		`-impl<S> Sink for Peekable<S>`
`78`		`- where S: Sink + Stream`
	`77`	`+impl<S, Item> Sink<Item> for Peekable<S>`
	`78`	`+ where S: Sink<Item> + Stream`
`79`	`79`	`{`
`80`		`- type SinkItem = S::SinkItem;`
`81`	`80`	`type SinkError = S::SinkError;`
`82`	`81`
`83`		`- delegate_sink!(stream);`
	`82`	`+ delegate_sink!(stream, Item);`
`84`	`83`	`}`
`85`		`-*/`
Original file line number	Diff line number	Diff line change
`@@ -2,6 +2,7 @@ use core::pin::Pin;`
`2`	`2`	`use futures_core::future::Future;`
`3`	`3`	`use futures_core::stream::{FusedStream, Stream};`
`4`	`4`	`use futures_core::task::{Waker, Poll};`
	`5`	`+use futures_sink::Sink;`
`5`	`6`	`use pin_utils::{unsafe_pinned, unsafe_unpinned};`
`6`	`7`
`7`	`8`	`/// A stream combinator which skips elements of a stream while a predicate`
`@@ -109,14 +110,13 @@ impl<St, Fut, F> Stream for SkipWhile<St, Fut, F>`
`109`	`110`	`}`
`110`	`111`	`}`
`111`	`112`
`112`		`-/* TODO`
`113`	`113`	`// Forwarding impl of Sink from the underlying stream`
`114`		`-impl<S, R, P> Sink for SkipWhile<S, R, P>`
`115`		`- where S: Sink + Stream, R: IntoFuture`
	`114`	`+impl<S, Fut, F, Item> Sink<Item> for SkipWhile<S, Fut, F>`
	`115`	`+ where S: Stream + Sink<Item>,`
	`116`	`+ F: FnMut(&S::Item) -> Fut,`
	`117`	`+ Fut: Future<Output = bool>,`
`116`	`118`	`{`
`117`		`- type SinkItem = S::SinkItem;`
`118`	`119`	`type SinkError = S::SinkError;`
`119`	`120`
`120`		`- delegate_sink!(stream);`
	`121`	`+ delegate_sink!(stream, Item);`
`121`	`122`	`}`
`122`		`-*/`