futures_util/stream/
inspect_err.rs

1use futures_core::{Stream, Poll};
2use futures_core::task;
3use futures_sink::{Sink};
4
5/// Do something with the error of a stream, passing it on.
6///
7/// This is created by the `Stream::inspect_err` method.
8#[derive(Debug)]
9#[must_use = "streams do nothing unless polled"]
10pub struct InspectErr<S, F> where S: Stream {
11    stream: S,
12    inspect: F,
13}
14
15pub fn new<S, F>(stream: S, f: F) -> InspectErr<S, F>
16    where S: Stream,
17          F: FnMut(&S::Error) -> (),
18{
19    InspectErr {
20        stream: stream,
21        inspect: f,
22    }
23}
24
25impl<S: Stream, F> InspectErr<S, F> {
26    /// Acquires a reference to the underlying stream that this combinator is
27    /// pulling from.
28    pub fn get_ref(&self) -> &S {
29        &self.stream
30    }
31
32    /// Acquires a mutable reference to the underlying stream that this
33    /// combinator is pulling from.
34    ///
35    /// Note that care must be taken to avoid tampering with the state of the
36    /// stream which may otherwise confuse this combinator.
37    pub fn get_mut(&mut self) -> &mut S {
38        &mut self.stream
39    }
40
41    /// Consumes this combinator, returning the underlying stream.
42    ///
43    /// Note that this may discard intermediate state of this combinator, so
44    /// care should be taken to avoid losing resources when this is called.
45    pub fn into_inner(self) -> S {
46        self.stream
47    }
48}
49
50// Forwarding impl of Sink from the underlying stream
51impl<S, F> Sink for InspectErr<S, F>
52    where S: Sink + Stream
53{
54    type SinkItem = S::SinkItem;
55    type SinkError = S::SinkError;
56
57    delegate_sink!(stream);
58}
59
60impl<S, F> Stream for InspectErr<S, F>
61    where S: Stream,
62          F: FnMut(&S::Error),
63{
64    type Item = S::Item;
65    type Error = S::Error;
66
67    fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<S::Item>, S::Error> {
68        self.stream.poll_next(cx).map_err(|e| {
69            (self.inspect)(&e);
70            e
71        })
72    }
73}
futures_util/stream/inspect_err.rs

futures_util/stream/
inspect_err.rs
