futures_util/sink/
buffer.rsuse std::collections::VecDeque;
use futures_core::{Poll, Async, Stream};
use futures_core::task;
use futures_sink::Sink;
#[derive(Debug)]
#[must_use = "sinks do nothing unless polled"]
pub struct Buffer<S: Sink> {
sink: S,
buf: VecDeque<S::SinkItem>,
cap: usize,
}
pub fn new<S: Sink>(sink: S, amt: usize) -> Buffer<S> {
Buffer {
sink: sink,
buf: VecDeque::with_capacity(amt),
cap: amt,
}
}
impl<S: Sink> Buffer<S> {
pub fn get_ref(&self) -> &S {
&self.sink
}
pub fn get_mut(&mut self) -> &mut S {
&mut self.sink
}
pub fn into_inner(self) -> S {
self.sink
}
fn try_empty_buffer(&mut self, cx: &mut task::Context) -> Poll<(), S::SinkError> {
try_ready!(self.sink.poll_ready(cx));
while let Some(item) = self.buf.pop_front() {
self.sink.start_send(item)?;
if self.buf.len() != 0 {
try_ready!(self.sink.poll_ready(cx));
}
}
Ok(Async::Ready(()))
}
}
impl<S> Stream for Buffer<S> where S: Sink + Stream {
type Item = S::Item;
type Error = S::Error;
fn poll_next(&mut self, cx: &mut task::Context) -> Poll<Option<S::Item>, S::Error> {
self.sink.poll_next(cx)
}
}
impl<S: Sink> Sink for Buffer<S> {
type SinkItem = S::SinkItem;
type SinkError = S::SinkError;
fn poll_ready(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
if self.cap == 0 {
return self.sink.poll_ready(cx);
}
self.try_empty_buffer(cx)?;
Ok(if self.buf.len() >= self.cap {
Async::Pending
} else {
Async::Ready(())
})
}
fn start_send(&mut self, item: Self::SinkItem) -> Result<(), Self::SinkError> {
if self.cap == 0 {
self.sink.start_send(item)
} else {
self.buf.push_back(item);
Ok(())
}
}
fn poll_flush(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
try_ready!(self.try_empty_buffer(cx));
debug_assert!(self.buf.is_empty());
self.sink.poll_flush(cx)
}
fn poll_close(&mut self, cx: &mut task::Context) -> Poll<(), Self::SinkError> {
try_ready!(self.try_empty_buffer(cx));
debug_assert!(self.buf.is_empty());
self.sink.poll_close(cx)
}
}