use crate::envelope::PersistedEnvelope;
pub trait Encode<E: ?Sized>: Send + Sync + 'static {
type Error: core::error::Error + Send + Sync + 'static;
fn encode(&self, event: &E) -> Result<bytes::Bytes, Self::Error>;
}
pub trait Decode<E: ?Sized>: Send + Sync + 'static {
type Output<'a>
where
Self: 'a;
type Error: core::error::Error + Send + Sync + 'static;
fn decode<'a>(&'a self, env: &'a PersistedEnvelope) -> Result<Self::Output<'a>, Self::Error>;
}
pub trait OwningCodec<E: ?Sized>: for<'a> Decode<E, Output<'a> = E> {}
impl<C, E: ?Sized> OwningCodec<E> for C where C: for<'a> Decode<E, Output<'a> = E> {}
#[cfg(feature = "serde")]
pub mod serde {
use alloc::vec::Vec;
use ::serde::{Serialize, de::DeserializeOwned};
use super::{Decode, Encode};
use crate::envelope::PersistedEnvelope;
pub trait SerdeFormat: Send + Sync + 'static {
type Error: core::error::Error + Send + Sync + 'static;
fn serialize<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Self::Error>;
fn deserialize<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Self::Error>;
}
pub struct SerdeCodec<F> {
format: F,
}
impl<F> SerdeCodec<F> {
pub const fn new(format: F) -> Self {
Self { format }
}
}
impl<F: Default> Default for SerdeCodec<F> {
fn default() -> Self {
Self::new(F::default())
}
}
impl<E, F> Encode<E> for SerdeCodec<F>
where
E: Serialize + Send + Sync + 'static,
F: SerdeFormat,
{
type Error = F::Error;
fn encode(&self, event: &E) -> Result<bytes::Bytes, Self::Error> {
self.format.serialize(event).map(bytes::Bytes::from)
}
}
impl<E, F> Decode<E> for SerdeCodec<F>
where
E: DeserializeOwned + Send + Sync + 'static,
F: SerdeFormat,
{
type Output<'a>
= E
where
Self: 'a;
type Error = F::Error;
fn decode<'a>(
&'a self,
env: &'a PersistedEnvelope,
) -> Result<Self::Output<'a>, Self::Error> {
self.format.deserialize(env.payload())
}
}
impl<F> core::fmt::Debug for SerdeCodec<F> {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("SerdeCodec")
.field("format", &core::any::type_name::<F>())
.finish()
}
}
#[cfg(feature = "json")]
pub mod json {
use alloc::vec::Vec;
use ::serde::{Serialize, de::DeserializeOwned};
use super::{SerdeCodec, SerdeFormat};
#[derive(Debug, Clone, Copy, Default)]
pub struct Json;
impl SerdeFormat for Json {
type Error = serde_json::Error;
fn serialize<T: Serialize>(&self, value: &T) -> Result<Vec<u8>, Self::Error> {
serde_json::to_vec(value)
}
fn deserialize<T: DeserializeOwned>(&self, bytes: &[u8]) -> Result<T, Self::Error> {
serde_json::from_slice(bytes)
}
}
pub type JsonCodec = SerdeCodec<Json>;
}
}
#[cfg(feature = "bytemuck")]
pub mod bytemuck {
use ::bytemuck::{AnyBitPattern, NoUninit, PodCastError};
use bytes::Bytes;
use thiserror::Error;
use super::{Decode, Encode};
use crate::envelope::PersistedEnvelope;
#[derive(Debug, Error)]
#[error("bytemuck cast error: {0}")]
pub struct BytemuckError(arrayvec::ArrayString<64>);
impl From<PodCastError> for BytemuckError {
fn from(value: PodCastError) -> Self {
use core::fmt::Write;
let mut buf = arrayvec::ArrayString::<64>::new();
let _ = write!(buf, "{value:?}");
Self(buf)
}
}
#[derive(Debug, Default, Clone, Copy)]
pub struct BytemuckCodec;
impl<E> Encode<E> for BytemuckCodec
where
E: NoUninit + Send + Sync + 'static,
{
type Error = core::convert::Infallible;
fn encode(&self, event: &E) -> Result<Bytes, Self::Error> {
Ok(Bytes::copy_from_slice(::bytemuck::bytes_of(event)))
}
}
impl<E> Decode<E> for BytemuckCodec
where
E: AnyBitPattern + NoUninit + Send + Sync + 'static,
{
type Output<'a>
= &'a E
where
Self: 'a;
type Error = BytemuckError;
fn decode<'a>(&'a self, env: &'a PersistedEnvelope) -> Result<&'a E, Self::Error> {
::bytemuck::try_from_bytes(env.payload()).map_err(BytemuckError::from)
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::wire;
#[repr(C)]
#[derive(
Clone, Copy, Debug, PartialEq, ::bytemuck::AnyBitPattern, ::bytemuck::NoUninit,
)]
struct Pos {
x: f32,
y: f32,
z: f32,
_pad: f32,
}
fn build_test_envelope(payload: &[u8]) -> PersistedEnvelope {
let et = crate::value::EventType::from_static_str("Pos");
let pl = crate::value::Payload::from_bytes(bytes::Bytes::copy_from_slice(payload))
.expect("valid payload");
let sv = crate::value::SchemaVersion::INITIAL;
let frame = wire::encode_frame(sv, &et, &pl, None).expect("wire encode_frame ok");
PersistedEnvelope::try_new(
mnesis::Version::INITIAL,
frame.value,
sv,
frame.offsets.event_type,
frame.offsets.payload,
None,
)
.expect("envelope construction ok")
}
#[test]
fn round_trip_yields_equal_value() {
let codec = BytemuckCodec;
let original = Pos {
x: 1.0,
y: 2.0,
z: 3.0,
_pad: 0.0,
};
let bytes = codec.encode(&original).unwrap();
let env = build_test_envelope(&bytes);
let decoded: &Pos = codec.decode(&env).unwrap();
assert_eq!(decoded, &original);
}
#[test]
fn decode_borrows_from_envelope_payload() {
let codec = BytemuckCodec;
let original = Pos {
x: 1.0,
y: 2.0,
z: 3.0,
_pad: 0.0,
};
let bytes = codec.encode(&original).unwrap();
let env = build_test_envelope(&bytes);
let decoded: &Pos = codec.decode(&env).unwrap();
let env_payload_ptr = env.payload().as_ptr();
let decoded_ptr: *const u8 = std::ptr::from_ref::<Pos>(decoded).cast();
assert_eq!(
env_payload_ptr, decoded_ptr,
"BytemuckCodec must borrow from envelope payload, not copy"
);
}
#[test]
fn decode_rejects_wrong_size() {
let codec = BytemuckCodec;
let env = build_test_envelope(&[0u8; 8]);
let result: Result<&Pos, _> = codec.decode(&env);
let err = result
.copied()
.expect_err("wrong-size payload must be rejected");
assert!(
err.to_string().contains("SizeMismatch"),
"error must carry the PodCastError cause, got: {err}"
);
}
}
}
#[cfg(feature = "rkyv")]
pub mod rkyv {
use ::rkyv::{
Archive, Serialize,
api::high::{HighSerializer, HighValidator, to_bytes_in},
bytecheck::CheckBytes,
rancor,
ser::allocator::ArenaHandle,
util::AlignedVec,
};
use bytes::Bytes;
use super::{Decode, Encode};
use crate::envelope::PersistedEnvelope;
#[derive(Debug, Default, Clone, Copy)]
pub struct RkyvCodec;
impl<E> Encode<E> for RkyvCodec
where
E: for<'a> Serialize<HighSerializer<AlignedVec, ArenaHandle<'a>, rancor::Error>>
+ Send
+ Sync
+ 'static,
{
type Error = rancor::Error;
fn encode(&self, event: &E) -> Result<Bytes, Self::Error> {
let aligned = to_bytes_in::<_, rancor::Error>(event, AlignedVec::new())?;
Ok(Bytes::from(aligned.into_vec()))
}
}
impl<E> Decode<E> for RkyvCodec
where
E: Archive + Send + Sync + 'static,
E::Archived: for<'a> CheckBytes<HighValidator<'a, rancor::Error>>,
{
type Output<'a>
= &'a E::Archived
where
Self: 'a;
type Error = rancor::Error;
fn decode<'a>(
&'a self,
env: &'a PersistedEnvelope,
) -> Result<&'a E::Archived, Self::Error> {
::rkyv::access::<E::Archived, rancor::Error>(env.payload())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::wire;
#[derive(::rkyv::Archive, ::rkyv::Serialize, ::rkyv::Deserialize, Debug, PartialEq, Eq)]
struct Move {
steps: u32,
dir: u8,
}
fn build_test_envelope(payload: &[u8]) -> PersistedEnvelope {
let et = crate::value::EventType::from_static_str("Move");
let pl = crate::value::Payload::from_bytes(bytes::Bytes::copy_from_slice(payload))
.expect("valid payload");
let sv = crate::value::SchemaVersion::INITIAL;
let frame = wire::encode_frame(sv, &et, &pl, None).expect("wire encode_frame ok");
PersistedEnvelope::try_new(
mnesis::Version::INITIAL,
frame.value,
sv,
frame.offsets.event_type,
frame.offsets.payload,
None,
)
.expect("envelope construction ok")
}
#[test]
fn round_trip_yields_equal_archived_fields() {
let codec = RkyvCodec;
let original = Move { steps: 42, dir: 3 };
let bytes = codec.encode(&original).unwrap();
let env = build_test_envelope(&bytes);
let archived: &ArchivedMove =
<RkyvCodec as Decode<Move>>::decode(&codec, &env).unwrap();
assert_eq!(archived.steps, 42);
assert_eq!(archived.dir, 3);
}
}
}