async_std/stream/stream/
timeout.rs

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

async_std/stream/stream/
timeout.rs
