pollable_map/stream/
timeout_map.rs

1use crate::common::Timed;
2use crate::error::TimedError;
3use crate::stream::StreamMap;
4use core::ops::{Deref, DerefMut};
5use core::pin::Pin;
6use core::task::{Context, Poll};
7use core::time::Duration;
8use futures::stream::FusedStream;
9use futures::{Stream, StreamExt};
10
11pub struct TimeoutStreamMap<K, S> {
12    duration: Duration,
13    map: StreamMap<K, Timed<S>>,
14}
15
16impl<K, S> Deref for TimeoutStreamMap<K, S> {
17    type Target = StreamMap<K, Timed<S>>;
18    fn deref(&self) -> &Self::Target {
19        &self.map
20    }
21}
22
23impl<K, S> DerefMut for TimeoutStreamMap<K, S> {
24    fn deref_mut(&mut self) -> &mut Self::Target {
25        &mut self.map
26    }
27}
28
29impl<K, S> TimeoutStreamMap<K, S>
30where
31    K: Clone + PartialEq,
32    S: Stream,
33{
34    /// Create an empty [`TimeoutStreamMap`]
35    pub fn new(duration: Duration) -> Self {
36        Self {
37            duration,
38            map: StreamMap::new(),
39        }
40    }
41
42    /// Insert a stream into the map with a unique key.
43    /// The function will return true if the map does not have the key present,
44    /// otherwise it will return false
45    pub fn insert(&mut self, key: K, stream: S) -> bool {
46        self.map.insert(key, Timed::new(stream, self.duration))
47    }
48}
49
50impl<K, S> Stream for TimeoutStreamMap<K, S>
51where
52    K: Clone,
53    S: Stream,
54{
55    type Item = (K, Result<S::Item, TimedError>);
56    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
57        self.map.poll_next_unpin(cx)
58    }
59
60    fn size_hint(&self) -> (usize, Option<usize>) {
61        self.map.size_hint()
62    }
63}
64
65impl<K, S> FusedStream for TimeoutStreamMap<K, S>
66where
67    K: Clone,
68    S: Stream,
69{
70    fn is_terminated(&self) -> bool {
71        self.map.is_terminated()
72    }
73}
74
75#[cfg(test)]
76mod test {
77    use crate::{error::TimedError, stream::timeout_map::TimeoutStreamMap};
78    use futures::StreamExt;
79    use std::time::Duration;
80
81    #[test]
82    fn timeout_map() {
83        let mut list = TimeoutStreamMap::new(Duration::from_millis(100));
84        assert!(list.insert(0, futures::stream::pending::<()>()));
85
86        futures::executor::block_on(async move {
87            let result = list.next().await;
88            let Some((0, Err(e))) = result else {
89                unreachable!("result is err");
90            };
91
92            assert_eq!(e, TimedError);
93        });
94    }
95
96    #[test]
97    fn valid_stream() {
98        let mut list = TimeoutStreamMap::new(Duration::from_secs(10));
99        assert!(list.insert(1, futures::stream::once(async { 0 }).boxed()));
100
101        futures::executor::block_on(async move {
102            let result = list.next().await;
103            let Some((1, Ok(val))) = result else {
104                unreachable!("result is err");
105            };
106
107            assert_eq!(val, 0);
108        });
109    }
110}
pollable_map/stream/timeout_map.rs

pollable_map/stream/
timeout_map.rs
