Change Sink's associated item type to a generic Sink<Item>

Author: seanmonstar

futures-sink/src/channel_impls.rs

@@ -3,8 +3,7 @@ use futures_core::task::Waker;
 use futures_channel::mpsc::{Sender, SendError, TrySendError, UnboundedSender};
 use std::pin::Pin;
 
-impl<T> Sink for Sender<T> {
-    type SinkItem = T;
+impl<T> Sink<T> for Sender<T> {
     type SinkError = SendError;
 
     fn poll_ready(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {
@@ -25,8 +24,7 @@ impl<T> Sink for Sender<T> {
     }
 }
 
-impl<T> Sink for UnboundedSender<T> {
-    type SinkItem = T;
+impl<T> Sink<T> for UnboundedSender<T> {
     type SinkError = SendError;
 
     fn poll_ready(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {
@@ -47,8 +45,7 @@ impl<T> Sink for UnboundedSender<T> {
     }
 }
 
-impl<'a, T> Sink for &'a UnboundedSender<T> {
-    type SinkItem = T;
+impl<'a, T> Sink<T> for &'a UnboundedSender<T> {
     type SinkError = SendError;
 
     fn poll_ready(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {

futures-sink/src/lib.rs

@@ -48,10 +48,7 @@ use core::pin::Pin;
 /// importance: you can use it to send an entire stream to a sink, which is
 /// the simplest way to ultimately consume a stream.
 #[must_use = "sinks do nothing unless polled"]
-pub trait Sink {
-    /// The type of value that the sink accepts.
-    type SinkItem;
-
+pub trait Sink<Item> {
     /// The type of value produced by the sink when an error occurs.
     type SinkError;
 
@@ -88,7 +85,7 @@ pub trait Sink {
     ///
     /// In most cases, if the sink encounters an error, the sink will
     /// permanently be unable to receive items.
-    fn start_send(self: Pin<&mut Self>, item: Self::SinkItem)
+    fn start_send(self: Pin<&mut Self>, item: Item)
                   -> Result<(), Self::SinkError>;
 
     /// Flush any remaining output from this sink.
@@ -119,15 +116,14 @@ pub trait Sink {
     fn poll_close(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>>;
 }
 
-impl<'a, S: ?Sized + Sink + Unpin> Sink for &'a mut S {
-    type SinkItem = S::SinkItem;
+impl<'a, S: ?Sized + Sink<Item> + Unpin, Item> Sink<Item> for &'a mut S {
     type SinkError = S::SinkError;
 
     fn poll_ready(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {
         Pin::new(&mut **self).poll_ready(waker)
     }
 
-    fn start_send(mut self: Pin<&mut Self>, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+    fn start_send(mut self: Pin<&mut Self>, item: Item) -> Result<(), Self::SinkError> {
         Pin::new(&mut **self).start_send(item)
     }
 
@@ -140,15 +136,14 @@ impl<'a, S: ?Sized + Sink + Unpin> Sink for &'a mut S {
     }
 }
 
-impl<'a, S: ?Sized + Sink> Sink for Pin<&'a mut S> {
-    type SinkItem = S::SinkItem;
+impl<'a, S: ?Sized + Sink<Item>, Item> Sink<Item> for Pin<&'a mut S> {
     type SinkError = S::SinkError;
 
     fn poll_ready(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {
         S::poll_ready((*self).as_mut(), waker)
     }
 
-    fn start_send(mut self: Pin<&mut Self>, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+    fn start_send(mut self: Pin<&mut Self>, item: Item) -> Result<(), Self::SinkError> {
         S::start_send((*self).as_mut(), item)
     }
 
@@ -173,15 +168,14 @@ mod if_alloc {
     #[derive(Copy, Clone, Debug)]
     pub enum VecSinkError {}
 
-    impl<T> Sink for ::alloc::vec::Vec<T> {
-        type SinkItem = T;
+    impl<T> Sink<T> for ::alloc::vec::Vec<T> {
         type SinkError = VecSinkError;
 
         fn poll_ready(self: Pin<&mut Self>, _: &Waker) -> Poll<Result<(), Self::SinkError>> {
             Poll::Ready(Ok(()))
         }
 
-        fn start_send(self: Pin<&mut Self>, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+        fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::SinkError> {
             // TODO: impl<T> Unpin for Vec<T> {}
             unsafe { Pin::get_unchecked_mut(self) }.push(item);
             Ok(())
@@ -196,15 +190,14 @@ mod if_alloc {
         }
     }
 
-    impl<T> Sink for ::alloc::collections::VecDeque<T> {
-        type SinkItem = T;
+    impl<T> Sink<T> for ::alloc::collections::VecDeque<T> {
         type SinkError = VecSinkError;
 
         fn poll_ready(self: Pin<&mut Self>, _: &Waker) -> Poll<Result<(), Self::SinkError>> {
             Poll::Ready(Ok(()))
         }
 
-        fn start_send(self: Pin<&mut Self>, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+        fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::SinkError> {
             // TODO: impl<T> Unpin for Vec<T> {}
             unsafe { Pin::get_unchecked_mut(self) }.push_back(item);
             Ok(())
@@ -219,15 +212,14 @@ mod if_alloc {
         }
     }
 
-    impl<S: ?Sized + Sink + Unpin> Sink for ::alloc::boxed::Box<S> {
-        type SinkItem = S::SinkItem;
+    impl<S: ?Sized + Sink<Item> + Unpin, Item> Sink<Item> for ::alloc::boxed::Box<S> {
         type SinkError = S::SinkError;
 
         fn poll_ready(mut self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {
             Pin::new(&mut **self).poll_ready(waker)
         }
 
-        fn start_send(mut self: Pin<&mut Self>, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+        fn start_send(mut self: Pin<&mut Self>, item: Item) -> Result<(), Self::SinkError> {
             Pin::new(&mut **self).start_send(item)
         }
 
@@ -247,13 +239,11 @@ pub use self::if_alloc::*;
 #[cfg(feature = "either")]
 use either::Either;
 #[cfg(feature = "either")]
-impl<A, B> Sink for Either<A, B>
-    where A: Sink,
-          B: Sink<SinkItem=<A as Sink>::SinkItem,
-                  SinkError=<A as Sink>::SinkError>
+impl<A, B, Item> Sink<Item> for Either<A, B>
+    where A: Sink<Item>,
+          B: Sink<Item, SinkError=A::SinkError>,
 {
-    type SinkItem = <A as Sink>::SinkItem;
-    type SinkError = <A as Sink>::SinkError;
+    type SinkError = A::SinkError;
 
     fn poll_ready(self: Pin<&mut Self>, waker: &Waker) -> Poll<Result<(), Self::SinkError>> {
         unsafe {
@@ -264,7 +254,7 @@ impl<A, B> Sink for Either<A, B>
         }
     }
 
-    fn start_send(self: Pin<&mut Self>, item: Self::SinkItem) -> Result<(), Self::SinkError> {
+    fn start_send(self: Pin<&mut Self>, item: Item) -> Result<(), Self::SinkError> {
         unsafe {
             match Pin::get_unchecked_mut(self) {
                 Either::Left(x) => Pin::new_unchecked(x).start_send(item),

futures-util/src/compat/compat01as03.rs

@@ -210,16 +210,15 @@ where
     }
 }
 
-impl<S, SinkItem> Sink03 for Compat01As03Sink<S, SinkItem>
+impl<S, SinkItem> Sink03<SinkItem> for Compat01As03Sink<S, SinkItem>
 where
     S: Sink01<SinkItem = SinkItem>,
 {
-    type SinkItem = SinkItem;
     type SinkError = S::SinkError;
 
     fn start_send(
         mut self: Pin<&mut Self>,
-        item: Self::SinkItem,
+        item: SinkItem,
     ) -> Result<(), Self::SinkError> {
         debug_assert!(self.buffer.is_none());
         self.buffer = Some(item);

futures-util/src/compat/compat03as01.rs

@@ -15,6 +15,7 @@ use futures_core::{
 use futures_sink::Sink as Sink03;
 use crate::task::{ArcWake as ArcWake03, WakerRef};
 use std::{
+    marker::PhantomData,
     mem,
     pin::Pin,
     sync::Arc,
@@ -32,6 +33,14 @@ pub struct Compat<T> {
     pub(crate) inner: T,
 }
 
+/// Converts a futures 0.3 Sink object to a futures 0.1-compatible version
+#[derive(Debug)]
+#[must_use = "sinks do nothing unless polled"]
+pub struct CompatSink<T, Item> {
+    inner: T,
+    _phantom: PhantomData<fn(Item)>,
+}
+
 impl<T> Compat<T> {
     /// Returns the inner item.
     pub fn into_inner(self) -> T {
@@ -48,6 +57,21 @@ impl<T> Compat<T> {
     }
 }
 
+impl<T, Item> CompatSink<T, Item> {
+    /// Returns the inner item.
+    pub fn into_inner(self) -> T {
+        self.inner
+    }
+
+    /// Creates a new [`CompatSink`].
+    pub fn new(inner: T) -> Self {
+        CompatSink {
+            inner,
+            _phantom: PhantomData,
+        }
+    }
+}
+
 fn poll_03_to_01<T, E>(x: task03::Poll<Result<T, E>>)
     -> Result<Async01<T>, E>
 {
@@ -87,18 +111,18 @@ where
     }
 }
 
-impl<T> Sink01 for Compat<T>
+impl<T, Item> Sink01 for CompatSink<T, Item>
 where
-    T: Sink03 + Unpin,
+    T: Sink03<Item> + Unpin,
 {
-    type SinkItem = T::SinkItem;
+    type SinkItem = Item;
     type SinkError = T::SinkError;
 
     fn start_send(
         &mut self,
         item: Self::SinkItem,
     ) -> StartSend01<Self::SinkItem, Self::SinkError> {
-        with_context(self, |mut inner, waker| {
+        with_sink_context(self, |mut inner, waker| {
             match inner.as_mut().poll_ready(waker) {
                 task03::Poll::Ready(Ok(())) => {
                     inner.start_send(item).map(|()| AsyncSink01::Ready)
@@ -110,11 +134,11 @@ where
     }
 
     fn poll_complete(&mut self) -> Poll01<(), Self::SinkError> {
-        with_context(self, |inner, waker| poll_03_to_01(inner.poll_flush(waker)))
+        with_sink_context(self, |inner, waker| poll_03_to_01(inner.poll_flush(waker)))
     }
 
     fn close(&mut self) -> Poll01<(), Self::SinkError> {
-        with_context(self, |inner, waker| poll_03_to_01(inner.poll_close(waker)))
+        with_sink_context(self, |inner, waker| poll_03_to_01(inner.poll_close(waker)))
     }
 }
 
@@ -172,6 +196,16 @@ where
     f(Pin::new(&mut compat.inner), &waker)
 }
 
+fn with_sink_context<T, Item, R, F>(compat: &mut CompatSink<T, Item>, f: F) -> R
+where
+    T: Unpin,
+    F: FnOnce(Pin<&mut T>, &task03::Waker) -> R,
+{
+    let current = Current::new();
+    let waker = current.as_waker();
+    f(Pin::new(&mut compat.inner), &waker)
+}
+
 #[cfg(feature = "io-compat")]
 mod io {
     use super::*;

futures-util/src/compat/mod.rs

@@ -12,4 +12,4 @@ pub use self::compat01as03::{Compat01As03, Compat01As03Sink, Future01CompatExt,
 pub use self::compat01as03::{AsyncRead01CompatExt, AsyncWrite01CompatExt};
 
 mod compat03as01;
-pub use self::compat03as01::Compat;
+pub use self::compat03as01::{Compat, CompatSink};

futures-util/src/lib.rs

@@ -45,7 +45,7 @@ pub mod core_reexport {
 }
 
 macro_rules! delegate_sink {
-    ($field:ident) => {
+    ($field:ident, $item:ty) => {
         fn poll_ready(
             self: Pin<&mut Self>,
             waker: &$crate::core_reexport::task::Waker,
@@ -55,7 +55,7 @@ macro_rules! delegate_sink {
 
         fn start_send(
             self: Pin<&mut Self>,
-            item: Self::SinkItem
+            item: $item,
         ) -> Result<(), Self::SinkError> {
             self.$field().start_send(item)
         }

futures-util/src/sink/buffer.rs

@@ -9,23 +9,23 @@ use alloc::collections::VecDeque;
 /// number of values when the underlying sink is unable to accept them.
 #[derive(Debug)]
 #[must_use = "sinks do nothing unless polled"]
-pub struct Buffer<Si: Sink> {
+pub struct Buffer<Si: Sink<Item>, Item> {
     sink: Si,
-    buf: VecDeque<Si::SinkItem>,
+    buf: VecDeque<Item>,
 
     // Track capacity separately from the `VecDeque`, which may be rounded up
     capacity: usize,
 }
 
-impl<Si: Sink + Unpin> Unpin for Buffer<Si> {}
+impl<Si: Sink<Item> + Unpin, Item> Unpin for Buffer<Si, Item> {}
 
-impl<Si: Sink> Buffer<Si> {
+impl<Si: Sink<Item>, Item> Buffer<Si, Item> {
     unsafe_pinned!(sink: Si);
-    unsafe_unpinned!(buf: VecDeque<Si::SinkItem>);
+    unsafe_unpinned!(buf: VecDeque<Item>);
     unsafe_unpinned!(capacity: usize);
 
 
-    pub(super) fn new(sink: Si, capacity: usize) -> Buffer<Si> {
+    pub(super) fn new(sink: Si, capacity: usize) -> Self {
         Buffer {
             sink,
             buf: VecDeque::with_capacity(capacity),
@@ -56,16 +56,15 @@ impl<Si: Sink> Buffer<Si> {
 }
 
 // Forwarding impl of Stream from the underlying sink
-impl<S> Stream for Buffer<S> where S: Sink + Stream {
+impl<S, Item> Stream for Buffer<S, Item> where S: Sink<Item> + Stream {
     type Item = S::Item;
 
     fn poll_next(self: Pin<&mut Self>, waker: &Waker) -> Poll<Option<S::Item>> {
         self.sink().poll_next(waker)
     }
 }
 
-impl<Si: Sink> Sink for Buffer<Si> {
-    type SinkItem = Si::SinkItem;
+impl<Si: Sink<Item>, Item> Sink<Item> for Buffer<Si, Item> {
     type SinkError = Si::SinkError;
 
     fn poll_ready(
@@ -89,7 +88,7 @@ impl<Si: Sink> Sink for Buffer<Si> {
 
     fn start_send(
         mut self: Pin<&mut Self>,
-        item: Self::SinkItem,
+        item: Item,
     ) -> Result<(), Self::SinkError> {
         if self.capacity == 0 {
             self.as_mut().sink().start_send(item)