1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

use std::marker::Unpin;

use futures::Stream;

#[allow(unreachable_pub)]
pub use self::limiter::{IntoLimiter, Limiter};

#[cfg(feature = "leaky-bucket")]
mod rate_limiter;

mod limiter;

impl<T: ?Sized> LimiterExt for T where T: Stream {}

pub trait LimiterExt: Stream {
    #[inline]
    fn limiter<L>(self, l: L) -> IntoLimiter<Self, L>
        where
            Self: Sized + Stream + Unpin,
            L: Limiter + Unpin,
    {
        assert_stream::<Self::Item, _>(IntoLimiter::new(self, l))
    }

    #[cfg(feature = "leaky-bucket")]
    #[inline]
    fn leaky_bucket_limiter(
        self,
        rate_limiter: leaky_bucket::RateLimiter,
    ) -> IntoLimiter<Self, rate_limiter::LeakyBucketRateLimiter>
        where
            Self: Sized + Stream + Unpin,
    {
        let l = rate_limiter::LeakyBucketRateLimiter::new(rate_limiter);
        assert_stream::<Self::Item, _>(IntoLimiter::new(self, l))
    }

    #[cfg(feature = "governor")]
    #[inline]
    fn governor_limiter<D, C, MW>(
        self,
        rate_limiter: &governor::RateLimiter<governor::state::NotKeyed, D, C, MW>,
    ) -> governor::RatelimitedStream<Self, D, C, MW>
        where
            D: governor::state::DirectStateStore,
            C: governor::clock::Clock + governor::clock::ReasonablyRealtime,
            MW: governor::middleware::RateLimitingMiddleware<
                C::Instant,
                NegativeOutcome=governor::NotUntil<<C as governor::clock::Clock>::Instant>,
            >,
            Self: Sized + Stream + Unpin,
            Self::Item: Unpin,
    {
        use governor::state::StreamRateLimitExt;
        assert_stream::<Self::Item, _>(self.ratelimit_stream(rate_limiter))
    }
}

// Just a helper function to ensure the streams we're returning all have the
// right implementations.
#[inline]
pub(crate) fn assert_stream<T, S>(stream: S) -> S
    where
        S: Stream<Item=T>,
{
    stream
}