lib.rs

pub struct BufferInterval<T, S, I>
where
    T: Stream<Item = I>,
    S: Store<Item = I>,
    I: Clone,
{
    interval: Interval,
    rx: T,
    store: S,
}

impl<T, S, I> Stream for BufferInterval<T, S, I>
where
    T: Stream<Item = I>,
    S: Store<Item = I>,
    I: Clone,
{
    type Item = S;
    type Error = Error;

    fn poll(&mut self) -> Result<Async<Option<Self::Item>>> {
        loop {
            match self.interval.poll()? {
                Async::Ready(Some(_inst)) => {
                    let store = self.store.trancate();
                    return Ok(Async::Ready(Some(store)));
                }
                Async::Ready(None) => {
                    unreachable!();
                }
                Async::NotReady => {
                    trace!("not until to trancate");
                }
            };

            match self.rx.poll()? {
                Async::Ready(Some(val)) => self.store.store(val),
                Async::NotReady => return Ok(Async::NotReady),
                Async::Ready(None) => return Ok(Async::Ready(None)),
            };
        }
    }
}