stream_ext/
limiter.rs

1use std::fmt;
2use std::marker::Unpin;
3use std::pin::Pin;
4use std::task::{Context, Poll};
5
6use futures::ready;
7use futures::stream::{FusedStream, Stream};
8
9pub trait Limiter {
10    fn acquire(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<()>>;
11}
12
13pub struct IntoLimiter<St, L> {
14    stream: St,
15    limiter: L,
16}
17
18impl<St, L> fmt::Debug for IntoLimiter<St, L>
19    where
20        St: fmt::Debug,
21        L: fmt::Debug,
22{
23    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
24        f.debug_struct("IntoLimiter")
25            .field("stream", &self.stream)
26            .field("limiter", &self.limiter)
27            .finish()
28    }
29}
30
31impl<St, L> IntoLimiter<St, L> {
32    #[inline]
33    pub(crate) fn new(stream: St, limiter: L) -> Self {
34        Self { stream, limiter }
35    }
36}
37
38impl<St, L> FusedStream for IntoLimiter<St, L>
39    where
40        St: FusedStream + Unpin,
41        L: Limiter + Unpin,
42{
43    #[inline]
44    fn is_terminated(&self) -> bool {
45        self.stream.is_terminated()
46    }
47}
48
49impl<St, F> Stream for IntoLimiter<St, F>
50    where
51        St: Stream + Unpin,
52        F: Limiter + Unpin,
53{
54    type Item = St::Item;
55
56    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
57        match ready!(Pin::new(&mut self.limiter).acquire(cx)) {
58            Some(()) => Pin::new(&mut self.stream).poll_next(cx),
59            None => Poll::Ready(None),
60        }
61    }
62
63    #[inline]
64    fn size_hint(&self) -> (usize, Option<usize>) {
65        self.stream.size_hint()
66    }
67}
stream_ext/limiter.rs

stream_ext/
limiter.rs
