From 8aa520e2bd21363b76f84ddd623a34087dcbde85 Mon Sep 17 00:00:00 2001 From: Eliza Weisman Date: Mon, 7 Oct 2019 14:02:04 -0700 Subject: [PATCH] io: add missing utility functions (#1632) The standard library's `io` module has small utilities such as `repeat`, `empty`, and `sink`, which return `Read` and `Write` implementations. These can come in handy in some circiumstances. `tokio::io` has no equivalents that implement `AsyncRead`/`AsyncWrite`. This commit adds `repeat`, `empty`, and `sink` helpers to `tokio::io`. --- tokio-io/src/io/copy.rs | 42 ++++++++++++++++++++- tokio-io/src/io/empty.rs | 79 +++++++++++++++++++++++++++++++++++++++ tokio-io/src/io/mod.rs | 11 ++++++ tokio-io/src/io/repeat.rs | 67 +++++++++++++++++++++++++++++++++ tokio-io/src/io/sink.rs | 77 ++++++++++++++++++++++++++++++++++++++ tokio-io/src/lib.rs | 5 ++- tokio/src/io.rs | 4 +- 7 files changed, 281 insertions(+), 4 deletions(-) create mode 100644 tokio-io/src/io/empty.rs create mode 100644 tokio-io/src/io/repeat.rs create mode 100644 tokio-io/src/io/sink.rs diff --git a/tokio-io/src/io/copy.rs b/tokio-io/src/io/copy.rs index e3bc63e14..006b45f80 100644 --- a/tokio-io/src/io/copy.rs +++ b/tokio-io/src/io/copy.rs @@ -5,6 +5,13 @@ use std::io; use std::pin::Pin; use std::task::{Context, Poll}; +/// A future that asynchronously copies the entire contents of a reader into a +/// writer. +/// +/// This struct is generally created by calling [`copy`][copy]. Please +/// see the documentation of `copy()` for more details. +/// +/// [copy]: fn.copy.html #[derive(Debug)] #[must_use = "futures do nothing unless you `.await` or poll them"] pub struct Copy<'a, R: ?Sized, W: ?Sized> { @@ -17,7 +24,40 @@ pub struct Copy<'a, R: ?Sized, W: ?Sized> { buf: Box<[u8]>, } -pub(crate) fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> Copy<'a, R, W> +/// Asynchronously copies the entire contents of a reader into a writer. +/// +/// This function returns a future that will continuously read data from +/// `reader` and then write it into `writer` in a streaming fashion until +/// `reader` returns EOF. +/// +/// On success, the total number of bytes that were copied from +/// `reader` to `writer` is returned. +/// +/// This is an asynchronous version of [`std::io::copy`][std]. +/// +/// # Errors +/// +/// The returned future will finish with an error will return an error +/// immediately if any call to `poll_read` or `poll_write` returns an error. +/// +/// # Examples +/// +/// ``` +/// use tokio_io as io; +/// +/// # async fn dox() -> std::io::Result<()> { +/// let mut reader: &[u8] = b"hello"; +/// let mut writer: Vec = vec![]; +/// +/// io::copy(&mut reader, &mut writer).await?; +/// +/// assert_eq!(&b"hello"[..], &writer[..]); +/// # Ok(()) +/// # } +/// ``` +/// +/// [std]: https://doc.rust-lang.org/std/io/fn.copy.html +pub fn copy<'a, R, W>(reader: &'a mut R, writer: &'a mut W) -> Copy<'a, R, W> where R: AsyncRead + Unpin + ?Sized, W: AsyncWrite + Unpin + ?Sized, diff --git a/tokio-io/src/io/empty.rs b/tokio-io/src/io/empty.rs new file mode 100644 index 000000000..3717ebc89 --- /dev/null +++ b/tokio-io/src/io/empty.rs @@ -0,0 +1,79 @@ +use crate::{AsyncBufRead, AsyncRead}; +use std::fmt; +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +// An async reader which is always at EOF. +/// +/// This struct is generally created by calling [`empty`]. Please see +/// the documentation of [`empty()`][`empty`] for more details. +/// +/// This is an asynchronous version of [`std::io::empty`][std]. +/// +/// [`empty`]: fn.empty.html +/// [std]: https://doc.rust-lang.org/std/io/struct.Empty.html +pub struct Empty { + _p: (), +} + +/// Creates a new empty async reader. +/// +/// All reads from the returned reader will return `Poll::Ready(Ok(0))`. +/// +/// This is an asynchronous version of [`std::io::empty`][std]. +/// +/// # Examples +/// +/// A slightly sad example of not reading anything into a buffer: +/// +/// ```rust +/// # use tokio_io::{self as io, AsyncReadExt}; +/// # async fn dox() { +/// let mut buffer = String::new(); +/// io::empty().read_to_string(&mut buffer).await.unwrap(); +/// assert!(buffer.is_empty()); +/// # } +/// ``` +/// +/// [std]: https://doc.rust-lang.org/std/io/fn.empty.html +pub fn empty() -> Empty { + Empty { _p: () } +} + +impl AsyncRead for Empty { + #[inline] + fn poll_read( + self: Pin<&mut Self>, + _: &mut Context<'_>, + _: &mut [u8], + ) -> Poll> { + Poll::Ready(Ok(0)) + } +} + +impl AsyncBufRead for Empty { + #[inline] + fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(&[])) + } + + #[inline] + fn consume(self: Pin<&mut Self>, _: usize) {} +} + +impl fmt::Debug for Empty { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.pad("Empty { .. }") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::(); + } +} diff --git a/tokio-io/src/io/mod.rs b/tokio-io/src/io/mod.rs index ab3a2b42d..522188a7f 100644 --- a/tokio-io/src/io/mod.rs +++ b/tokio-io/src/io/mod.rs @@ -6,6 +6,7 @@ mod buf_stream; mod buf_writer; mod chain; mod copy; +mod empty; mod flush; mod lines; mod read; @@ -14,7 +15,9 @@ mod read_line; mod read_to_end; mod read_to_string; mod read_until; +mod repeat; mod shutdown; +mod sink; mod take; mod write; mod write_all; @@ -31,6 +34,14 @@ pub use self::buf_reader::BufReader; pub use self::buf_stream::BufStream; #[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 pub use self::buf_writer::BufWriter; +#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 +pub use self::copy::{copy, Copy}; +#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 +pub use self::empty::{empty, Empty}; +#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 +pub use self::repeat::{repeat, Repeat}; +#[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411 +pub use self::sink::{sink, Sink}; // used by `BufReader` and `BufWriter` // https://github.com/rust-lang/rust/blob/master/src/libstd/sys_common/io.rs#L1 diff --git a/tokio-io/src/io/repeat.rs b/tokio-io/src/io/repeat.rs new file mode 100644 index 000000000..79a1df171 --- /dev/null +++ b/tokio-io/src/io/repeat.rs @@ -0,0 +1,67 @@ +use crate::AsyncRead; + +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +/// An async reader which yields one byte over and over and over and over and +/// over and... +/// +/// This struct is generally created by calling [`repeat`][repeat]. Please +/// see the documentation of `repeat()` for more details. +/// +/// This is an asynchronous version of [`std::io::Repeat`][std]. +/// +/// [repeat]: fn.repeat.html +/// [std]: https://doc.rust-lang.org/std/io/struct.Repeat.html +#[derive(Debug)] +pub struct Repeat { + byte: u8, +} + +/// Creates an instance of an async reader that infinitely repeats one byte. +/// +/// All reads from this reader will succeed by filling the specified buffer with +/// the given byte. +/// +/// This is an asynchronous version of [`std::io::repeat`][std]. +/// +/// # Examples +/// +/// ``` +/// # use tokio_io::{self as io, AsyncReadExt}; +/// # async fn dox() { +/// let mut buffer = [0; 3]; +/// io::repeat(0b101).read_exact(&mut buffer).await.unwrap(); +/// assert_eq!(buffer, [0b101, 0b101, 0b101]); +/// # } +/// ``` +/// +/// [std]: https://doc.rust-lang.org/std/io/fn.repeat.html +pub fn repeat(byte: u8) -> Repeat { + Repeat { byte } +} + +impl AsyncRead for Repeat { + #[inline] + fn poll_read( + self: Pin<&mut Self>, + _: &mut Context<'_>, + buf: &mut [u8], + ) -> Poll> { + for byte in &mut *buf { + *byte = self.byte; + } + Poll::Ready(Ok(buf.len())) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::(); + } +} diff --git a/tokio-io/src/io/sink.rs b/tokio-io/src/io/sink.rs new file mode 100644 index 000000000..75ade2145 --- /dev/null +++ b/tokio-io/src/io/sink.rs @@ -0,0 +1,77 @@ +use crate::AsyncWrite; +use std::fmt; +use std::io; +use std::pin::Pin; +use std::task::{Context, Poll}; + +/// An async writer which will move data into the void. +/// +/// This struct is generally created by calling [`sink`][sink]. Please +/// see the documentation of `sink()` for more details. +/// +/// This is an asynchronous version of `std::io::Sink`. +/// +/// [sink]: fn.sink.html +pub struct Sink { + _p: (), +} + +/// Creates an instance of an async writer which will successfully consume all +/// data. +/// +/// All calls to `poll_write` on the returned instance will return +/// `Poll::Ready(Ok(buf.len()))` and the contents of the buffer will not be +/// inspected. +/// +/// This is an asynchronous version of `std::io::sink`. +/// +/// # Examples +/// +/// ```rust +/// # use tokio_io::{self as io, AsyncWriteExt}; +/// # async fn dox() { +/// let buffer = vec![1, 2, 3, 5, 8]; +/// let num_bytes = io::sink().write(&buffer).await.unwrap(); +/// assert_eq!(num_bytes, 5); +/// # } +/// ``` +pub fn sink() -> Sink { + Sink { _p: () } +} + +impl AsyncWrite for Sink { + #[inline] + fn poll_write( + self: Pin<&mut Self>, + _: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Poll::Ready(Ok(buf.len())) + } + + #[inline] + fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + #[inline] + fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } +} + +impl fmt::Debug for Sink { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.pad("Sink { .. }") + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn assert_unpin() { + crate::is_unpin::(); + } +} diff --git a/tokio-io/src/lib.rs b/tokio-io/src/lib.rs index 25dfad4cf..22b415f1b 100644 --- a/tokio-io/src/lib.rs +++ b/tokio-io/src/lib.rs @@ -34,7 +34,10 @@ pub use self::async_read::AsyncRead; pub use self::async_write::AsyncWrite; #[cfg(feature = "util")] -pub use self::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufStream, BufWriter}; +pub use self::io::{ + copy, empty, repeat, sink, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, BufStream, + BufWriter, Copy, Empty, Repeat, Sink, +}; // Re-export `Buf` and `BufMut` since they are part of the API pub use bytes::{Buf, BufMut}; diff --git a/tokio/src/io.rs b/tokio/src/io.rs index f45906e1e..d694e08d5 100644 --- a/tokio/src/io.rs +++ b/tokio/src/io.rs @@ -41,8 +41,8 @@ pub use tokio_fs::{stderr, stdin, stdout, Stderr, Stdin, Stdout}; pub use tokio_io::split::split; pub use tokio_io::{ - AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader, - BufWriter, + empty, repeat, sink, AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite, + AsyncWriteExt, BufReader, BufWriter, Empty, Repeat, Sink, }; // Re-export io::Error so that users don't have to deal