use crate::{
body::Body,
error::{DecodeError, DecodeErrorKind, EncodeError},
message::{Encode, Message, View},
sink::Sink,
};
use bytes::{Bytes, BytesMut};
const HEADER_LEN: usize = 2;
pub type Converter<T> = fn(T) -> anyhow::Result<T>;
pub trait Versioned: Sized + 'static {
type Latest: Body;
const LATEST: u16;
const UPGRADES: &'static [Converter<Self>] = &[];
const DOWNGRADES: &'static [Converter<Self>] = &[];
fn wrap_latest(latest: Self::Latest) -> Self;
fn unwrap_latest(self) -> anyhow::Result<Self::Latest>;
fn decode_version(version: u16, body: Bytes) -> Result<Self, DecodeError>;
fn encode_version<S: Sink>(&self, version: u16, out: &mut S) -> Result<(), EncodeError>;
fn decode_body(body: Bytes, version: u16) -> Result<Self::Latest, DecodeError> {
if version == Self::LATEST {
return Self::Latest::from_body(body);
}
check_decode_version::<Self>(version)?;
let mut value = Self::decode_version(version, body)?;
let skip = usize::from(version) - 1;
for upgrade in Self::UPGRADES.iter().skip(skip) {
value = upgrade(value).map_err(DecodeError::conversion)?;
}
value.unwrap_latest().map_err(DecodeError::conversion)
}
fn open_body(
body: Bytes,
version: u16,
) -> Result<View<<Self::Latest as Message>::Wire>, DecodeError>
where
Self::Latest: Message,
{
if version == Self::LATEST {
return Self::Latest::view(body);
}
let latest = Self::decode_body(body, version)?;
View::new(latest.encode_to_bytes())
}
fn encode_body<S: Sink>(
value: &Self::Latest,
version: u16,
out: &mut S,
) -> Result<(), EncodeError> {
if version == Self::LATEST {
value.put_body(out);
return Ok(());
}
downgrade::<Self>(value, version)?.encode_version(version, out)
}
fn decode(bytes: Bytes) -> Result<Self::Latest, DecodeError> {
let (version, body) = split_header(bytes)?;
Self::decode_body(body, version)
}
fn open(bytes: Bytes) -> Result<View<<Self::Latest as Message>::Wire>, DecodeError>
where
Self::Latest: Message,
{
let (version, body) = split_header(bytes)?;
Self::open_body(body, version)
}
fn seal<S: Sink>(value: &Self::Latest, version: u16, out: &mut S) -> Result<(), EncodeError> {
if version == Self::LATEST {
out.put_u16_le(version);
value.put_body(out);
return Ok(());
}
let converted = downgrade::<Self>(value, version)?;
out.put_u16_le(version);
converted.encode_version(version, out)
}
fn sealed_len(value: &Self::Latest) -> usize {
HEADER_LEN + value.body_len()
}
fn seal_to_vec(value: &Self::Latest, version: u16) -> Result<Vec<u8>, EncodeError> {
let mut out = Vec::with_capacity(Self::sealed_len(value));
Self::seal(value, version, &mut out)?;
Ok(out)
}
fn seal_to_bytes(value: &Self::Latest, version: u16) -> Result<Bytes, EncodeError> {
let mut out = BytesMut::with_capacity(Self::sealed_len(value));
Self::seal(value, version, &mut out)?;
Ok(out.freeze())
}
fn seal_to_slice(
value: &Self::Latest,
version: u16,
out: &mut [u8],
) -> Result<usize, EncodeError> {
if version != Self::LATEST {
let sealed = Self::seal_to_vec(value, version)?;
check_capacity(sealed.len(), out.len())?;
out[..sealed.len()].copy_from_slice(&sealed);
return Ok(sealed.len());
}
let needed = Self::sealed_len(value);
check_capacity(needed, out.len())?;
let mut cursor = &mut out[..needed];
Self::seal(value, version, &mut cursor)?;
Ok(needed)
}
fn deserialize_with_embedded_version(payload: &[u8]) -> anyhow::Result<Self::Latest> {
Self::decode(Bytes::copy_from_slice(payload)).map_err(decode_error)
}
fn serialize_with_embedded_version(
value: &Self::Latest,
version: u16,
) -> anyhow::Result<Vec<u8>> {
Self::seal_to_vec(value, version).map_err(encode_error)
}
fn deserialize(payload: &[u8], version: u16) -> anyhow::Result<Self::Latest> {
Self::decode_body(Bytes::copy_from_slice(payload), version).map_err(decode_error)
}
fn serialize(value: &Self::Latest, version: u16) -> anyhow::Result<Vec<u8>> {
let mut out = Vec::with_capacity(value.body_len());
Self::encode_body(value, version, &mut out).map_err(encode_error)?;
Ok(out)
}
}
fn decode_error(error: DecodeError) -> anyhow::Error {
match error.into_conversion() {
Ok(conversion) => conversion,
Err(error) => error.into(),
}
}
fn encode_error(error: EncodeError) -> anyhow::Error {
match error {
EncodeError::Conversion(conversion) => conversion,
EncodeError::BufferTooSmall { .. } | EncodeError::UnsupportedVersion(_) => error.into(),
}
}
fn check_capacity(needed: usize, available: usize) -> Result<(), EncodeError> {
if needed > available {
return Err(EncodeError::BufferTooSmall { needed, available });
}
Ok(())
}
fn split_header(mut bytes: Bytes) -> Result<(u16, Bytes), DecodeError> {
if bytes.len() < HEADER_LEN {
return Err(DecodeError::with_detail(
DecodeErrorKind::Truncated,
bytes.len(),
"missing version header",
));
}
let body = bytes.split_off(HEADER_LEN);
let version = u16::from_le_bytes([bytes[0], bytes[1]]);
Ok((version, body))
}
fn check_decode_version<T: Versioned>(version: u16) -> Result<(), DecodeError> {
if version == 0 || version > T::LATEST {
return Err(DecodeError::with_detail(
DecodeErrorKind::UnsupportedVersion,
0,
format!("version {version}, latest is {}", T::LATEST),
));
}
Ok(())
}
fn downgrade<T: Versioned>(value: &T::Latest, version: u16) -> Result<T, EncodeError> {
if version == 0 || version > T::LATEST {
return Err(EncodeError::UnsupportedVersion(version));
}
let steps = usize::from(T::LATEST - version);
let mut converted = T::wrap_latest(value.clone());
for downgrade in T::DOWNGRADES.iter().take(steps) {
converted = downgrade(converted).map_err(EncodeError::Conversion)?;
}
Ok(converted)
}