pollable_map/futures/
timeout_map.rs

1use crate::common::Timed;
2use crate::futures::FutureMap;
3use futures::{Stream, StreamExt};
4use std::future::Future;
5use std::ops::{Deref, DerefMut};
6use std::pin::Pin;
7use std::task::{Context, Poll};
8use std::time::Duration;
9
10pub struct TimeoutFutureMap<K, F> {
11    duration: Duration,
12    map: FutureMap<K, Timed<F>>,
13}
14
15impl<K, F> Deref for TimeoutFutureMap<K, F> {
16    type Target = FutureMap<K, Timed<F>>;
17    fn deref(&self) -> &Self::Target {
18        &self.map
19    }
20}
21
22impl<K, F> DerefMut for TimeoutFutureMap<K, F> {
23    fn deref_mut(&mut self) -> &mut Self::Target {
24        &mut self.map
25    }
26}
27
28impl<K, F> TimeoutFutureMap<K, F>
29where
30    K: Clone + PartialEq + Send + Unpin + 'static,
31    F: Future + Send + Unpin + 'static,
32{
33    /// Create an empty [`TimeoutFutureMap`]
34    pub fn new(duration: Duration) -> Self {
35        Self {
36            duration,
37            map: FutureMap::new(),
38        }
39    }
40
41    /// Insert a future into the map with a unique key.
42    /// The function will return true if the map does not have the key present,
43    /// otherwise it will return false
44    pub fn insert(&mut self, key: K, future: F) -> bool {
45        self.map.insert(key, Timed::new(future, self.duration))
46    }
47}
48
49impl<K, F> Stream for TimeoutFutureMap<K, F>
50where
51    K: Clone + PartialEq + Send + Unpin + 'static,
52    F: Future + Send + Unpin + 'static,
53{
54    type Item = (K, std::io::Result<F::Output>);
55    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
56        self.map.poll_next_unpin(cx)
57    }
58
59    fn size_hint(&self) -> (usize, Option<usize>) {
60        self.map.size_hint()
61    }
62}
pollable_map/futures/timeout_map.rs

pollable_map/futures/
timeout_map.rs
