From 7c0aea04b2d2a7c999c6449f0c87dc1b3c15d490 Mon Sep 17 00:00:00 2001 From: Taiki Endo Date: Tue, 27 Aug 2019 19:37:40 +0900 Subject: [PATCH] Use our own io::{Empy, Repeat, Sink} --- futures-io/src/lib.rs | 16 -------- futures-util/src/io/empty.rs | 63 ++++++++++++++++++++++++++++++++ futures-util/src/io/mod.rs | 9 +++++ futures-util/src/io/repeat.rs | 69 +++++++++++++++++++++++++++++++++++ futures-util/src/io/sink.rs | 66 +++++++++++++++++++++++++++++++++ futures/src/lib.rs | 8 ++-- 6 files changed, 211 insertions(+), 20 deletions(-) create mode 100644 futures-util/src/io/empty.rs create mode 100644 futures-util/src/io/repeat.rs create mode 100644 futures-util/src/io/sink.rs diff --git a/futures-io/src/lib.rs b/futures-io/src/lib.rs index 4cb3b95996..ff2aff9f46 100644 --- a/futures-io/src/lib.rs +++ b/futures-io/src/lib.rs @@ -416,14 +416,6 @@ mod if_std { unsafe_delegate_async_read_to_stdio!(); } - impl AsyncRead for io::Repeat { - unsafe_delegate_async_read_to_stdio!(); - } - - impl AsyncRead for io::Empty { - unsafe_delegate_async_read_to_stdio!(); - } - impl + Unpin> AsyncRead for io::Cursor { unsafe_delegate_async_read_to_stdio!(); } @@ -547,10 +539,6 @@ mod if_std { delegate_async_write_to_stdio!(); } - impl AsyncWrite for io::Sink { - delegate_async_write_to_stdio!(); - } - macro_rules! deref_async_seek { () => { fn poll_seek(mut self: Pin<&mut Self>, cx: &mut Context<'_>, pos: SeekFrom) @@ -651,10 +639,6 @@ mod if_std { delegate_async_buf_read_to_stdio!(); } - impl AsyncBufRead for io::Empty { - delegate_async_buf_read_to_stdio!(); - } - impl + Unpin> AsyncBufRead for io::Cursor { delegate_async_buf_read_to_stdio!(); } diff --git a/futures-util/src/io/empty.rs b/futures-util/src/io/empty.rs new file mode 100644 index 0000000000..545231f7d1 --- /dev/null +++ b/futures-util/src/io/empty.rs @@ -0,0 +1,63 @@ +use futures_core::task::{Context, Poll}; +use futures_io::{AsyncBufRead, AsyncRead, Initializer}; +use std::fmt; +use std::io; +use std::pin::Pin; + +/// Stream for the [`empty()`] function. +#[must_use = "streams do nothing unless polled"] +pub struct Empty { + _priv: (), +} + +/// Constructs a new handle to an empty reader. +/// +/// All reads from the returned reader will return `Poll::Ready(Ok(0))`. +/// +/// # Examples +/// +/// A slightly sad example of not reading anything into a buffer: +/// +/// ``` +/// # futures::executor::block_on(async { +/// use futures::io::{self, AsyncReadExt}; +/// +/// let mut buffer = String::new(); +/// io::empty().read_to_string(&mut buffer).await?; +/// assert!(buffer.is_empty()); +/// # Ok::<(), Box>(()) }).unwrap(); +/// ``` +pub fn empty() -> Empty { + Empty { _priv: () } +} + +impl AsyncRead for Empty { + #[inline] + fn poll_read( + self: Pin<&mut Self>, + _: &mut Context<'_>, + _: &mut [u8], + ) -> Poll> { + Poll::Ready(Ok(0)) + } + + #[inline] + unsafe fn initializer(&self) -> Initializer { + Initializer::nop() + } +} + +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 { .. }") + } +} diff --git a/futures-util/src/io/mod.rs b/futures-util/src/io/mod.rs index 98ea18ee53..1657cf5fc4 100644 --- a/futures-util/src/io/mod.rs +++ b/futures-util/src/io/mod.rs @@ -41,6 +41,9 @@ pub use self::copy_into::CopyInto; mod copy_buf_into; pub use self::copy_buf_into::CopyBufInto; +mod empty; +pub use self::empty::{empty, Empty}; + mod flush; pub use self::flush::Flush; @@ -73,9 +76,15 @@ pub use self::read_to_string::ReadToString; mod read_until; pub use self::read_until::ReadUntil; +mod repeat; +pub use self::repeat::{repeat, Repeat}; + mod seek; pub use self::seek::Seek; +mod sink; +pub use self::sink::{sink, Sink}; + mod split; pub use self::split::{ReadHalf, WriteHalf}; diff --git a/futures-util/src/io/repeat.rs b/futures-util/src/io/repeat.rs new file mode 100644 index 0000000000..fc66e0ffe6 --- /dev/null +++ b/futures-util/src/io/repeat.rs @@ -0,0 +1,69 @@ +use futures_core::task::{Context, Poll}; +use futures_io::{AsyncRead, Initializer, IoSliceMut}; +use std::fmt; +use std::io; +use std::pin::Pin; + +/// Stream for the [`repeat()`] function. +#[must_use = "streams do nothing unless polled"] +pub struct Repeat { + byte: u8, +} + +/// Creates an instance of a reader that infinitely repeats one byte. +/// +/// All reads from this reader will succeed by filling the specified buffer with +/// the given byte. +/// +/// # Examples +/// +/// ``` +/// # futures::executor::block_on(async { +/// use futures::io::{self, AsyncReadExt}; +/// +/// let mut buffer = [0; 3]; +/// io::repeat(0b101).read_exact(&mut buffer).await.unwrap(); +/// assert_eq!(buffer, [0b101, 0b101, 0b101]); +/// # Ok::<(), Box>(()) }).unwrap(); +/// ``` +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 slot in &mut *buf { + *slot = self.byte; + } + Poll::Ready(Ok(buf.len())) + } + + #[inline] + fn poll_read_vectored( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &mut [IoSliceMut<'_>], + ) -> Poll> { + let mut nwritten = 0; + for buf in bufs { + nwritten += ready!(self.as_mut().poll_read(cx, buf))?; + } + Poll::Ready(Ok(nwritten)) + } + + #[inline] + unsafe fn initializer(&self) -> Initializer { + Initializer::nop() + } +} + +impl fmt::Debug for Repeat { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.pad("Repeat { .. }") + } +} diff --git a/futures-util/src/io/sink.rs b/futures-util/src/io/sink.rs new file mode 100644 index 0000000000..ee027a1d04 --- /dev/null +++ b/futures-util/src/io/sink.rs @@ -0,0 +1,66 @@ +use futures_core::task::{Context, Poll}; +use futures_io::{AsyncWrite, IoSlice}; +use std::fmt; +use std::io; +use std::pin::Pin; + +/// Stream for the [`sink()`] function. +#[must_use = "streams do nothing unless polled"] +pub struct Sink { + _priv: (), +} + +/// Creates an instance of a 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. +/// +/// # Examples +/// +/// ```rust +/// # futures::executor::block_on(async { +/// use futures::io::{self, AsyncWriteExt}; +/// +/// let buffer = vec![1, 2, 3, 5, 8]; +/// let num_bytes = io::sink().write(&buffer).await?; +/// assert_eq!(num_bytes, 5); +/// # Ok::<(), Box>(()) }).unwrap(); +/// ``` +pub fn sink() -> Sink { + Sink { _priv: () } +} + +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_write_vectored( + self: Pin<&mut Self>, + _: &mut Context<'_>, + bufs: &[IoSlice<'_>], + ) -> Poll> { + Poll::Ready(Ok(bufs.iter().map(|b| b.len()).sum())) + } + + #[inline] + fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + #[inline] + fn poll_close(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 { .. }") + } +} diff --git a/futures/src/lib.rs b/futures/src/lib.rs index 16da728262..395b78766c 100644 --- a/futures/src/lib.rs +++ b/futures/src/lib.rs @@ -300,10 +300,10 @@ pub mod io { pub use futures_util::io::{ AsyncReadExt, AsyncWriteExt, AsyncSeekExt, AsyncBufReadExt, AllowStdIo, - BufReader, BufWriter, Chain, Close, CopyInto, CopyBufInto, Flush, - IntoSink, Lines, Read, ReadExact, ReadHalf, ReadLine, ReadToEnd, - ReadToString, ReadUntil, ReadVectored, Seek, Take, Window, Write, - WriteAll, WriteHalf, WriteVectored, + BufReader, BufWriter, Chain, Close, CopyInto, CopyBufInto, empty, Empty, + Flush, IntoSink, Lines, Read, ReadExact, ReadHalf, ReadLine, ReadToEnd, + ReadToString, ReadUntil, ReadVectored, repeat, Repeat, Seek, sink, Sink, + Take, Window, Write, WriteAll, WriteHalf, WriteVectored, }; }