From e614b4e9313f7f77aa3dd65f61c8df1cc63c1ddc Mon Sep 17 00:00:00 2001 From: Povilas Balciunas Date: Sun, 15 Dec 2019 13:44:46 +1300 Subject: [PATCH 1/4] Add tests for length delimited codec --- tests/length_delimited.rs | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) create mode 100644 tests/length_delimited.rs diff --git a/tests/length_delimited.rs b/tests/length_delimited.rs new file mode 100644 index 0000000..ea8d188 --- /dev/null +++ b/tests/length_delimited.rs @@ -0,0 +1,29 @@ +use bytes::Bytes; +use futures::io::Cursor; +use futures::{executor, SinkExt, StreamExt}; +use futures_codec::{Framed, LengthCodec}; + +#[test] +fn same_msgs_are_received_as_were_sent() { + let cur = Cursor::new(vec![0; 256]); + let mut framed = Framed::new(cur, LengthCodec {}); + + let send_msgs = async { + framed.send(Bytes::from("msg1")).await.unwrap(); + framed.send(Bytes::from("msg2")).await.unwrap(); + framed.send(Bytes::from("msg3")).await.unwrap(); + }; + executor::block_on(send_msgs); + + let (mut cur, _) = framed.release(); + cur.set_position(0); + let framed = Framed::new(cur, LengthCodec {}); + + let recv_msgs = framed.take(3) + .map(|res| res.unwrap()) + .map(|buf| String::from_utf8(buf.to_vec()).unwrap()) + .collect::>(); + let msgs: Vec = executor::block_on(recv_msgs); + + assert!(msgs == vec!["msg1", "msg2", "msg3"]); +} From d058b401059b11ddbe6b1ac7644acafd7fbb3afa Mon Sep 17 00:00:00 2001 From: Povilas Balciunas Date: Sat, 21 Dec 2019 12:16:21 +1300 Subject: [PATCH 2/4] Add failing test This test panics at decode() if src.len() - U64_LENGTH because of underflow. --- src/codec/length.rs | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/codec/length.rs b/src/codec/length.rs index e90cb74..8cc4673 100644 --- a/src/codec/length.rs +++ b/src/codec/length.rs @@ -75,3 +75,23 @@ impl Decoder for LengthCodec { } } } + +#[cfg(test)] +mod tests { + use super::*; + + mod decode { + use super::*; + + #[test] + fn it_returns_bytes_withouth_length_header() { + let mut codec = LengthCodec{ }; + + let mut src = BytesMut::with_capacity(5); + src.put(&[0, 0, 0, 0, 0, 0, 0, 3u8, 1, 2, 3, 4][..]); + let item = codec.decode(&mut src).unwrap(); + + assert!(item == Some(Bytes::from(&[1u8, 2, 3][..]))); + } + } +} From d19e61aa7ced819c1316ad59558da56d3f04bf72 Mon Sep 17 00:00:00 2001 From: Povilas Balciunas Date: Sat, 21 Dec 2019 12:54:56 +1300 Subject: [PATCH 3/4] Fix length encoded frame parsing The problem was that 'src' buffer was advanced too soon. This caused a couple of bugs at least in 2 scenarios: 1. when encoded frame was smaller than 8 bytes. 2. when 'src' did not have enough bytes do decode a complete frame. ## Problem 1 Say we encode a frame [1, 2, 3]. The resulting buffer is [0, 0, 0, 0, 0, 0, 0, 3, 1, 2, 3] - with prepended length header. Then decode() takes this buffer, correctly parses len as 3, and then does this comparison: if src.len() - U64_LENGTH Given that we already advanced src buffer, src.len() now is 3 (11 - 8). This results in uderflow: 3 - 8, which panics. ## Problem 2 Say we have framed a TCP stream where data is coming in chunks but not necessarily in the size of our frames. Say we call decode() for the first time with src = [0, 0, 0, 0, 0, 0, 0, 64, 1, 2, 3]. Obviously this is not a full frame: it's size is 64 but only first 3 bytes have arrived. In this case decode() used to advance the src buffer, but still return None. So when a next TCP packet arrives, decode() was called for the second time. decode() then used to attempt to parse len header again. And this time it would parse the bytes [1, 2, 3, ...] as a len header which is actually our data. --- src/codec/length.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/codec/length.rs b/src/codec/length.rs index 8cc4673..2494559 100644 --- a/src/codec/length.rs +++ b/src/codec/length.rs @@ -67,8 +67,13 @@ impl Decoder for LengthCodec { return Ok(None); } - let len = src.get_u64() as usize; + let mut len_bytes = [0u8; U64_LENGTH]; + len_bytes.copy_from_slice(&src[..U64_LENGTH]); + let len = u64::from_be_bytes(len_bytes) as usize; + if src.len() - U64_LENGTH >= len { + // Skip the length header we already read. + src.advance(U64_LENGTH); Ok(Some(src.split_to(len).freeze())) } else { Ok(None) From b286b1496699d6bbd8cc21f61c369b1e412294ea Mon Sep 17 00:00:00 2001 From: Povilas Balciunas Date: Sun, 15 Dec 2019 14:58:57 +1300 Subject: [PATCH 4/4] Implement codec for types supporting serde serialization --- Cargo.toml | 7 +++++- src/codec/mod.rs | 3 +++ src/codec/serde.rs | 60 ++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 2 +- tests/serde.rs | 47 ++++++++++++++++++++++++++++++++++++ 5 files changed, 117 insertions(+), 2 deletions(-) create mode 100644 src/codec/serde.rs create mode 100644 tests/serde.rs diff --git a/Cargo.toml b/Cargo.toml index 9ea81a2..91582ae 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,8 @@ categories = ["asynchronous", "network-programming"] [features] json = [ "serde", "serde_json" ] cbor = [ "serde", "serde_cbor" ] -default = [ "json", "cbor" ] +serde_bincode = [ "serde", "bincode" ] +default = [ "json", "cbor", "serde_bincode" ] [dependencies] @@ -36,3 +37,7 @@ optional = true [dependencies.serde_cbor] version = '0.10.2' optional = true + +[dependencies.bincode] +version = "1.2.1" +optional = true diff --git a/src/codec/mod.rs b/src/codec/mod.rs index f39f7b9..0eaca3b 100644 --- a/src/codec/mod.rs +++ b/src/codec/mod.rs @@ -12,3 +12,6 @@ pub use self::lines::LinesCodec; #[cfg(feature = "cbor")] mod cbor; #[cfg(feature = "cbor")] pub use self::cbor::{CborCodec, CborCodecError}; + +#[cfg(feature = "bincode")] mod serde; +#[cfg(feature = "bincode")] pub use self::serde::SerdeCodec; diff --git a/src/codec/serde.rs b/src/codec/serde.rs new file mode 100644 index 0000000..840df2f --- /dev/null +++ b/src/codec/serde.rs @@ -0,0 +1,60 @@ +use std::io; +use std::marker::PhantomData; +use bytes::{BytesMut, Bytes}; +use serde::Serialize; +use serde::de::DeserializeOwned; +use bincode; + +use crate::encoder::Encoder; +use crate::decoder::Decoder; +use crate::codec::LengthCodec; + +/// Encodes/decodes types implementing Serde Serialize/Deserialize traits. +/// It is built on top of `LengthCodec`. +pub struct SerdeCodec { + inner: LengthCodec, + phantom: PhantomData, +} + +impl Default for SerdeCodec { + fn default() -> Self { + Self { + inner: LengthCodec {}, + phantom: PhantomData, + } + } +} + +impl Encoder for SerdeCodec { + type Item = T; + type Error = io::Error; + + fn encode(&mut self, src: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> { + let data = bincode::serialize(&src).map_err(to_io_err)?; + let bytes = Bytes::from(data); + self.inner.encode(bytes, dst) + } +} + +impl Decoder for SerdeCodec { + type Item = T; + type Error = io::Error; + + fn decode(&mut self, src: &mut BytesMut) -> Result, Self::Error> { + match self.inner.decode(src)? { + Some(bytes) => { + bincode::deserialize(&bytes) + .map_err(to_io_err) + .map(|item| Some(item)) + } + None => Ok(None), + } + } +} + +fn to_io_err(err: Box) -> io::Error { + match *err { + bincode::ErrorKind::Io(e) => e, + other => io::Error::new(io::ErrorKind::Other, other), + } +} diff --git a/src/lib.rs b/src/lib.rs index 9c2309f..8096099 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,7 +22,7 @@ //! ``` mod codec; -pub use codec::{BytesCodec, LengthCodec, LinesCodec}; +pub use codec::{BytesCodec, LengthCodec, LinesCodec, SerdeCodec}; #[cfg(feature = "json")] pub use codec::{JsonCodec, JsonCodecError}; #[cfg(feature = "cbor")] pub use codec::{CborCodec, CborCodecError}; diff --git a/tests/serde.rs b/tests/serde.rs new file mode 100644 index 0000000..36d8a42 --- /dev/null +++ b/tests/serde.rs @@ -0,0 +1,47 @@ +use futures::io::Cursor; +use futures::{executor, SinkExt, StreamExt}; +use futures_codec::{Framed, SerdeCodec}; +use serde::{Serialize, Deserialize}; + +#[derive(Serialize, Deserialize, Debug, PartialEq)] +struct Person { + name: String, + age: u8, +} + +impl Person { + fn new(name: &str, age: u8) -> Self { + Self { + name: name.into(), + age, + } + } +} + +#[test] +fn serializes_serde_enabled_structures() { + let cur = Cursor::new(vec![0; 4096]); + let mut framed = Framed::new(cur, SerdeCodec::default()); + + let send_msgs = async { + framed.send(Person::new("John", 11)).await.unwrap(); + framed.send(Person::new("Paul", 12)).await.unwrap(); + framed.send(Person::new("Mike", 13)).await.unwrap(); + }; + executor::block_on(send_msgs); + + let (mut cur, _) = framed.release(); + cur.set_position(0); + let framed = Framed::new(cur, SerdeCodec::default()); + + let recv_msgs = framed.take(3) + .map(|res| res.unwrap()) + .collect::>(); + let items: Vec = executor::block_on(recv_msgs); + + assert!(items == vec![ + Person::new("John", 11), + Person::new("Paul", 12), + Person::new("Mike", 13), + ]) +}