use crate::buf::WriteBuf;
use crate::error::{Error, Result};
use crate::framing::{Frame, Framer};
use crate::varint::{self, MAX_LEN_U64};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct VarintPrefixed {
max_payload: usize,
canonical: bool,
}
impl VarintPrefixed {
pub const DEFAULT_MAX_PAYLOAD: usize = 8 * 1024 * 1024;
#[inline]
#[must_use]
pub const fn new() -> Self {
Self {
max_payload: Self::DEFAULT_MAX_PAYLOAD,
canonical: true,
}
}
#[inline]
#[must_use]
pub const fn with_max_payload(mut self, max: usize) -> Self {
self.max_payload = max;
self
}
#[inline]
#[must_use]
pub const fn with_canonical(mut self, canonical: bool) -> Self {
self.canonical = canonical;
self
}
#[inline]
#[must_use]
pub const fn max_payload(&self) -> usize {
self.max_payload
}
#[inline]
#[must_use]
pub const fn is_canonical(&self) -> bool {
self.canonical
}
fn too_large(&self, len: u64) -> Error {
Error::FrameTooLarge {
len: usize::try_from(len).unwrap_or(usize::MAX),
limit: self.max_payload,
}
}
}
impl Default for VarintPrefixed {
#[inline]
fn default() -> Self {
Self::new()
}
}
impl Framer for VarintPrefixed {
fn next_frame<'a>(&self, input: &'a [u8]) -> Result<Option<Frame<'a>>> {
let (len, header_size) =
match varint::decode_slice(input, MAX_LEN_U64, u64::BITS, self.canonical) {
Ok(parsed) => parsed,
Err(Error::UnexpectedEof) => return Ok(None),
Err(e) => return Err(e),
};
let consumed = usize::try_from(len)
.ok()
.filter(|&len| len <= self.max_payload)
.and_then(|len| len.checked_add(header_size))
.ok_or_else(|| self.too_large(len))?;
let Some(payload) = input.get(header_size..consumed) else {
return Ok(None);
};
Ok(Some(Frame::new(payload, consumed)))
}
fn write_frame(&self, payload: &[u8], out: &mut WriteBuf<'_>) -> Result<()> {
if payload.len() > self.max_payload {
return Err(Error::FrameTooLarge {
len: payload.len(),
limit: self.max_payload,
});
}
let total = self.max_frame_len(payload.len()).ok_or(Error::BufferFull)?;
if total > out.remaining() {
return Err(Error::BufferFull);
}
varint::encode_usize(payload.len(), out)?;
out.write_bytes(payload)
}
#[inline]
fn max_frame_len(&self, payload_len: usize) -> Option<usize> {
payload_len.checked_add(varint::encoded_len_usize(payload_len))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn round_trip_across_prefix_sizes() {
let framer = VarintPrefixed::new();
let mut out = vec![0u8; 20_000];
for len in [0usize, 1, 127, 128, 16_383, 16_384] {
let payload = vec![0x5Au8; len];
let mut buf = WriteBuf::new(&mut out);
framer.write_frame(&payload, &mut buf).unwrap();
let n = buf.position();
assert_eq!(Some(n), framer.max_frame_len(len));
let frame = framer.next_frame(&out[..n]).unwrap().unwrap();
assert_eq!(frame.payload(), payload.as_slice());
assert_eq!(frame.consumed(), n);
for cut in 0..n {
assert_eq!(
framer.next_frame(&out[..cut]),
Ok(None),
"len={len} cut={cut}"
);
}
}
}
#[test]
fn limit_boundaries() {
let framer = VarintPrefixed::new().with_max_payload(3);
assert_eq!(
framer.next_frame(&[3, 1, 2, 3]).unwrap().unwrap().payload(),
&[1, 2, 3]
);
assert_eq!(
framer.next_frame(&[4]),
Err(Error::FrameTooLarge { len: 4, limit: 3 })
);
let mut out = [0u8; 8];
let mut buf = WriteBuf::new(&mut out);
assert_eq!(
framer.write_frame(&[0; 4], &mut buf),
Err(Error::FrameTooLarge { len: 4, limit: 3 })
);
assert_eq!(buf.position(), 0);
}
#[test]
fn hostile_prefixes() {
let framer = VarintPrefixed::new().with_max_payload(usize::MAX);
assert_eq!(framer.next_frame(&[0x80; 11]), Err(Error::VarintOverflow));
assert_eq!(framer.next_frame(&[0x80; 10]), Err(Error::VarintOverflow));
assert_eq!(framer.next_frame(&[0x80; 9]), Ok(None));
let max = [0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0x01];
assert!(matches!(
framer.next_frame(&max),
Err(Error::FrameTooLarge { .. })
));
}
#[test]
fn write_is_atomic_on_buffer_full() {
let framer = VarintPrefixed::new();
let mut out = [0xEEu8; 3];
let mut buf = WriteBuf::new(&mut out);
assert_eq!(framer.write_frame(b"abc", &mut buf), Err(Error::BufferFull));
assert_eq!(out, [0xEE; 3]);
}
#[test]
fn accessors() {
let framer = VarintPrefixed::new()
.with_max_payload(5)
.with_canonical(false);
assert_eq!(framer.max_payload(), 5);
assert!(!framer.is_canonical());
assert!(VarintPrefixed::new().is_canonical());
}
#[test]
fn default_matches_new() {
assert_eq!(VarintPrefixed::default(), VarintPrefixed::new());
assert_eq!(VarintPrefixed::new().max_frame_len(usize::MAX), None);
}
}