feat(io): add BytesFramed support (#749)

* feat: add BytesCodec, CapacityDelimiter, BytesFramed and BytesFramedExt trait

* docs: added BytesFramed documentation, renamed CapacityDelimiter to NoopFramer.

* refactor(io): removed BytesFramed, added new_bytes as a Frame implementation.

* docs(io): fix NoopFramer api doc description

* docs(io): fix BytesCoded api doc text

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

* refactor(io): added BytesFramed type alias back for better readability

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

* refactor(io): return Self and make it more readable

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

* refactor(io): remove unnecessary reserve

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

* style(io): some code formatting

* style(io): fixed code formatting on api doc example

* style: fixed lint issue

* style: ommit this function arg

* fix: clippy

* fix: compio-io/tests/framed.rs

---------

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

Author: ararog

compio-io/src/framed/codec/bytes.rs

@@ -0,0 +1,65 @@
+//! [`Encoder`]/[`Decoder`] implementation with Bytes
+//!
+//! This module provides a codec implementation for bytes serialization and
+//! deserialization (noop per se).
+//!
+//! # Examples
+//!
+//! ```
+//! use compio_buf::{IoBuf, bytes::Bytes};
+//! use compio_io::framed::codec::{Decoder, Encoder, bytes::BytesCodec};
+//!
+//! let mut codec = BytesCodec::new();
+//! let data = Bytes::from("Hello, world!");
+//!
+//! // Encoding
+//! let mut buffer = Vec::new();
+//! codec.encode(data.clone(), &mut buffer).unwrap();
+//!
+//! // Decoding
+//! let decoded = codec.decode(&buffer.as_slice()).unwrap();
+//! assert_eq!(decoded, data);
+//! ```
+use std::io::{self, Write};
+
+use compio_buf::{IoBuf, IoBufMut, Slice, bytes::Bytes};
+
+use crate::framed::codec::{Decoder, Encoder};
+
+/// A codec for bytes serialization and deserialization.
+///
+/// This codec can be used to write into and read from [`Bytes`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct BytesCodec;
+
+impl BytesCodec {
+    /// Creates a new `BytesCodec`.
+    pub fn new() -> Self {
+        Self {}
+    }
+}
+
+impl Default for BytesCodec {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<B: IoBufMut> Encoder<Bytes, B> for BytesCodec {
+    type Error = io::Error;
+
+    fn encode(&mut self, item: Bytes, buf: &mut B) -> Result<(), Self::Error> {
+        let mut writer = buf.as_writer();
+        writer.write_all(&item)?;
+        Ok(())
+    }
+}
+
+impl<B: IoBuf> Decoder<Bytes, B> for BytesCodec {
+    type Error = io::Error;
+
+    fn decode(&mut self, buf: &Slice<B>) -> Result<Bytes, Self::Error> {
+        let inner = buf.as_ref().to_vec();
+        Ok(Bytes::from(inner))
+    }
+}

compio-io/src/framed/codec/mod.rs

@@ -5,6 +5,8 @@ use std::io;
 
 use compio_buf::{IoBuf, IoBufMut, Slice};
 
+pub mod bytes;
+
 #[cfg(feature = "codec-serde-json")]
 pub mod serde_json;
 

compio-io/src/framed/frame.rs

@@ -226,6 +226,50 @@ impl<B: IoBufMut> Framer<B> for AnyDelimited<'_> {
 /// Delimiter that uses newline characters (`\n`) as delimiters.
 pub type LineDelimited = CharDelimited<'\n'>;
 
+/// A framer that does nothing.
+///
+/// It simply reserves space in the buffer without adding any framing.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct NoopFramer {
+    max_size: usize,
+}
+
+impl Default for NoopFramer {
+    fn default() -> Self {
+        Self { max_size: 4096 }
+    }
+}
+
+impl NoopFramer {
+    /// Creates a new `NoopFramer` framer.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Returns the size of the capacity.
+    pub fn max_size(&self) -> usize {
+        self.max_size
+    }
+}
+
+impl<B: IoBufMut> Framer<B> for NoopFramer {
+    fn enclose(&mut self, _: &mut B) {}
+
+    fn extract(&mut self, buf: &Slice<B>) -> io::Result<Option<Frame>> {
+        if buf.is_empty() {
+            return Ok(None);
+        }
+
+        let len = if buf.len() < self.max_size {
+            buf.len()
+        } else {
+            self.max_size
+        };
+
+        Ok(Some(Frame::new(0, len, 0)))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use compio_buf::{IntoInner, IoBufMut};
@@ -248,6 +292,22 @@ mod tests {
         assert_eq!(payload.as_init(), b"hello");
     }
 
+    #[test]
+    fn test_noop_framer() {
+        let mut framer = NoopFramer::new();
+
+        let mut buf = Vec::from(b"hello");
+        framer.enclose(&mut buf);
+        assert_eq!(&buf.as_slice()[..5], b"hello");
+
+        let buf = buf.slice(..);
+        let frame = framer.extract(&buf).unwrap().unwrap();
+        let buf = buf.into_inner();
+        assert_eq!(frame, Frame::new(0, 5, 0));
+        let payload = frame.slice(buf);
+        assert_eq!(payload.as_init(), b"hello");
+    }
+
     #[test]
     fn test_char_delimited() {
         let mut framer = CharDelimited::<'ℝ'>::new();

compio-io/src/framed/mod.rs

@@ -5,10 +5,17 @@
 
 use std::marker::PhantomData;
 
-use compio_buf::IoBufMut;
+use compio_buf::{IoBufMut, bytes::Bytes};
 use futures_util::FutureExt;
 
-use crate::{AsyncRead, framed::codec::Decoder, util::Splittable};
+use crate::{
+    AsyncRead,
+    framed::{
+        codec::{Decoder, bytes::BytesCodec},
+        frame::NoopFramer,
+    },
+    util::Splittable,
+};
 
 pub mod codec;
 pub mod frame;
@@ -143,3 +150,19 @@ impl<C, F> Framed<(), (), C, F, (), (), ()> {
         }
     }
 }
+
+type BytesFramed<R, W> = Framed<R, W, BytesCodec, NoopFramer, Bytes, Bytes>;
+
+impl BytesFramed<(), ()> {
+    /// Creates a new `Framed` with the given I/O object, codec, and framer with
+    /// bytes as the input and output type.
+    pub fn new_bytes() -> Self {
+        Framed {
+            read_state: read::State::empty(),
+            write_state: write::State::empty(),
+            codec: BytesCodec::new(),
+            framer: NoopFramer::new(),
+            types: PhantomData,
+        }
+    }
+}

compio-io/tests/framed.rs

@@ -5,7 +5,7 @@ use std::{
 
 use compio_buf::{BufResult, IoBuf, IoBufMut};
 use compio_io::{
-    AsyncRead, AsyncReadAt, AsyncWrite, AsyncWriteAt,
+    AsyncRead, AsyncReadAt, AsyncWrite, AsyncWriteAt, BufReader,
     framed::{Framed, codec::serde_json::SerdeJsonCodec, frame::LengthDelimited},
 };
 use futures_executor::block_on;
@@ -104,3 +104,21 @@ fn test_framed() {
         assert!(res.is_none());
     })
 }
+
+#[test]
+fn test_bytes_framed() {
+    block_on(async {
+        let buf = b"Hello, world!".to_vec();
+        let r = BufReader::with_capacity(5, Cursor::new(buf));
+
+        let mut framed = Framed::new_bytes().with_reader(r);
+        let mut s = String::new();
+        while let Some(result) = framed.next().await {
+            let bytes = result.unwrap();
+            s.push_str(str::from_utf8(&bytes).unwrap());
+        }
+        assert_eq!(s, "Hello, world!");
+        let res = framed.next().await;
+        assert!(res.is_none());
+    })
+}