diff --git a/tokio-io/src/async_buf_read.rs b/tokio-io/src/async_buf_read.rs new file mode 100644 index 000000000..9c2c55db8 --- /dev/null +++ b/tokio-io/src/async_buf_read.rs @@ -0,0 +1,94 @@ +use crate::AsyncRead; +use std::io; +use std::ops::DerefMut; +use std::pin::Pin; +use std::task::{Context, Poll}; + +/// Read bytes asynchronously. +/// +/// This trait inherits from `std::io::BufRead` and indicates that an I/O object is +/// **non-blocking**. All non-blocking I/O objects must return an error when +/// bytes are unavailable instead of blocking the current thread. +pub trait AsyncBufRead: AsyncRead { + /// Attempt to return the contents of the internal buffer, filling it with more data + /// from the inner reader if it is empty. + /// + /// On success, returns `Poll::Ready(Ok(buf))`. + /// + /// If no data is available for reading, the method returns + /// `Poll::Pending` and arranges for the current task (via + /// `cx.waker().wake_by_ref()`) to receive a notification when the object becomes + /// readable or is closed. + /// + /// This function is a lower-level call. It needs to be paired with the + /// [`consume`] method to function properly. When calling this + /// method, none of the contents will be "read" in the sense that later + /// calling [`poll_read`] may return the same contents. As such, [`consume`] must + /// be called with the number of bytes that are consumed from this buffer to + /// ensure that the bytes are never returned twice. + /// + /// An empty buffer returned indicates that the stream has reached EOF. + /// + /// [`poll_read`]: AsyncRead::poll_read + /// [`consume`]: AsyncBufRead::consume + fn poll_fill_buf<'a>( + self: Pin<&'a mut Self>, + cx: &mut Context<'_>, + ) -> Poll>; + + /// Tells this buffer that `amt` bytes have been consumed from the buffer, + /// so they should no longer be returned in calls to [`poll_read`]. + /// + /// This function is a lower-level call. It needs to be paired with the + /// [`poll_fill_buf`] method to function properly. This function does + /// not perform any I/O, it simply informs this object that some amount of + /// its buffer, returned from [`poll_fill_buf`], has been consumed and should + /// no longer be returned. As such, this function may do odd things if + /// [`poll_fill_buf`] isn't called before calling it. + /// + /// The `amt` must be `<=` the number of bytes in the buffer returned by + /// [`poll_fill_buf`]. + /// + /// [`poll_read`]: AsyncRead::poll_read + /// [`poll_fill_buf`]: AsyncBufRead::poll_fill_buf + fn consume(self: Pin<&mut Self>, amt: usize); +} + +macro_rules! deref_async_buf_read { + () => { + fn poll_fill_buf<'a>(self: Pin<&'a mut Self>, cx: &mut Context<'_>) + -> Poll> + { + Pin::new(&mut **self.get_mut()).poll_fill_buf(cx) + } + + fn consume(mut self: Pin<&mut Self>, amt: usize) { + Pin::new(&mut **self).consume(amt) + } + } +} + +impl AsyncBufRead for Box { + deref_async_buf_read!(); +} + +impl AsyncBufRead for &mut T { + deref_async_buf_read!(); +} + +impl

AsyncBufRead for Pin

+where + P: DerefMut + Unpin, + P::Target: AsyncBufRead, +{ + fn poll_fill_buf<'a>( + self: Pin<&'a mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + self.get_mut().as_mut().poll_fill_buf(cx) + } + + fn consume(self: Pin<&mut Self>, amt: usize) { + self.get_mut().as_mut().consume(amt) + } +} diff --git a/tokio-io/src/lib.rs b/tokio-io/src/lib.rs index 165cc1ca6..ad8b833d9 100644 --- a/tokio-io/src/lib.rs +++ b/tokio-io/src/lib.rs @@ -12,8 +12,10 @@ //! [found online]: https://tokio.rs/docs/ //! [low level details]: https://tokio.rs/docs/going-deeper-tokio/core-low-level/ +mod async_buf_read; mod async_read; mod async_write; +pub use self::async_buf_read::AsyncBufRead; pub use self::async_read::AsyncRead; pub use self::async_write::AsyncWrite; diff --git a/tokio/src/io/async_buf_read_ext.rs b/tokio/src/io/async_buf_read_ext.rs new file mode 100644 index 000000000..7f0db43f8 --- /dev/null +++ b/tokio/src/io/async_buf_read_ext.rs @@ -0,0 +1,6 @@ +use tokio_io::AsyncBufRead; + +/// An extension trait which adds utility methods to `AsyncBufRead` types. +pub trait AsyncBufReadExt: AsyncBufRead {} + +impl AsyncBufReadExt for R {} diff --git a/tokio/src/io/mod.rs b/tokio/src/io/mod.rs index e66dc1812..10d7b4b28 100644 --- a/tokio/src/io/mod.rs +++ b/tokio/src/io/mod.rs @@ -36,6 +36,7 @@ //! [`ErrorKind`]: enum.ErrorKind.html //! [`Result`]: type.Result.html +mod async_buf_read_ext; mod async_read_ext; mod async_write_ext; mod copy; @@ -45,13 +46,14 @@ mod read_to_end; mod write; mod write_all; +pub use self::async_buf_read_ext::AsyncBufReadExt; pub use self::async_read_ext::AsyncReadExt; pub use self::async_write_ext::AsyncWriteExt; // standard input, output, and error #[cfg(feature = "fs")] pub use tokio_fs::{stderr, stdin, stdout, Stderr, Stdin, Stdout}; -pub use tokio_io::{AsyncRead, AsyncWrite}; +pub use tokio_io::{AsyncBufRead, AsyncRead, AsyncWrite}; // Re-export io::Error so that users don't have to deal // with conflicts when `use`ing `tokio::io` and `std::io`.