use crate::config::{MAGIC_BYTES, MAX_PAYLOAD_SIZE, PROTOCOL_VERSION};
use crate::core::packet::{Packet, HEADER_SIZE};
use crate::error::{ProtocolError, Result};
use bytes::{BufMut, BytesMut};
use tokio_util::codec::{Decoder, Encoder};
pub struct PacketCodec;
impl Decoder for PacketCodec {
type Item = Packet;
type Error = ProtocolError;
fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Packet>> {
if src.len() < HEADER_SIZE {
return Ok(None);
}
if src[0..4] != MAGIC_BYTES {
return Err(ProtocolError::InvalidHeader);
}
if src[4] != PROTOCOL_VERSION {
return Err(ProtocolError::UnsupportedVersion(src[4]));
}
let len = u32::from_be_bytes([src[5], src[6], src[7], src[8]]) as usize;
if len > MAX_PAYLOAD_SIZE {
return Err(ProtocolError::OversizedPacket(len));
}
let total_len = HEADER_SIZE + len;
if src.len() < total_len {
return Ok(None); }
let buf = src.split_to(total_len).freeze();
Packet::from_bytes(&buf).map(Some)
}
}
impl Encoder<Packet> for PacketCodec {
type Error = ProtocolError;
fn encode(&mut self, packet: Packet, dst: &mut BytesMut) -> Result<()> {
if packet.payload.len() > MAX_PAYLOAD_SIZE {
return Err(ProtocolError::OversizedPacket(packet.payload.len()));
}
let total_size = HEADER_SIZE + packet.payload.len();
dst.reserve(total_size);
dst.put_slice(&MAGIC_BYTES);
dst.put_u8(PROTOCOL_VERSION);
dst.put_u32(packet.payload.len() as u32);
dst.put_slice(&packet.payload);
Ok(())
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
fn header(magic: [u8; 4], version: u8, len: u32) -> BytesMut {
let mut buf = BytesMut::new();
buf.put_slice(&magic);
buf.put_u8(version);
buf.put_u32(len);
buf
}
#[test]
fn oversized_length_is_rejected_from_the_header_alone() {
let mut buf = header(MAGIC_BYTES, PROTOCOL_VERSION, (MAX_PAYLOAD_SIZE + 1) as u32);
assert!(matches!(
PacketCodec.decode(&mut buf),
Err(ProtocolError::OversizedPacket(_))
));
let mut buf = header(MAGIC_BYTES, PROTOCOL_VERSION, u32::MAX);
assert!(matches!(
PacketCodec.decode(&mut buf),
Err(ProtocolError::OversizedPacket(_))
));
}
#[test]
fn bad_magic_or_version_is_rejected_from_the_header_alone() {
let mut buf = header(*b"HTTP", PROTOCOL_VERSION, 4);
assert!(matches!(
PacketCodec.decode(&mut buf),
Err(ProtocolError::InvalidHeader)
));
let mut buf = header(MAGIC_BYTES, PROTOCOL_VERSION.wrapping_add(1), 4);
assert!(matches!(
PacketCodec.decode(&mut buf),
Err(ProtocolError::UnsupportedVersion(_))
));
}
#[test]
fn valid_header_waits_for_the_payload() {
let mut buf = header(MAGIC_BYTES, PROTOCOL_VERSION, 4);
assert!(PacketCodec.decode(&mut buf).unwrap().is_none());
buf.put_slice(b"ping");
let packet = PacketCodec.decode(&mut buf).unwrap().unwrap();
assert_eq!(packet.payload, b"ping");
assert!(buf.is_empty());
}
#[test]
fn largest_allowed_payload_round_trips_and_larger_is_not_sent() {
let mut buf = BytesMut::new();
let packet = Packet {
version: PROTOCOL_VERSION,
payload: vec![7u8; MAX_PAYLOAD_SIZE],
};
PacketCodec.encode(packet, &mut buf).unwrap();
let decoded = PacketCodec.decode(&mut buf).unwrap().unwrap();
assert_eq!(decoded.payload.len(), MAX_PAYLOAD_SIZE);
let mut buf = BytesMut::new();
let too_big = Packet {
version: PROTOCOL_VERSION,
payload: vec![0u8; MAX_PAYLOAD_SIZE + 1],
};
assert!(matches!(
PacketCodec.encode(too_big, &mut buf),
Err(ProtocolError::OversizedPacket(_))
));
assert!(buf.is_empty(), "nothing is written for a rejected packet");
}
}