fix(codec): do not advance buffer on eof and error (#8)

Co-authored-by: Sam Andreae <[email protected]>
Co-authored-by: Sam Andreae <[email protected]>

CHANGELOG.md

@@ -6,6 +6,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic
 Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.6.2]
+
+### Fixed
+
+- Error handling in `CborCodec` and `JsonCodec` `decode`, more specifically not advancing the data buffer on partial decoding. See [#7](https://github.com/mxinden/asynchronous-codec/pull/7) for details.
+
 ## [0.6.1] - 2022-11-08
 
 ### Added

Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "asynchronous-codec"
 edition = "2018"
-version = "0.6.1"
+version = "0.6.2"
 authors = ["Max Inden <[email protected]>"]
 description = "Utilities for encoding and decoding frames using `async/await`"
 license = "MIT"

src/codec/cbor.rs

@@ -119,10 +119,10 @@ where
         let res: Result<Dec, _> = serde::de::Deserialize::deserialize(&mut de);
 
         // If we ran out before parsing, return none and try again later
-        let res = match res {
-            Ok(v) => Ok(Some(v)),
-            Err(e) if e.is_eof() => Ok(None),
-            Err(e) => Err(e.into()),
+        let item = match res {
+            Ok(item) => item,
+            Err(e) if e.is_eof() => return Ok(None),
+            Err(e) => return Err(e.into()),
         };
 
         // Update offset from iterator
@@ -131,7 +131,7 @@ where
         // Advance buffer
         buf.advance(offset);
 
-        res
+        Ok(Some(item))
     }
 }
 
@@ -217,4 +217,52 @@ mod test {
 
         assert_eq!(buff.len(), 0);
     }
+
+    #[test]
+    fn cbor_codec_eof_reached() {
+        let mut codec = CborCodec::<TestStruct, TestStruct>::new();
+        let mut buff = BytesMut::new();
+
+        let item1 = TestStruct {
+            name: "Test name".to_owned(),
+            data: 34,
+        };
+        codec.encode(item1.clone(), &mut buff).unwrap();
+
+        // Split the buffer into two.
+        let mut buff_start = buff.clone().split_to(4);
+        let buff_end = buff.clone().split_off(4);
+
+        // Attempt to decode the first half of the buffer. This should return `Ok(None)` and not
+        // advance the buffer.
+        assert_eq!(codec.decode(&mut buff_start).unwrap(), None);
+        assert_eq!(buff_start.len(), 4);
+
+        // Combine the buffer back together.
+        buff_start.extend(buff_end.iter());
+
+        // It should now decode successfully.
+        let item2 = codec.decode(&mut buff).unwrap().unwrap();
+        assert_eq!(item1, item2);
+    }
+
+    #[test]
+    fn cbor_codec_decode_error() {
+        let mut codec = CborCodec::<TestStruct, TestStruct>::new();
+        let mut buff = BytesMut::new();
+
+        let item1 = TestStruct {
+            name: "Test name".to_owned(),
+            data: 34,
+        };
+        codec.encode(item1.clone(), &mut buff).unwrap();
+
+        // Split the end off the buffer.
+        let mut buff_end = buff.clone().split_off(4);
+        let buff_end_length = buff_end.len();
+
+        // Attempting to decode should return an error.
+        assert!(codec.decode(&mut buff_end).is_err());
+        assert_eq!(buff_end.len(), buff_end_length);
+    }
 }

src/codec/json.rs

@@ -115,11 +115,11 @@ where
         let mut iter = de.into_iter::<Dec>();
 
         // Attempt to fetch an item and generate response
-        let res = match iter.next() {
-            Some(Ok(v)) => Ok(Some(v)),
-            Some(Err(ref e)) if e.is_eof() => Ok(None),
-            Some(Err(e)) => Err(e.into()),
-            None => Ok(None),
+        let item = match iter.next() {
+            Some(Ok(item)) => item,
+            Some(Err(ref e)) if e.is_eof() => return Ok(None),
+            Some(Err(e)) => return Err(e.into()),
+            None => return Ok(None),
         };
 
         // Update offset from iterator
@@ -128,7 +128,7 @@ where
         // Advance buffer
         buf.advance(offset);
 
-        res
+        Ok(Some(item))
     }
 }
 
@@ -214,4 +214,52 @@ mod test {
 
         assert_eq!(buff.len(), 0);
     }
+
+    #[test]
+    fn json_codec_eof_reached() {
+        let mut codec = JsonCodec::<TestStruct, TestStruct>::new();
+        let mut buff = BytesMut::new();
+
+        let item1 = TestStruct {
+            name: "Test name".to_owned(),
+            data: 34,
+        };
+        codec.encode(item1.clone(), &mut buff).unwrap();
+
+        // Split the buffer into two.
+        let mut buff_start = buff.clone().split_to(4);
+        let buff_end = buff.clone().split_off(4);
+
+        // Attempt to decode the first half of the buffer. This should return `Ok(None)` and not
+        // advance the buffer.
+        assert_eq!(codec.decode(&mut buff_start).unwrap(), None);
+        assert_eq!(buff_start.len(), 4);
+
+        // Combine the buffer back together.
+        buff_start.extend(buff_end.iter());
+
+        // It should now decode successfully.
+        let item2 = codec.decode(&mut buff).unwrap().unwrap();
+        assert_eq!(item1, item2);
+    }
+
+    #[test]
+    fn json_codec_decode_error() {
+        let mut codec = JsonCodec::<TestStruct, TestStruct>::new();
+        let mut buff = BytesMut::new();
+
+        let item1 = TestStruct {
+            name: "Test name".to_owned(),
+            data: 34,
+        };
+        codec.encode(item1.clone(), &mut buff).unwrap();
+
+        // Split the end off the buffer.
+        let mut buff_end = buff.clone().split_off(4);
+        let buff_end_length = buff_end.len();
+
+        // Attempting to decode should return an error.
+        assert!(codec.decode(&mut buff_end).is_err());
+        assert_eq!(buff_end.len(), buff_end_length);
+    }
 }