#![allow(clippy::unwrap_used, reason = "tests")]
#![allow(clippy::expect_used, reason = "tests")]
#![allow(
unsafe_code,
reason = "test codec uses pointer cast to simulate zero-copy"
)]
use std::fmt;
use mnesis::*;
use mnesis_inmemory::InMemoryStore;
use mnesis_store::Repository;
use mnesis_store::Store;
use mnesis_store::{Decode, Encode, PersistedEnvelope};
struct U32Codec;
impl Encode<[u32]> for U32Codec {
type Error = std::io::Error;
fn encode(&self, event: &[u32]) -> Result<bytes::Bytes, Self::Error> {
let buf: Vec<u8> = event.iter().flat_map(|n| n.to_le_bytes()).collect();
Ok(bytes::Bytes::from(buf))
}
}
impl Decode<[u32]> for U32Codec {
type Output<'a> = &'a [u32];
type Error = std::io::Error;
fn decode<'a>(&'a self, env: &'a PersistedEnvelope) -> Result<&'a [u32], Self::Error> {
let payload = env.payload();
if !payload.len().is_multiple_of(4) {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"payload length not multiple of 4",
));
}
let (prefix, shorts, suffix) = unsafe { payload.align_to::<u32>() };
if !prefix.is_empty() || !suffix.is_empty() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"unaligned",
));
}
Ok(shorts)
}
}
#[test]
fn borrowing_codec_decode_borrows_from_payload() {
let codec = U32Codec;
let original: Vec<u32> = vec![1, 2, 3];
let encoded = codec.encode(&original).unwrap();
let env = PersistedEnvelope::for_decode("event", &encoded).unwrap();
let decoded = codec.decode(&env).unwrap();
assert_eq!(decoded, &[1, 2, 3]);
}
#[test]
fn borrowing_codec_decode_rejects_bad_payload() {
let codec = U32Codec;
let bad = vec![1, 2, 3];
let env = PersistedEnvelope::for_decode("event", &bad).unwrap();
assert!(codec.decode(&env).is_err());
}
#[test]
fn borrowing_codec_is_send_sync() {
fn assert_send_sync<T: Send + Sync + 'static>() {}
assert_send_sync::<U32Codec>();
}
#[derive(Debug, Clone, Copy, PartialEq)]
#[repr(C)]
struct CounterEvent {
delta: i32,
}
impl Message for CounterEvent {}
impl DomainEvent for CounterEvent {
fn name(&self) -> &'static str {
"CounterChanged"
}
}
#[derive(Default, Debug, Clone, PartialEq)]
struct CounterState {
value: i64,
}
impl AggregateState for CounterState {
type Event = CounterEvent;
fn initial() -> Self {
Self::default()
}
fn apply(mut self, event: &CounterEvent) -> Self {
self.value += i64::from(event.delta);
self
}
}
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
struct CounterId(String);
impl fmt::Display for CounterId {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
f.write_str(&self.0)
}
}
impl AsRef<[u8]> for CounterId {
fn as_ref(&self) -> &[u8] {
self.0.as_bytes()
}
}
#[derive(Debug, thiserror::Error)]
#[error("counter error")]
struct CounterError;
struct CounterAggregate;
impl Aggregate for CounterAggregate {
type State = CounterState;
type Error = CounterError;
type Id = CounterId;
}
struct CounterBorrowingCodec;
impl Encode<CounterEvent> for CounterBorrowingCodec {
type Error = std::io::Error;
fn encode(&self, event: &CounterEvent) -> Result<bytes::Bytes, Self::Error> {
Ok(bytes::Bytes::copy_from_slice(&event.delta.to_le_bytes()))
}
}
impl Decode<CounterEvent> for CounterBorrowingCodec {
type Output<'a> = &'a CounterEvent;
type Error = std::io::Error;
fn decode<'a>(&'a self, env: &'a PersistedEnvelope) -> Result<&'a CounterEvent, Self::Error> {
let payload = env.payload();
if payload.len() != std::mem::size_of::<CounterEvent>() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"wrong payload size",
));
}
let (prefix, events, suffix) = unsafe { payload.align_to::<CounterEvent>() };
if !prefix.is_empty() || suffix.is_empty() && events.is_empty() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"alignment or size mismatch",
));
}
Ok(&events[0])
}
}
#[tokio::test]
async fn zero_copy_save_and_load_roundtrip() {
let store = Store::new(InMemoryStore::new());
let es = store.repository().codec(CounterBorrowingCodec).build();
let mut agg = AggregateRoot::<CounterAggregate>::new(CounterId("ctr-1".into()));
let events = [CounterEvent { delta: 10 }, CounterEvent { delta: -3 }];
es.save(&mut agg, &events_from_slice(&events))
.await
.unwrap();
let loaded: AggregateRoot<CounterAggregate> = es.load(CounterId("ctr-1".into())).await.unwrap();
assert_eq!(loaded.state().value, 7);
assert_eq!(loaded.version(), Some(Version::new(2).unwrap()));
}
#[tokio::test]
async fn zero_copy_load_empty_stream() {
let store = Store::new(InMemoryStore::new());
let es = store.repository().codec(CounterBorrowingCodec).build();
let loaded: AggregateRoot<CounterAggregate> = es.load(CounterId("ctr-1".into())).await.unwrap();
assert_eq!(loaded.state().value, 0);
assert_eq!(loaded.version(), None);
}
#[tokio::test]
async fn zero_copy_multi_save_load() {
let store = Store::new(InMemoryStore::new());
let es = store.repository().codec(CounterBorrowingCodec).build();
let mut agg1 = AggregateRoot::<CounterAggregate>::new(CounterId("ctr-1".into()));
es.save(&mut agg1, &events_from_slice(&[CounterEvent { delta: 5 }]))
.await
.unwrap();
let mut agg2: AggregateRoot<CounterAggregate> =
es.load(CounterId("ctr-1".into())).await.unwrap();
es.save(&mut agg2, &events_from_slice(&[CounterEvent { delta: 3 }]))
.await
.unwrap();
let final_agg: AggregateRoot<CounterAggregate> =
es.load(CounterId("ctr-1".into())).await.unwrap();
assert_eq!(final_agg.state().value, 8);
assert_eq!(final_agg.version(), Some(Version::new(2).unwrap()));
}
fn events_from_slice<E: mnesis::DomainEvent + Clone>(slice: &[E]) -> mnesis::Events<E, 32> {
let (first, rest) = slice
.split_first()
.expect("save requires at least one event");
let mut events = mnesis::Events::new(first.clone());
for event in rest {
events.add(event.clone());
}
events
}