Add tokio_util::codec::{Encoder, Decoder} support

This allows CBOR to operate as an underlying frame.

Author: npmccallum

Cargo.toml

@@ -19,6 +19,8 @@ maintenance = { status = "passively-maintained" }
 [dependencies]
 half = "1.2.0"
 serde = { version = "1.0.14", default-features = false }
+tokio-util = { version = "0.3", optional = true }
+bytes = { version = "0.5", optional = true }
 
 [dev-dependencies]
 serde_derive = { version = "1.0.14", default-features = false }
@@ -31,3 +33,4 @@ alloc = ["serde/alloc"]
 std = ["serde/std" ]
 unsealed_read_write = []
 tags = []
+codec = ["tokio-util/codec", "bytes"]

src/codec.rs

@@ -0,0 +1,80 @@
+//! Encoding and decoding for tokio
+
+use crate::error::Category;
+use crate::Error;
+
+use std::marker::PhantomData;
+
+use bytes::{buf::ext::BufMutExt, buf::Buf, BytesMut};
+use serde::de::DeserializeOwned;
+use serde::Serialize;
+use tokio_util::codec;
+
+/// A `tokio_util::codec::Encoder` for CBOR frames
+pub struct Encoder<T: Serialize>(PhantomData<T>);
+
+impl<T: Serialize> Default for Encoder<T> {
+    fn default() -> Self {
+        Self(PhantomData)
+    }
+}
+
+impl<T: Serialize> codec::Encoder<&T> for Encoder<T> {
+    type Error = Error;
+
+    fn encode(&mut self, item: &T, dst: &mut BytesMut) -> Result<(), Error> {
+        crate::to_writer(dst.writer(), item)
+    }
+}
+
+/// A `tokio_util::codec::Decoder` for CBOR frames
+pub struct Decoder<T: DeserializeOwned>(PhantomData<T>);
+
+impl<T: DeserializeOwned> Default for Decoder<T> {
+    fn default() -> Self {
+        Self(PhantomData)
+    }
+}
+
+impl<T: DeserializeOwned> codec::Decoder for Decoder<T> {
+    type Item = T;
+    type Error = Error;
+
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Error> {
+        let mut bytes: &[u8] = src.as_ref();
+        let starting = bytes.len();
+
+        let item: T = match crate::from_reader(&mut bytes) {
+            Err(e) if e.classify() == Category::Eof => return Ok(None),
+            Ok(v) => v,
+            e => e?,
+        };
+
+        let ending = bytes.len();
+        src.advance(starting - ending);
+        Ok(Some(item))
+    }
+}
+
+/// A Codec for CBOR frames
+#[derive(Default)]
+pub struct Codec<T: Serialize, U: DeserializeOwned>(Encoder<T>, Decoder<U>);
+
+impl<T: Serialize, U: DeserializeOwned> codec::Encoder<&T> for Codec<T, U> {
+    type Error = Error;
+
+    #[inline]
+    fn encode(&mut self, item: &T, dst: &mut BytesMut) -> Result<(), Error> {
+        self.0.encode(item, dst)
+    }
+}
+
+impl<T: Serialize, U: DeserializeOwned> codec::Decoder for Codec<T, U> {
+    type Item = U;
+    type Error = Error;
+
+    #[inline]
+    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Error> {
+        self.1.decode(src)
+    }
+}

src/lib.rs

@@ -335,6 +335,9 @@ mod write;
 #[cfg(feature = "std")]
 pub mod value;
 
+#[cfg(feature = "codec")]
+pub mod codec;
+
 // Re-export the [items recommended by serde](https://serde.rs/conventions.html).
 #[doc(inline)]
 pub use crate::de::{Deserializer, StreamDeserializer};

tests/codec.rs

@@ -0,0 +1,47 @@
+#![cfg(feature = "codec")]
+
+use bytes::{BufMut, BytesMut};
+use serde_cbor::{codec::Codec, error::Category};
+use tokio_util::codec::{Decoder, Encoder};
+
+#[test]
+fn decode() {
+    let mut codec = Codec::<u8, u8>::default();
+
+    assert_eq!(
+        codec
+            .decode(&mut b"\xFF"[..].into())
+            .err()
+            .unwrap()
+            .classify(),
+        Category::Syntax
+    );
+
+    assert_eq!(
+        codec
+            .decode(&mut b"\x24"[..].into())
+            .err()
+            .unwrap()
+            .classify(),
+        Category::Data
+    );
+
+    assert_eq!(codec.decode(&mut b"\x07"[..].into()).unwrap().unwrap(), 7);
+
+    let mut buf = BytesMut::with_capacity(2);
+    assert_eq!(codec.decode(&mut buf).unwrap(), None);
+    buf.put_u8(0x18);
+    assert_eq!(codec.decode(&mut buf).unwrap(), None);
+    buf.put_u8(0x18);
+    assert_eq!(codec.decode(&mut buf).unwrap(), Some(24));
+}
+
+#[test]
+fn encode() {
+    let mut codec = Codec::<u8, u8>::default();
+    let mut buf = BytesMut::new();
+
+    codec.encode(&7, &mut buf).unwrap();
+    assert_eq!(buf.len(), 1);
+    assert_eq!(buf[0], 7);
+}