use std::io::{self, Read, Write};
use crate::Schema;
pub type Len = u32;
pub const MAX_FRAME: usize = 256 << 20;
pub fn write_frame<T: Schema>(w: &mut impl Write, message: &T) -> io::Result<()> {
write_frame_bytes(w, &crate::to_bytes(message))
}
pub fn write_frame_bytes(w: &mut impl Write, bytes: &[u8]) -> io::Result<()> {
let len = frame_len(bytes.len())?;
let mut frame = Vec::with_capacity(size_of::<Len>() + bytes.len());
frame.extend_from_slice(&len.to_le_bytes());
frame.extend_from_slice(bytes);
w.write_all(&frame)?;
w.flush()
}
pub fn read_frame_bytes(r: &mut impl Read) -> io::Result<Option<Vec<u8>>> {
let Some(len) = read_len(r)? else {
return Ok(None);
};
let len = body_len(len)?;
let mut bytes = vec![0; len];
r.read_exact(&mut bytes)?;
Ok(Some(bytes))
}
pub fn read_frame<T: Schema>(r: &mut impl Read) -> io::Result<Option<T>> {
let Some(bytes) = read_frame_bytes(r)? else {
return Ok(None);
};
decode_frame(&bytes).map(Some)
}
pub(super) fn body_len(len: Len) -> io::Result<usize> {
let len = len as usize;
if len > MAX_FRAME {
return Err(io::Error::new(io::ErrorKind::InvalidData, "frame too long"));
}
Ok(len)
}
pub(super) fn frame_len(len: usize) -> io::Result<Len> {
Len::try_from(len)
.ok()
.filter(|_| len <= MAX_FRAME)
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "frame too long"))
}
pub(super) fn decode_frame<T: Schema>(bytes: &[u8]) -> io::Result<T> {
crate::read_bytes(bytes)
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, crate::Errors(e)))
}
pub(super) fn read_len(r: &mut impl Read) -> io::Result<Option<Len>> {
let mut len = [0; size_of::<Len>()];
loop {
match r.read(&mut len[..1]) {
Ok(0) => return Ok(None),
Ok(_) => break,
Err(e) if e.kind() == io::ErrorKind::Interrupted => {}
Err(e) => return Err(e),
}
}
r.read_exact(&mut len[1..])?;
Ok(Some(Len::from_le_bytes(len)))
}