async_std/stream/stream/
timeout.rs

1use std::error::Error;
2use std::fmt;
3use std::pin::Pin;
4use std::time::Duration;
5use std::future::Future;
6
7use futures_timer::Delay;
8use pin_project_lite::pin_project;
9
10use crate::stream::Stream;
11use crate::task::{Context, Poll};
12
13pin_project! {
14    /// A stream with timeout time set
15    #[derive(Debug)]
16    pub struct Timeout<S: Stream> {
17        #[pin]
18        stream: S,
19        #[pin]
20        delay: Delay,
21    }
22}
23
24impl<S: Stream> Timeout<S> {
25    pub(crate) fn new(stream: S, dur: Duration) -> Self {
26        let delay = Delay::new(dur);
27
28        Self { stream, delay }
29    }
30}
31
32impl<S: Stream> Stream for Timeout<S> {
33    type Item = Result<S::Item, TimeoutError>;
34
35    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
36        let this = self.project();
37
38        match this.stream.poll_next(cx) {
39            Poll::Ready(Some(v)) => Poll::Ready(Some(Ok(v))),
40            Poll::Ready(None) => Poll::Ready(None),
41            Poll::Pending => match this.delay.poll(cx) {
42                Poll::Ready(_) => Poll::Ready(Some(Err(TimeoutError { _private: () }))),
43                Poll::Pending => Poll::Pending,
44            },
45        }
46    }
47}
48
49/// An error returned when a stream times out.
50#[cfg_attr(feature = "docs", doc(cfg(unstable)))]
51#[cfg(any(feature = "unstable", feature = "docs"))]
52#[derive(Clone, Copy, Debug, Eq, PartialEq)]
53pub struct TimeoutError {
54    _private: (),
55}
56
57impl Error for TimeoutError {}
58
59impl fmt::Display for TimeoutError {
60    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
61        "stream has timed out".fmt(f)
62    }
63}
async_std/stream/stream/timeout.rs

async_std/stream/stream/
timeout.rs
