use crate::stream::Stream;
use crate::time::{Delay, Duration, Instant};
use std::future::Future;
use std::marker::Unpin;
use std::pin::Pin;
use std::task::{self, Poll};
use pin_project_lite::pin_project;
pub fn throttle<T>(duration: Duration, stream: T) -> Throttle<T>
where
T: Stream,
{
let delay = if duration == Duration::from_millis(0) {
None
} else {
Some(Delay::new_timeout(Instant::now() + duration, duration))
};
Throttle {
delay,
duration,
has_delayed: true,
stream,
}
}
pin_project! {
#[derive(Debug)]
#[must_use = "streams do nothing unless polled"]
pub struct Throttle<T> {
delay: Option<Delay>,
duration: Duration,
has_delayed: bool,
#[pin]
stream: T,
}
}
impl<T: Unpin> Throttle<T> {
pub fn get_ref(&self) -> &T {
&self.stream
}
pub fn get_mut(&mut self) -> &mut T {
&mut self.stream
}
pub fn into_inner(self) -> T {
self.stream
}
}
impl<T: Stream> Stream for Throttle<T> {
type Item = T::Item;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Option<Self::Item>> {
if !self.has_delayed && self.delay.is_some() {
ready!(Pin::new(self.as_mut().project().delay.as_mut().unwrap()).poll(cx));
*self.as_mut().project().has_delayed = true;
}
let value = ready!(self.as_mut().project().stream.poll_next(cx));
if value.is_some() {
let dur = self.duration;
if let Some(ref mut delay) = self.as_mut().project().delay {
delay.reset(Instant::now() + dur);
}
*self.as_mut().project().has_delayed = false;
}
Poll::Ready(value)
}
}