diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/codec/decode.rs | 251 | ||||
| -rw-r--r-- | src/codec/encode.rs | 336 | ||||
| -rw-r--r-- | src/lib.rs | 11 | ||||
| -rw-r--r-- | src/transport/receiver.rs | 13 | ||||
| -rw-r--r-- | src/transport/sender.rs | 12 | ||||
| -rw-r--r-- | src/transport/tcp.rs | 32 | ||||
| -rw-r--r-- | src/transport/udp.rs | 26 | ||||
| -rw-r--r-- | src/types/prefix.rs | 3 | ||||
| -rw-r--r-- | src/types/prefix/count.rs | 23 | ||||
| -rw-r--r-- | src/types/prefix/length.rs | 42 |
10 files changed, 415 insertions, 334 deletions
diff --git a/src/codec/decode.rs b/src/codec/decode.rs index 529c448..d068760 100644 --- a/src/codec/decode.rs +++ b/src/codec/decode.rs @@ -1,150 +1,133 @@ -//! Decoding: reading protocol values from a byte stream. use std::{ collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet}, hash::Hash, - io::Read, sync::Arc, }; -use crate::{DEFAULT_BUFFER_LEN, codec::error::Result}; -use crate::{codec::error::CodecError, context::Context}; +use crate::{codec::error::Result, codec::error::CodecError, context::Context}; macro_rules! impl_decode { - (u8) => { - impl<Data> Decode<Data> for u8 { - fn decode(reader: &mut dyn Read, _: &Context<Data>) -> Result<Self> + ($t:ty) => { + impl<Data> Decode<Data> for $t { + fn decode(buf: &mut &[u8], _: &Context<Data>) -> Result<Self> where Self: Sized, { - let mut byte = [0u8; 1]; - reader.read_exact(&mut byte)?; - Ok(byte[0]) + const BYTES: usize = std::mem::size_of::<$t>(); + if buf.len() < BYTES { + return Err(CodecError::IoError(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "insufficient bytes", + ))); + } + let (bytes, rest) = buf.split_at(BYTES); + *buf = rest; + Ok(<$t>::from_be_bytes(bytes.try_into().unwrap())) } - fn decode_slice(reader: &mut dyn Read, _: &Context<Data>) -> Result<Vec<Self>> + fn decode_slice(buf: &mut &[u8], _: &Context<Data>) -> Result<Vec<Self>> where Self: Sized, { - let mut buf = Vec::new(); - reader.read_to_end(&mut buf)?; - Ok(buf) + const BYTES: usize = std::mem::size_of::<$t>(); + let count = buf.len() / BYTES; + let mut vec = Vec::with_capacity(count); + for _ in 0..count { + let (bytes, rest) = buf.split_at(BYTES); + *buf = rest; + vec.push(<$t>::from_be_bytes(bytes.try_into().unwrap())); + } + Ok(vec) } } }; - ($type: ty) => { - impl<Data> Decode<Data> for $type { - // decode number using big endian - fn decode(reader: &mut dyn Read, _: &Context<Data>) -> Result<Self> - where - Self: Sized, - { - const BYTES: usize = (<$type>::BITS / 8) as usize; - let mut bytes = [0; BYTES]; - reader.read_exact(&mut bytes)?; - Ok(<$type>::from_be_bytes(bytes)) - } +} - fn decode_slice(reader: &mut dyn Read, _: &Context<Data>) -> Result<Vec<Self>> +macro_rules! impl_decode_tuples { + ($($generic: ident),+) => { + impl<Data, $($generic),+> Decode<Data> for ($($generic,)+) + where + $($generic: Decode<Data>),+ + { + #[allow(non_snake_case)] + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let mut buf = Vec::with_capacity(DEFAULT_BUFFER_LEN); - reader.read_to_end(&mut buf); - const BYTES: usize = <$type>::BITS as usize / 8; - Ok(buf - .chunks(BYTES) - .map(|slice| { - let Some(bytes): Option<&[u8; BYTES]> = slice.as_array() else { - unreachable!() - }; - <$type>::from_be_bytes(*bytes) - }) - .collect()) + $( + let $generic = $generic::decode(buf, ctx)?; + )+ + Ok(($($generic,)+)) } } }; } -macro_rules! impl_decode_tuples { - ($($generic: ident),+) => { - impl<Data, $($generic),+> Decode<Data> for ($($generic,)+) where $($generic: Decode<Data>,)+ { - #[allow(non_snake_case)] - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> - where - Self: Sized { - $( - let $generic = $generic::decode(reader, ctx)?; - )+ - Ok(($($generic,)+)) - } - } - }; -} - -/// Read a value from a byte stream +/// Zero-copy decoding: reads directly from a byte slice pub trait Decode<Data> { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized; - // TODO : change for anything else than Vec - /// assume reader contains only the slice to decode - fn decode_slice(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Vec<Self>> + fn decode_slice(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Vec<Self>> where Self: Sized, { - let mut buf = Vec::new(); - loop { - match Self::decode(reader, ctx) { - Ok(value) => buf.push(value), + let mut vec = Vec::new(); + while !buf.is_empty() { + match Self::decode(buf, ctx) { + Ok(value) => vec.push(value), Err(CodecError::IoError(err)) - if let std::io::ErrorKind::UnexpectedEof = err.kind() => + if err.kind() == std::io::ErrorKind::UnexpectedEof => { - return Ok(buf); + return Ok(vec); } Err(err) => return Err(err), } } + Ok(vec) } } -// ~ Decode arrays +// ── Decode arrays ─────────────────────────────────────────────────────── impl<Data, T: Decode<Data>> Decode<Data> for Vec<T> { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let len = usize::decode(reader, ctx)?; + let len = usize::decode(buf, ctx)?; let mut vec = Vec::with_capacity(len); for _ in 0..len { - vec.push(T::decode(reader, ctx)?); + vec.push(T::decode(buf, ctx)?); } Ok(vec) } } impl<Data, T: Decode<Data>> Decode<Data> for Option<T> { - /// Assume is always some - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - T::decode(reader, ctx).map(Some) + if buf.is_empty() { + return Ok(None); + } + T::decode(buf, ctx).map(Some) } } -// ~ Decode slices +// ── Decode slices ─────────────────────────────────────────────────────── impl<Data, T, const S: usize> Decode<Data> for [T; S] where T: Decode<Data> + Default + Copy, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = T::decode_slice(reader, ctx)?; + let slice = T::decode_slice(buf, ctx)?; let got = slice.len(); slice .as_array() @@ -157,7 +140,7 @@ impl<Data, T> Decode<Data> for &[T] where T: Decode<Data> + Default + Copy, { - fn decode(_: &mut dyn Read, _: &Context<Data>) -> Result<Self> + fn decode(_: &mut &[u8], _: &Context<Data>) -> Result<Self> where Self: Sized, { @@ -165,18 +148,22 @@ where } } -// ~ Decode set +// ── Decode set ────────────────────────────────────────────────────────── impl<Data, T> Decode<Data> for HashSet<T> where T: Decode<Data> + Eq + Hash, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = T::decode_slice(reader, ctx)?; - Ok(slice.into_iter().collect()) + let len = usize::decode(buf, ctx)?; + let mut set = HashSet::with_capacity(len); + for _ in 0..len { + set.insert(T::decode(buf, ctx)?); + } + Ok(set) } } @@ -184,43 +171,57 @@ impl<Data, T> Decode<Data> for BTreeSet<T> where T: Decode<Data> + Ord, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = T::decode_slice(reader, ctx)?; - Ok(slice.into_iter().collect()) + let len = usize::decode(buf, ctx)?; + let mut set = BTreeSet::new(); + for _ in 0..len { + set.insert(T::decode(buf, ctx)?); + } + Ok(set) } } -// ~ Decode misc +// ── Decode misc ───────────────────────────────────────────────────────── impl<Data, T> Decode<Data> for BinaryHeap<T> where T: Decode<Data> + Ord, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = T::decode_slice(reader, ctx)?; - Ok(BinaryHeap::from_iter(slice)) + let len = usize::decode(buf, ctx)?; + let mut heap = BinaryHeap::with_capacity(len); + for _ in 0..len { + heap.push(T::decode(buf, ctx)?); + } + Ok(heap) } } -// ~ Decode maps +// ── Decode maps ───────────────────────────────────────────────────────── impl<Data, K, V> Decode<Data> for HashMap<K, V> where K: Decode<Data> + Eq + Hash, V: Decode<Data>, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = <(K, V) as Decode<Data>>::decode_slice(reader, ctx)?; - Ok(slice.into_iter().collect()) + let len = usize::decode(buf, ctx)?; + let mut map = HashMap::with_capacity(len); + for _ in 0..len { + let key = K::decode(buf, ctx)?; + let val = V::decode(buf, ctx)?; + map.insert(key, val); + } + Ok(map) } } @@ -229,19 +230,25 @@ where K: Decode<Data> + Ord, V: Decode<Data>, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = <(K, V) as Decode<Data>>::decode_slice(reader, ctx)?; - Ok(slice.into_iter().collect()) + let len = usize::decode(buf, ctx)?; + let mut map = BTreeMap::new(); + for _ in 0..len { + let key = K::decode(buf, ctx)?; + let val = V::decode(buf, ctx)?; + map.insert(key, val); + } + Ok(map) } } -// ~ Decode string +// ── Decode string - requires length prefix ────────────────────────────── impl<Data> Decode<Data> for &str { - fn decode(_: &mut dyn Read, _: &Context<Data>) -> Result<Self> + fn decode(_: &mut &[u8], _: &Context<Data>) -> Result<Self> where Self: Sized, { @@ -250,17 +257,17 @@ impl<Data> Decode<Data> for &str { } impl<Data> Decode<Data> for String { - fn decode(reader: &mut dyn Read, _: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let mut bytes: Vec<u8> = Vec::new(); - reader.read_to_end(&mut bytes)?; - Ok(String::from_utf8_lossy(&bytes).to_string()) + // String must be length-prefixed - decode as LenPrefixed<u16, String> + crate::types::prefix::length::LenPrefixed::<u16, String>::decode(buf, ctx) + .map(|p| p.data().clone()) } } -// ~ Decode primitive +// ── Decode primitive ──────────────────────────────────────────────────── impl_decode!(u8); impl_decode!(u16); @@ -277,39 +284,43 @@ impl_decode!(i128); impl_decode!(isize); impl<Data> Decode<Data> for f64 { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let bits = u64::decode(reader, ctx)?; - let value = f64::from_bits(bits); - Ok(value) + let bits = u64::decode(buf, ctx)?; + Ok(f64::from_bits(bits)) } } impl<Data> Decode<Data> for f32 { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let bits = u32::decode(reader, ctx)?; - let value = f32::from_bits(bits); - Ok(value) + let bits = u32::decode(buf, ctx)?; + Ok(f32::from_bits(bits)) } } impl<Data> Decode<Data> for bool { - fn decode(reader: &mut dyn Read, _: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], _: &Context<Data>) -> Result<Self> where Self: Sized, { - let mut byte = [0_u8]; - reader.read_exact(&mut byte)?; - Ok(byte[0] != 0) + if buf.is_empty() { + return Err(CodecError::IoError(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "insufficient bytes for bool", + ))); + } + let b = buf[0]; + *buf = &buf[1..]; + Ok(b != 0) } } -// ~ Decode tuple +// ── Decode tuple ──────────────────────────────────────────────────────── impl_decode_tuples!(A); impl_decode_tuples!(A, B); @@ -329,14 +340,14 @@ impl_decode_tuples!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O); impl_decode_tuples!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P); impl_decode_tuples!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P, Q); -// ~ Decode pointer +// ── Decode pointer ────────────────────────────────────────────────────── impl<Data, T> Decode<Data> for std::sync::Arc<T> where T: Decode<Data>, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> { - T::decode(reader, ctx).map(Arc::new) + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> { + T::decode(buf, ctx).map(Arc::new) } } @@ -344,11 +355,11 @@ impl<Data, T> Decode<Data> for Arc<[T]> where T: Decode<Data>, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> Result<Self> where Self: Sized, { - let slice = T::decode_slice(reader, ctx)?; + let slice = T::decode_slice(buf, ctx)?; Ok(slice.into()) } } diff --git a/src/codec/encode.rs b/src/codec/encode.rs index 7463995..735aa70 100644 --- a/src/codec/encode.rs +++ b/src/codec/encode.rs @@ -1,49 +1,36 @@ -//! Encoding: writing protocol values into a byte stream. use crate::codec::error::Result; use crate::context::Context; -use std::{ - collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet, LinkedList, VecDeque}, - io::Write, -}; +use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet, LinkedList, VecDeque}; macro_rules! impl_encode { - (u8) => { - impl<Data> Encode<Data> for u8 { - /// write number using big endian - fn encode(&self, buffer: &mut dyn Write, _: &Context<Data>) -> Result<usize> { - let size = buffer.write(&self.to_be_bytes())?; - Ok(size) + ($t:ty) => { + impl<Data> Encode<Data> for $t { + fn encode(&self, buffer: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> { + buffer.extend_from_slice(&self.to_be_bytes()); + Ok(std::mem::size_of::<$t>()) } - /// write raw bytes - fn encode_slice( - slice: &[Self], - buffer: &mut dyn Write, - _: &Context<Data>, - ) -> Result<usize> { - buffer.write(slice).map_err(Into::into) - } - } - }; - ($t: ty) => { - impl<Data> Encode<Data> for $t { - /// write number using big endian - fn encode(&self, buffer: &mut dyn Write, _: &Context<Data>) -> Result<usize> { - let size = buffer.write(&self.to_be_bytes())?; - Ok(size) + fn encode_slice(slice: &[Self], buffer: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> + where + Self: Sized, + { + let len = slice.len() * std::mem::size_of::<$t>(); + buffer.reserve(len); + for item in slice { + buffer.extend_from_slice(&item.to_be_bytes()); + } + Ok(len) } - fn encode_slice( - slice: &[Self], - buffer: &mut dyn Write, - _: &Context<Data>, - ) -> Result<usize> { - let len = buffer.write( - &slice - .iter() - .flat_map(|n| n.to_be_bytes()) - .collect::<Box<[u8]>>(), - )?; + fn encode_iter<'a, I>(iter: I, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> + where + Self: 'a + Sized, + I: IntoIterator<Item = &'a Self>, + { + let mut len = 0; + for item in iter { + len += item.encode(buffer, ctx)?; + } Ok(len) } } @@ -57,7 +44,7 @@ macro_rules! impl_encode_tuples { $($generic: Encode<Data>),+ { #[allow(non_snake_case)] - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { let ($($generic,)+): &($($generic,)+) = self; let mut len = 0; $(len += $generic.encode(buffer, ctx)?;)+ @@ -67,10 +54,11 @@ macro_rules! impl_encode_tuples { }; } -/// Write a value into a byte stream +/// Zero-copy encoding: writes directly into a byte buffer pub trait Encode<Data> { - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize>; - fn encode_slice(slice: &[Self], buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize>; + + fn encode_slice(slice: &[Self], buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> where Self: Sized, { @@ -80,204 +68,257 @@ pub trait Encode<Data> { } Ok(len) } + + fn encode_iter<'a, I>(iter: I, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> + where + Self: 'a + Sized, + I: IntoIterator<Item = &'a Self>, + { + let mut len = 0; + for item in iter { + len += item.encode(buffer, ctx)?; + } + Ok(len) + } } -// ~ Encode arrays +// ── Encode arrays ─────────────────────────────────────────────────────── impl<Data, T: Encode<Data>> Encode<Data> for Vec<T> { - /// Encode each element of Vec - /// use `CountPrefix` to prefix the vector with number of elements - /// use `LenPrefix` to prefix the vector with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { T::encode_slice(self, buffer, ctx) } } impl<Data, T: Encode<Data>> Encode<Data> for VecDeque<T> { - /// Encode each element of VecDeque - /// use `CountPrefix` to prefix the VecDeque with number of elements - /// use `LenPrefix` to prefix the VecDeque with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let (front, _) = self.as_slices(); - T::encode_slice(front, buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let (front, back) = self.as_slices(); + let mut len = 0; + len += T::encode_slice(front, buffer, ctx)?; + len += T::encode_slice(back, buffer, ctx)?; + Ok(len) } } -impl<Data, T: Encode<Data>> Encode<Data> for LinkedList<T> -where - T: Clone, -{ - /// Encode each element of LinkedList (use clone..) - /// use `CountPrefix` to prefix the LinkedList with number of elements - /// use `LenPrefix` to prefix the LinkedList with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let mut view = Vec::with_capacity(self.len()); - view.extend(self.iter().cloned()); - T::encode_slice(&view, buffer, ctx) +impl<Data, T: Encode<Data>> Encode<Data> for LinkedList<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let mut len = 0; + for item in self { + len += item.encode(buffer, ctx)?; + } + Ok(len) } } -// ~ Encode slices +// ── Encode slices ─────────────────────────────────────────────────────── impl<Data, T: Encode<Data>> Encode<Data> for &[T] { - /// Encode each element of slice (use clone..) - /// use `CountPrefix` to prefix the slice with number of elements - /// use `LenPrefix` to prefix the slice with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { T::encode_slice(self, buffer, ctx) } } impl<Data, T: Encode<Data>, const S: usize> Encode<Data> for [T; S] { - /// Encode each element of slice (use clone..) - /// use `CountPrefix` to prefix the slice with number of elements - /// use `LenPrefix` to prefix the slice with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { T::encode_slice(self, buffer, ctx) } } impl<Data, T: Encode<Data>> Encode<Data> for [T] { - /// Encode each element of slice (use clone..) - /// use `CountPrefix` to prefix the slice with number of elements - /// use `LenPrefix` to prefix the slice with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + T::encode_slice(self, buffer, ctx) + } +} + +// ── Encode references to collections ──────────────────────────────────── + +impl<Data, T: Encode<Data>> Encode<Data> for &Vec<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { T::encode_slice(self, buffer, ctx) } } -// ~ Encode set +impl<Data, T: Encode<Data>> Encode<Data> for &VecDeque<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let (front, back) = self.as_slices(); + let mut len = 0; + len += T::encode_slice(front, buffer, ctx)?; + len += T::encode_slice(back, buffer, ctx)?; + Ok(len) + } +} -impl<Data, T: Encode<Data>> Encode<Data> for HashSet<T> +impl<Data, T: Encode<Data>> Encode<Data> for &LinkedList<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let mut len = 0; + for item in (*self).iter() { + len += item.encode(buffer, ctx)?; + } + Ok(len) + } +} + +impl<Data, T: Encode<Data>> Encode<Data> for &HashSet<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + T::encode_iter(self.iter(), buffer, ctx) + } +} + +impl<Data, T: Encode<Data>> Encode<Data> for &BTreeSet<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + T::encode_iter(self.iter(), buffer, ctx) + } +} + +impl<Data, T: Encode<Data>> Encode<Data> for &BinaryHeap<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + T::encode_slice(self.as_slice(), buffer, ctx) + } +} + +impl<Data, K, V> Encode<Data> for &HashMap<K, V> where - T: Clone, + K: Encode<Data>, + V: Encode<Data>, { - /// Encode each element of HashSet (use clone..) - /// use `CountPrefix` to prefix the HashSet with number of elements - /// use `LenPrefix` to prefix the HashSet with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let mut view = Vec::with_capacity(self.len()); - view.extend(self.iter().cloned()); - T::encode_slice(&view, buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let mut len = 0; + for (k, v) in (*self).iter() { + len += k.encode(buffer, ctx)?; + len += v.encode(buffer, ctx)?; + } + Ok(len) } } -impl<Data, T: Encode<Data>> Encode<Data> for BTreeSet<T> +impl<Data, K, V> Encode<Data> for &BTreeMap<K, V> where - T: Clone, + K: Encode<Data>, + V: Encode<Data>, { - /// Encode each element of BTreeSet (use clone..) - /// use `CountPrefix` to prefix the BTreeSet with number of elements - /// use `LenPrefix` to prefix the BTreeSet with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let mut view = Vec::with_capacity(self.len()); - view.extend(self.iter().cloned()); - T::encode_slice(&view, buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let mut len = 0; + for (k, v) in (*self).iter() { + len += k.encode(buffer, ctx)?; + len += v.encode(buffer, ctx)?; + } + Ok(len) } } -// ~ Encode misc +// ── Encode set (owned) ────────────────────────────────────────────────── + +impl<Data, T: Encode<Data>> Encode<Data> for HashSet<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + T::encode_iter(self.iter(), buffer, ctx) + } +} + +impl<Data, T: Encode<Data>> Encode<Data> for BTreeSet<T> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + T::encode_iter(self.iter(), buffer, ctx) + } +} + +// ── Encode misc ───────────────────────────────────────────────────────── impl<Data, T> Encode<Data> for BinaryHeap<T> where T: Encode<Data>, { - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { T::encode_slice(self.as_slice(), buffer, ctx) } } -// ~ Encode maps +// ── Encode maps (owned) ───────────────────────────────────────────────── impl<Data, K, V> Encode<Data> for HashMap<K, V> where - K: Encode<Data> + Clone, - V: Encode<Data> + Clone, + K: Encode<Data>, + V: Encode<Data>, { - /// Encode each element of HashMap (use clone..) - /// use `CountPrefix` to prefix the HashMap with number of pairs - /// use `LenPrefix` to prefix the HashMap with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let slice: Vec<(K, V)> = self.clone().into_iter().collect(); - <(K, V) as Encode<Data>>::encode_slice(&slice, buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let mut len = 0; + for (k, v) in self { + len += k.encode(buffer, ctx)?; + len += v.encode(buffer, ctx)?; + } + Ok(len) } } impl<Data, K, V> Encode<Data> for BTreeMap<K, V> where - K: Encode<Data> + Clone, - V: Encode<Data> + Clone, + K: Encode<Data>, + V: Encode<Data>, { - /// Encode each element of BTreeMap (use clone..) - /// use `CountPrefix` to prefix the BTreeMap with number of pairs - /// use `LenPrefix` to prefix the BTreeMap with encoded byte size - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let slice: Vec<(K, V)> = self.clone().into_iter().collect(); - <(K, V) as Encode<Data>>::encode_slice(&slice, buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + let mut len = 0; + for (k, v) in self { + len += k.encode(buffer, ctx)?; + len += v.encode(buffer, ctx)?; + } + Ok(len) } } -// ~ Encode string +// ── Encode string ─────────────────────────────────────────────────────── + impl<Data> Encode<Data> for String { - /// Encode the string using utf8 - /// use `CountPrefix` to prefix the string with char length - /// use `LenPrefix` to prefix the string with utf8 bytes length - fn encode(&self, buffer: &mut dyn Write, _: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> { let utf8 = self.as_bytes(); - buffer.write_all(utf8)?; + buffer.extend_from_slice(utf8); Ok(utf8.len()) } } impl<Data> Encode<Data> for &str { - /// Encode the string using utf8 - /// use `CountPrefix` to prefix the string with char length - /// use `LenPrefix` to prefix the string with utf8 bytes length - fn encode(&self, buffer: &mut dyn Write, _: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> { let utf8 = self.as_bytes(); - buffer.write_all(utf8)?; + buffer.extend_from_slice(utf8); Ok(utf8.len()) } } impl<Data, T: Encode<Data>> Encode<Data> for Option<T> { - /// Encode `T` if Some(T) or do nothing if None - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { match self { - Some(val) => { - let size = val.encode(buffer, ctx)?; - Ok(size) - } + Some(val) => val.encode(buffer, ctx), None => Ok(0), } } } -// ~ Encode primitive +// ── Encode primitive ──────────────────────────────────────────────────── + impl<Data> Encode<Data> for bool { - fn encode(&self, buffer: &mut dyn Write, _: &Context<Data>) -> Result<usize> { - let size = buffer.write(&[*self as u8])?; - Ok(size) + fn encode(&self, buffer: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> { + buffer.push(*self as u8); + Ok(1) } - fn encode_slice(slice: &[Self], buffer: &mut dyn Write, _: &Context<Data>) -> Result<usize> + fn encode_slice(slice: &[Self], buffer: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> where Self: Sized, { - let len = buffer.write(&slice.iter().map(|n| *n as u8).collect::<Box<[u8]>>())?; + let len = slice.len(); + buffer.reserve(len); + for &b in slice { + buffer.push(b as u8); + } Ok(len) } } -// ~ Encode pointer +// ── Encode pointer ────────────────────────────────────────────────────── impl<Data, T> Encode<Data> for std::sync::Arc<T> where T: Encode<Data>, { - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let data = self.as_ref(); - data.encode(buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + self.as_ref().encode(buffer, ctx) } } @@ -285,9 +326,8 @@ impl<Data, T> Encode<Data> for std::sync::Arc<[T]> where T: Encode<Data>, { - fn encode(&self, buffer: &mut dyn Write, ctx: &Context<Data>) -> Result<usize> { - let data = self.as_ref(); - data.encode(buffer, ctx) + fn encode(&self, buffer: &mut Vec<u8>, ctx: &Context<Data>) -> Result<usize> { + self.as_ref().encode(buffer, ctx) } } @@ -312,14 +352,14 @@ impl_encode!(f64); #[cfg(feature = "f128")] impl_encode!(f128); -// ~ Encode tuple +// ── Encode tuple ──────────────────────────────────────────────────────── impl<Data> Encode<Data> for () { - fn encode(&self, _: &mut dyn Write, _: &Context<Data>) -> Result<usize> { + fn encode(&self, _: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> { Ok(0) } - fn encode_slice(_: &[Self], _: &mut dyn Write, _: &Context<Data>) -> Result<usize> + fn encode_slice(_: &[Self], _: &mut Vec<u8>, _: &Context<Data>) -> Result<usize> where Self: Sized, { @@ -8,11 +8,8 @@ pub mod event; pub mod transport; pub mod types; -#[cfg(feature = "macros")] -pub use zr_protocol_macros as macros; +pub use zr_protocol_macros::Codec; +pub use zr_protocol_macros::Decode; +pub use zr_protocol_macros::Encode; -/// The default buffer length used to allocate chunks -pub(crate) const DEFAULT_BUFFER_LEN: usize = 2048; - -// TODO : rewrite tests -// TODO : benchmark with some client / server example +pub use types::prefix::{CountPrefix, LenPrefixed}; diff --git a/src/transport/receiver.rs b/src/transport/receiver.rs index 3a7c948..a165110 100644 --- a/src/transport/receiver.rs +++ b/src/transport/receiver.rs @@ -3,3 +3,16 @@ use crate::{codec::Codec, context::Context, transport::Result}; pub trait PacketReceiver<Data, Uid: PartialEq>: Send + Sync { fn recv<P: Codec<Data>>(&self, ctx: &Context<Data>) -> Result<(Uid, P)>; } + +pub trait PacketReceiverBuf<Data, Uid: PartialEq>: Send + Sync { + fn recv_buf(&self, ctx: &Context<Data>) -> Result<(Uid, Vec<u8>)>; +} + +impl<Data, Uid: PartialEq, R: PacketReceiverBuf<Data, Uid>> PacketReceiver<Data, Uid> for R { + fn recv<P: Codec<Data>>(&self, ctx: &Context<Data>) -> Result<(Uid, P)> { + let (uid, bytes) = self.recv_buf(ctx)?; + let mut reader = bytes.as_slice(); + let packet = P::decode(&mut reader, ctx)?; + Ok((uid, packet)) + } +}
\ No newline at end of file diff --git a/src/transport/sender.rs b/src/transport/sender.rs index 804914f..3c83c6e 100644 --- a/src/transport/sender.rs +++ b/src/transport/sender.rs @@ -3,3 +3,15 @@ use crate::{codec::Codec, context::Context, transport::Result}; pub trait PacketSender<Data>: Send + Sync { fn send<P: Codec<Data>>(&self, packet: P, ctx: &Context<Data>) -> Result<()>; } + +pub trait PacketSenderBuf<Data>: Send + Sync { + fn send_buf(&self, buf: &[u8]) -> Result<()>; +} + +impl<Data, S: PacketSenderBuf<Data>> PacketSender<Data> for S { + fn send<P: Codec<Data>>(&self, packet: P, ctx: &Context<Data>) -> Result<()> { + let mut buf = Vec::new(); + packet.encode(&mut buf, ctx)?; + self.send_buf(&buf) + } +}
\ No newline at end of file diff --git a/src/transport/tcp.rs b/src/transport/tcp.rs index 19d02d6..12ecaa0 100644 --- a/src/transport/tcp.rs +++ b/src/transport/tcp.rs @@ -1,35 +1,30 @@ use std::hash::Hash; -use std::io::Write; +use std::io::{Read, Write}; use std::marker::PhantomData; use std::net::{SocketAddr, TcpListener as StdTcpListener, TcpStream}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; -use crate::codec::Codec; use crate::context::Context; use crate::transport::connection::Connection; use crate::transport::error::TransportError; use crate::transport::listener::Listener; -use crate::transport::receiver::PacketReceiver; -use crate::transport::sender::PacketSender; +use crate::transport::receiver::PacketReceiverBuf; +use crate::transport::sender::PacketSenderBuf; use crate::transport::{Result, Transport}; -// TODO : check Tcp Transport - #[derive(Clone)] pub struct TcpSender { stream: Arc<Mutex<TcpStream>>, } -impl<Data> PacketSender<Data> for TcpSender { - fn send<P: Codec<Data>>(&self, packet: P, ctx: &Context<Data>) -> Result<()> { - let mut buf = Vec::new(); - packet.encode(&mut buf, ctx)?; +impl<Data> PacketSenderBuf<Data> for TcpSender { + fn send_buf(&self, buf: &[u8]) -> Result<()> { let mut guard = self .stream .lock() .map_err(|_| TransportError::LockPoisoned)?; - guard.write_all(&buf)?; + guard.write_all(buf)?; guard.flush()?; Ok(()) } @@ -40,17 +35,24 @@ pub struct TcpReceiver<Uid> { uid: Uid, } -impl<Data, Uid> PacketReceiver<Data, Uid> for TcpReceiver<Uid> +impl<Data, Uid> PacketReceiverBuf<Data, Uid> for TcpReceiver<Uid> where Uid: PartialEq + Clone + Send + Sync, { - fn recv<P: Codec<Data>>(&self, ctx: &Context<Data>) -> Result<(Uid, P)> { + fn recv_buf(&self, _ctx: &Context<Data>) -> Result<(Uid, Vec<u8>)> { let mut guard = self .stream .lock() .map_err(|_| TransportError::LockPoisoned)?; - let packet = P::decode(&mut *guard, ctx)?; - Ok((self.uid.clone(), packet)) + + // Read 4-byte big-endian length prefix + let mut len_buf = [0u8; 4]; + guard.read_exact(&mut len_buf)?; + let len = u32::from_be_bytes(len_buf) as usize; + + let mut buf = vec![0u8; len]; + guard.read_exact(&mut buf)?; + Ok((self.uid.clone(), buf)) } } diff --git a/src/transport/udp.rs b/src/transport/udp.rs index acb36f2..9b312cf 100644 --- a/src/transport/udp.rs +++ b/src/transport/udp.rs @@ -10,11 +10,11 @@ use crate::context::Context; use crate::transport::Result; use crate::transport::connection::Connection; use crate::transport::listener::Listener; -use crate::transport::receiver::PacketReceiver; -use crate::transport::sender::PacketSender; +use crate::transport::receiver::PacketReceiverBuf; +use crate::transport::sender::PacketSenderBuf; use crate::transport::{Transport, TransportError}; -// TODO : check Udp Transport +const UDP_BUF_SIZE: usize = 1500; #[derive(Clone)] pub struct UdpPeerSender { @@ -22,11 +22,9 @@ pub struct UdpPeerSender { peer: SocketAddr, } -impl<Data> PacketSender<Data> for UdpPeerSender { - fn send<P: crate::codec::Codec<Data>>(&self, packet: P, ctx: &Context<Data>) -> Result<()> { - let mut buf = Vec::new(); - packet.encode(&mut buf, ctx)?; - self.sock.send_to(&buf, self.peer)?; +impl<Data> PacketSenderBuf<Data> for UdpPeerSender { + fn send_buf(&self, buf: &[u8]) -> Result<()> { + self.sock.send_to(buf, self.peer)?; Ok(()) } } @@ -57,11 +55,11 @@ impl<Uid> UdpReceiver<Uid> { } } -impl<Data, Uid> PacketReceiver<Data, Uid> for UdpReceiver<Uid> +impl<Data, Uid> PacketReceiverBuf<Data, Uid> for UdpReceiver<Uid> where Uid: PartialEq + Clone + Send + Sync, { - fn recv<P: crate::codec::Codec<Data>>(&self, ctx: &Context<Data>) -> Result<(Uid, P)> { + fn recv_buf(&self, _ctx: &Context<Data>) -> Result<(Uid, Vec<u8>)> { let bytes = match &self.inner { UdpReceiverInner::Channel(rx) => rx .lock() @@ -69,15 +67,13 @@ where .recv() .map_err(|_| TransportError::ChannelClosed)?, UdpReceiverInner::Socket(sock) => { - let mut buf = vec![0u8; 65535]; + let mut buf = vec![0u8; UDP_BUF_SIZE]; let (n, _) = sock.recv_from(&mut buf)?; buf.truncate(n); buf } }; - let mut reader = bytes.as_slice(); - let packet = P::decode(&mut reader, ctx)?; - Ok((self.uid.clone(), packet)) + Ok((self.uid.clone(), bytes)) } } @@ -100,7 +96,7 @@ where fn accept(&mut self) -> Result<Connection<Uid, Data, UdpPeerSender, UdpReceiver<Uid>>> { loop { - let mut buf = vec![0u8; 65535]; + let mut buf = vec![0u8; UDP_BUF_SIZE]; let (n, src) = self.sock.recv_from(&mut buf)?; buf.truncate(n); diff --git a/src/types/prefix.rs b/src/types/prefix.rs index 67cf047..1f007cf 100644 --- a/src/types/prefix.rs +++ b/src/types/prefix.rs @@ -1,2 +1,5 @@ pub mod count; pub mod length; + +pub use count::CountPrefix; +pub use length::LenPrefixed; diff --git a/src/types/prefix/count.rs b/src/types/prefix/count.rs index 7099cec..04bc7c6 100644 --- a/src/types/prefix/count.rs +++ b/src/types/prefix/count.rs @@ -1,4 +1,4 @@ -use std::{io::Read, marker::PhantomData}; +use std::marker::PhantomData; use getset::Getters; @@ -68,20 +68,18 @@ where { fn encode( &self, - buffer: &mut dyn std::io::prelude::Write, + buffer: &mut Vec<u8>, ctx: &Context<Data>, - ) -> Result<usize, crate::codec::error::CodecError> + ) -> crate::codec::error::Result<usize> where Self: Sized, { - let vec = self.data.clone().into_iter().collect::<Vec<I>>(); + let vec: Vec<I> = self.data.clone().into_iter().collect(); let len: L = vec.len().try_into().map_err(|_| { crate::codec::error::CodecError::Custom("count exceeds prefix capacity".into()) })?; let mut l = len.encode(buffer, ctx)?; - for item in vec { - l += item.encode(buffer, ctx)?; - } + l += I::encode_slice(&vec, buffer, ctx)?; Ok(l) } } @@ -92,14 +90,17 @@ where L: Codec<Data> + Size, D: IntoIterator<Item = I> + FromIterator<I>, { - fn decode(reader: &mut dyn Read, ctx: &Context<Data>) -> crate::codec::error::Result<Self> + fn decode( + buf: &mut &[u8], + ctx: &Context<Data>, + ) -> crate::codec::error::Result<Self> where Self: Sized, { - let len = L::decode(reader, ctx)?.into_size(); + let len = L::decode(buf, ctx)?.into_size(); let mut data = Vec::with_capacity(len); for _ in 0..len { - let item = I::decode(reader, ctx)?; + let item = I::decode(buf, ctx)?; data.push(item); } Ok(Self { @@ -107,4 +108,4 @@ where _len: PhantomData, }) } -} +}
\ No newline at end of file diff --git a/src/types/prefix/length.rs b/src/types/prefix/length.rs index f30eea0..fe839d5 100644 --- a/src/types/prefix/length.rs +++ b/src/types/prefix/length.rs @@ -1,10 +1,9 @@ -use std::{io::Write, marker::PhantomData}; +use std::marker::PhantomData; use getset::Getters; use crate::{ - DEFAULT_BUFFER_LEN, - codec::{self, Codec, decode::Decode, encode::Encode}, + codec::{Codec, decode::Decode, encode::Encode}, context::Context, types::size::Size, }; @@ -56,14 +55,16 @@ where { fn encode( &self, - writer: &mut dyn Write, + buffer: &mut Vec<u8>, ctx: &Context<Data>, - ) -> Result<usize, codec::error::CodecError> { - let mut buf = Vec::with_capacity(DEFAULT_BUFFER_LEN); - self.data.encode(&mut buf, ctx)?; - let len = L::from_size(buf.len()); - let mut len = len.encode(writer, ctx)?; - len += writer.write(&buf)?; + ) -> crate::codec::error::Result<usize> { + let mut len = 0; + let mut inner_buf = Vec::with_capacity(64); + self.data.encode(&mut inner_buf, ctx)?; + let len_val = L::from_size(inner_buf.len()); + len += len_val.encode(buffer, ctx)?; + buffer.extend_from_slice(&inner_buf); + len += inner_buf.len(); Ok(len) } } @@ -73,17 +74,22 @@ where L: Codec<Data> + Size, D: Codec<Data>, { - fn decode( - reader: &mut dyn std::io::prelude::Read, - ctx: &Context<Data>, - ) -> crate::codec::error::Result<Self> + fn decode(buf: &mut &[u8], ctx: &Context<Data>) -> crate::codec::error::Result<Self> where Self: Sized, { - let len = L::decode(reader, ctx)?.into_size(); - let mut limited = vec![0_u8; len]; - reader.read_exact(&mut limited)?; - let data = D::decode(&mut &limited[..], ctx)?; + let len = L::decode(buf, ctx)?.into_size(); + if buf.len() < len { + return Err(crate::codec::error::CodecError::IoError( + std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "insufficient bytes for len-prefixed data", + ), + )); + } + let (mut data, rest) = buf.split_at(len); + *buf = rest; + let data = D::decode(&mut data, ctx)?; Ok(Self { data, _len: PhantomData, |
