Repository: alexcrichton/futures-rs
Pull request #701 (2018-01-12 18:18:28)

Forward Sink through `Inspect` and add Stream::inspect_err
See also issue #700 



Patch:
diff --git a/src/stream/inspect.rs b/src/stream/inspect.rs
index eed0855e..fc8f7f4e 100644
--- a/src/stream/inspect.rs
+++ b/src/stream/inspect.rs
@@ -45,6 +45,26 @@ impl<S: Stream, F> Inspect<S, F> {
     }
 }
 
+// Forwarding impl of Sink from the underlying stream
+impl<S, F> ::sink::Sink for Inspect<S, F>
+    where S: ::sink::Sink + Stream
+{
+    type SinkItem = S::SinkItem;
+    type SinkError = S::SinkError;
+
+    fn start_send(&mut self, item: S::SinkItem) -> ::StartSend<S::SinkItem, S::SinkError> {
+        self.stream.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), S::SinkError> {
+        self.stream.poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), S::SinkError> {
+        self.stream.close()
+    }
+}
+
 impl<S, F> Stream for Inspect<S, F>
     where S: Stream,
           F: FnMut(&S::Item),
diff --git a/src/stream/inspect_err.rs b/src/stream/inspect_err.rs
new file mode 100644
index 00000000..5c56a217
--- /dev/null
+++ b/src/stream/inspect_err.rs
@@ -0,0 +1,81 @@
+use {Stream, Poll};
+
+/// Do something with the error of a stream, passing it on.
+///
+/// This is created by the `Stream::inspect_err` method.
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct InspectErr<S, F> where S: Stream {
+    stream: S,
+    inspect: F,
+}
+
+pub fn new<S, F>(stream: S, f: F) -> InspectErr<S, F>
+    where S: Stream,
+          F: FnMut(&S::Error) -> (),
+{
+    InspectErr {
+        stream: stream,
+        inspect: f,
+    }
+}
+
+impl<S: Stream, F> InspectErr<S, F> {
+    /// Acquires a reference to the underlying stream that this combinator is
+    /// pulling from.
+    pub fn get_ref(&self) -> &S {
+        &self.stream
+    }
+
+    /// Acquires a mutable reference to the underlying stream that this
+    /// combinator is pulling from.
+    ///
+    /// Note that care must be taken to avoid tampering with the state of the
+    /// stream which may otherwise confuse this combinator.
+    pub fn get_mut(&mut self) -> &mut S {
+        &mut self.stream
+    }
+
+    /// Consumes this combinator, returning the underlying stream.
+    ///
+    /// Note that this may discard intermediate state of this combinator, so
+    /// care should be taken to avoid losing resources when this is called.
+    pub fn into_inner(self) -> S {
+        self.stream
+    }
+}
+
+// Forwarding impl of Sink from the underlying stream
+impl<S, F> ::sink::Sink for InspectErr<S, F>
+    where S: ::sink::Sink + Stream
+{
+    type SinkItem = S::SinkItem;
+    type SinkError = S::SinkError;
+
+    fn start_send(&mut self, item: S::SinkItem) -> ::StartSend<S::SinkItem, S::SinkError> {
+        self.stream.start_send(item)
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), S::SinkError> {
+        self.stream.poll_complete()
+    }
+
+    fn close(&mut self) -> Poll<(), S::SinkError> {
+        self.stream.close()
+    }
+}
+
+impl<S, F> Stream for InspectErr<S, F>
+    where S: Stream,
+          F: FnMut(&S::Error),
+{
+    type Item = S::Item;
+    type Error = S::Error;
+
+    fn poll(&mut self) -> Poll<Option<S::Item>, S::Error> {
+        self.stream.poll().map_err(|e| {
+            (self.inspect)(&e);
+            e
+        })
+    }
+}
diff --git a/src/stream/mod.rs b/src/stream/mod.rs
index 8ae0a538..01ae6d69 100644
--- a/src/stream/mod.rs
+++ b/src/stream/mod.rs
@@ -44,6 +44,7 @@ mod from_err;
 mod fuse;
 mod future;
 mod inspect;
+mod inspect_err;
 mod map;
 mod map_err;
 mod merge;
@@ -75,6 +76,7 @@ pub use self::from_err::FromErr;
 pub use self::fuse::Fuse;
 pub use self::future::StreamFuture;
 pub use self::inspect::Inspect;
+pub use self::inspect_err::InspectErr;
 pub use self::map::Map;
 pub use self::map_err::MapErr;
 #[allow(deprecated)]
@@ -1088,13 +1090,25 @@ pub trait Stream {
     ///
     /// This is similar to the `Iterator::inspect` method in the standard
     /// library where it allows easily inspecting each value as it passes
-    /// through the stream, for example to debug what's going.
+    /// through the stream, for example to debug what's going on.
     fn inspect<F>(self, f: F) -> Inspect<Self, F>
         where F: FnMut(&Self::Item),
               Self: Sized,
     {
         inspect::new(self, f)
     }
+
+    /// Do something with the error of this stream, afterwards passing it on.
+    ///
+    /// This is similar to the `Stream::inspect` method where it allows
+    /// easily inspecting the error as it passes through the stream, for
+    /// example to debug what's going on.
+    fn inspect_err<F>(self, f: F) -> InspectErr<Self, F>
+        where F: FnMut(&Self::Error),
+              Self: Sized,
+    {
+        inspect_err::new(self, f)
+    }
 }
 
 impl<'a, S: ?Sized + Stream> Stream for &'a mut S {


Test patch:
diff --git a/tests/stream.rs b/tests/stream.rs
index b0e8a361..eb756035 100644
--- a/tests/stream.rs
+++ b/tests/stream.rs
@@ -398,3 +398,17 @@ fn stream_poll_fn() {
 
     assert_eq!(read_stream.wait().count(), 5);
 }
+
+#[test]
+fn inspect() {
+    let mut seen = vec![];
+    assert_done(|| list().inspect(|&a| seen.push(a)).collect(), Ok(vec![1, 2, 3]));
+    assert_eq!(seen, [1, 2, 3]);
+}
+
+#[test]
+fn inspect_err() {
+    let mut seen = vec![];
+    assert_done(|| err_list().inspect_err(|&a| seen.push(a)).collect(), Err(3));
+    assert_eq!(seen, [3]);
+}