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

@@ -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<u8> = 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,

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<io::Result<usize>> {
+        Poll::Ready(Ok(0))
+    }
+}
+
+impl AsyncBufRead for Empty {
+    #[inline]
+    fn poll_fill_buf(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<&[u8]>> {
+        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::<Empty>();
+    }
+}

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

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<io::Result<usize>> {
+        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::<Repeat>();
+    }
+}

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<Result<usize, io::Error>> {
+        Poll::Ready(Ok(buf.len()))
+    }
+
+    #[inline]
+    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    #[inline]
+    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
+        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::<Sink>();
+    }
+}

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};

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,
+    empty, repeat, sink, AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncReadExt, AsyncWrite,
-    BufWriter,
+    AsyncWriteExt, BufReader, BufWriter, Empty, Repeat, Sink,
 };
 
 // Re-export io::Error so that users don't have to deal