use crate::BufStream; use bytes::Buf; use futures::{Async, Poll, Stream}; /// Converts a `Stream` of `Buf` types into a `BufStream`. /// /// While `Stream` and `BufStream` are very similar, they are not identical. The /// `stream` function returns a `BufStream` that is backed by the provided /// `Stream` type. pub fn stream(stream: T) -> FromStream where T: Stream, T::Item: Buf, { FromStream { stream } } /// `BufStream` returned by the [`stream`] function. #[derive(Debug)] pub struct FromStream { stream: T, } impl BufStream for FromStream where T: Stream, T::Item: Buf, { type Item = T::Item; type Error = T::Error; fn poll_buf(&mut self) -> Poll, Self::Error> { self.stream.poll() } } /// Converts a `BufStream` into a `Stream`. #[derive(Debug)] pub struct IntoStream { buf: T, } impl IntoStream { /// Create a new `Stream` from the provided `BufStream`. pub fn new(buf: T) -> Self { IntoStream { buf } } /// Get a reference to the inner `BufStream`. pub fn get_ref(&self) -> &T { &self.buf } /// Get a mutable reference to the inner `BufStream` pub fn get_mut(&mut self) -> &mut T { &mut self.buf } /// Get the inner `BufStream`. pub fn into_inner(self) -> T { self.buf } } impl Stream for IntoStream { type Item = T::Item; type Error = T::Error; fn poll(&mut self) -> Poll, Self::Error> { match self.buf.poll_buf()? { Async::Ready(Some(buf)) => Ok(Async::Ready(Some(buf))), Async::Ready(None) => Ok(Async::Ready(None)), Async::NotReady => Ok(Async::NotReady), } } }