diff --git a/tokio-buf/src/util/mod.rs b/tokio-buf/src/util/mod.rs index e080b8a78..14c631831 100644 --- a/tokio-buf/src/util/mod.rs +++ b/tokio-buf/src/util/mod.rs @@ -12,7 +12,7 @@ pub use self::collect::Collect; pub use self::from::FromBufStream; pub use self::iter::iter; pub use self::limit::Limit; -pub use self::stream::stream; +pub use self::stream::{stream, IntoStream}; pub mod error { //! Error types @@ -74,4 +74,14 @@ pub trait BufStreamExt: BufStream { { Limit::new(self, amount) } + + /// Creates a `Stream` from a `BufStream`. + /// + /// This produces a `Stream` of `BufStream::Items`. + fn into_stream(self) -> IntoStream + where + Self: Sized, + { + IntoStream::new(self) + } } diff --git a/tokio-buf/src/util/stream.rs b/tokio-buf/src/util/stream.rs index a735fb486..8716f23b8 100644 --- a/tokio-buf/src/util/stream.rs +++ b/tokio-buf/src/util/stream.rs @@ -1,10 +1,10 @@ use bytes::Buf; -use futures::{Poll, Stream}; +use futures::{Async, Poll, Stream}; use BufStream; /// Converts a `Stream` of `Buf` types into a `BufStream`. /// -/// While `Stream` and `BufSream` are very similar, they are not identical. The +/// 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 @@ -33,3 +33,44 @@ where 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), + } + } +}