feat(io): allow create a stream out of AsyncRead for BytesFramed (#767)

* fix(io): make also bytes method on AsyncReadExt/AsyncWriteExt feature gated

* feat(io): allow create a stream out of AsyncRead for BytesFramed

* test(io): removed file stream test

* refactor(io): renamed methods to read_only and write_only

* refactor(fs): removed unused imports

* test(fs): added unit test for file streaming

* feat: more generic {read,write}_only

* fix: test

---------

Co-authored-by: George Miao <gm@miao.dev>

Author: ararog

compio-fs/tests/file.rs

@@ -1,7 +1,8 @@
 use std::io::prelude::*;
 
 use compio_fs::{File, OpenOptions};
-use compio_io::{AsyncReadAtExt, AsyncWriteAt, AsyncWriteAtExt};
+use compio_io::{AsyncReadAtExt, AsyncReadExt, AsyncWriteAt, AsyncWriteAtExt};
+use futures_util::StreamExt;
 use tempfile::NamedTempFile;
 
 async fn setlen_run(file: &File, size: u64) {
@@ -43,6 +44,22 @@ async fn metadata() {
     assert_eq!(size, std_meta.len());
 }
 
+#[compio_macros::test]
+async fn file_stream() {
+    let mut tempfile = tempfile();
+    tempfile.write_all(HELLO).unwrap();
+
+    let file = File::open(tempfile.path()).await.unwrap();
+    let cursor = std::io::Cursor::new(file);
+    let mut file = cursor.read_only().bytes();
+    let mut s = String::new();
+    while let Some(result) = file.next().await {
+        let chunk = result.unwrap();
+        s.push_str(str::from_utf8(&chunk).unwrap());
+    }
+    assert_eq!(s, String::from_utf8_lossy(HELLO));
+}
+
 const HELLO: &[u8] = b"hello world...";
 
 async fn read_hello(file: &File) {

compio-io/src/read/ext.rs

@@ -201,16 +201,43 @@ pub trait AsyncReadExt: AsyncRead {
         Framed::new(codec, framer).with_duplex(self)
     }
 
-    #[cfg(feature = "bytes")]
     /// Convenience method to create a [`BytesFramed`] reader/writter
     /// out of a splittable.
+    #[cfg(feature = "bytes")]
     fn bytes(self) -> BytesFramed<Self::ReadHalf, Self::WriteHalf>
     where
         Self: Splittable + Sized,
     {
         BytesFramed::new_bytes().with_duplex(self)
     }
 
+    /// Create a [`Splittable`] that uses `Self` as [`ReadHalf`] and `()` as
+    /// [`WriteHalf`].
+    ///
+    /// This is useful for creating framed sink with only a reader,
+    /// using the [`AsyncReadExt::framed`] or [`AsyncReadExt::bytes`]
+    /// method, which require a [`Splittable`] to work.
+    ///
+    /// # Examples
+    ///
+    /// ```rust,ignore
+    /// use compio_io::{AsyncReadExt, framed::BytesFramed};
+    ///
+    /// let mut file_bytes = file.read_only().bytes();
+    /// while let Some(Ok(bytes)) = file_bytes.next().await {
+    ///     // process bytes
+    /// }
+    /// ```
+    ///
+    /// [`ReadHalf`]: Splittable::ReadHalf
+    /// [`WriteHalf`]: Splittable::WriteHalf
+    fn read_only(self) -> ReadOnly<Self>
+    where
+        Self: Sized,
+    {
+        ReadOnly(self)
+    }
+
     /// Creates an adaptor which reads at most `limit` bytes from it.
     ///
     /// This function returns a new instance of `AsyncRead` which will read
@@ -313,3 +340,29 @@ pub trait AsyncReadAtExt: AsyncReadAt {
 }
 
 impl<A: AsyncReadAt + ?Sized> AsyncReadAtExt for A {}
+
+/// An adaptor which implements [`Splittable`] for any [`AsyncRead`], with the
+/// write half being `()`.
+///
+/// This can be used to create a framed stream with only a reader, using
+/// the [`AsyncReadExt::framed`] or [`AsyncReadExt::bytes`] method.
+pub struct ReadOnly<R>(pub R);
+
+impl<R: AsyncRead> AsyncRead for ReadOnly<R> {
+    async fn read<T: IoBufMut>(&mut self, buf: T) -> BufResult<usize, T> {
+        self.0.read(buf).await
+    }
+
+    async fn read_vectored<T: IoVectoredBufMut>(&mut self, buf: T) -> BufResult<usize, T> {
+        self.0.read_vectored(buf).await
+    }
+}
+
+impl<R> Splittable for ReadOnly<R> {
+    type ReadHalf = R;
+    type WriteHalf = ();
+
+    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
+        (self.0, ())
+    }
+}

compio-io/src/write/ext.rs

@@ -132,16 +132,41 @@ pub trait AsyncWriteExt: AsyncWrite {
         Framed::new(codec, framer).with_duplex(self)
     }
 
-    #[cfg(feature = "bytes")]
     /// Convenience method to create a [`BytesFramed`] reader/writer
     /// out of a splittable.
+    #[cfg(feature = "bytes")]
     fn bytes(self) -> BytesFramed<Self::ReadHalf, Self::WriteHalf>
     where
         Self: Splittable + Sized,
     {
         BytesFramed::new_bytes().with_duplex(self)
     }
 
+    /// Create a [`Splittable`] that uses `Self` as [`WriteHalf`] and `()` as
+    /// [`ReadHalf`].
+    ///
+    /// This is useful for creating framed sink with only a writer,
+    /// using the [`AsyncWriteExt::framed`] or [`AsyncWriteExt::bytes`]
+    /// method, which require a [`Splittable`] to work.
+    ///
+    /// # Examples
+    ///
+    /// ```rust,ignore
+    /// use compio_io::{AsyncWriteExt, framed::BytesFramed};
+    ///
+    /// let mut file_bytes = file.write_only().bytes();
+    /// file_bytes.send(Bytes::from("hello world")).await?;
+    /// ```
+    ///
+    /// [`ReadHalf`]: Splittable::ReadHalf
+    /// [`WriteHalf`]: Splittable::WriteHalf
+    fn write_only(self) -> WriteOnly<Self>
+    where
+        Self: Sized,
+    {
+        WriteOnly(self)
+    }
+
     write_scalar!(u8, to_be_bytes, to_le_bytes);
     write_scalar!(u16, to_be_bytes, to_le_bytes);
     write_scalar!(u32, to_be_bytes, to_le_bytes);
@@ -186,3 +211,33 @@ pub trait AsyncWriteAtExt: AsyncWriteAt {
 }
 
 impl<A: AsyncWriteAt + ?Sized> AsyncWriteAtExt for A {}
+
+/// An adaptor which implements [`Splittable`] for any [`AsyncWrite`], with the
+/// read half being `()`.
+///
+/// This can be used to create a framed sink with only a writer, using
+/// the [`AsyncWriteExt::framed`] or [`AsyncWriteExt::bytes`] method.
+pub struct WriteOnly<W>(pub W);
+
+impl<W: AsyncWrite> AsyncWrite for WriteOnly<W> {
+    async fn write<T: IoBuf>(&mut self, buf: T) -> BufResult<usize, T> {
+        self.0.write(buf).await
+    }
+
+    async fn flush(&mut self) -> IoResult<()> {
+        self.0.flush().await
+    }
+
+    async fn shutdown(&mut self) -> IoResult<()> {
+        self.0.shutdown().await
+    }
+}
+
+impl<W> Splittable for WriteOnly<W> {
+    type ReadHalf = ();
+    type WriteHalf = W;
+
+    fn split(self) -> (Self::ReadHalf, Self::WriteHalf) {
+        ((), self.0)
+    }
+}