spate_core/
deser.rs

1//! Deserialization contract: one borrowed payload in, 0..N records out,
2//! push-style.
3//!
4//! The [`RecFamily`] lifetime→type family lets a lifetime-parameterized
5//! record type cross generic and dyn boundaries (ADR-0013). Deserializers
6//! for borrowing record types implement `Deserializer<F>` for a family `F`
7//! whose `Rec<'buf>` borrows the payload buffer; owned record types use
8//! the provided [`Owned`] family.
9
10use crate::checkpoint::AckRef;
11use crate::error::DeserError;
12use crate::record::{Flow, RawPayload, Record};
13
14/// Maps a payload-buffer lifetime to the deserialized record type.
15///
16/// The family tag itself is `'static`, a type-level function that is never
17/// instantiated. Chains stay nameable and `Send` while the records they
18/// process borrow ephemeral buffers.
19pub trait RecFamily: 'static {
20    /// The record type produced for payloads borrowing `'buf`.
21    type Rec<'buf>: Send;
22}
23
24/// Family for record types that do not borrow (owned payloads).
25#[derive(Debug)]
26pub struct Owned<T>(std::marker::PhantomData<fn() -> T>);
27
28impl<T: Send + 'static> RecFamily for Owned<T> {
29    type Rec<'buf> = T;
30}
31
32/// Push-style receiver for deserialized records (0..N per payload).
33pub trait EmitRecord<'buf, T> {
34    /// Hand one record to the chain. A [`Flow::Blocked`] return means
35    /// downstream is full; the deserializer stops emitting and the driver
36    /// retries the batch from the current payload cursor.
37    fn emit(&mut self, rec: Record<T>) -> Flow;
38}
39
40/// One borrowed payload in, 0..N records out through `out`.
41///
42/// Dyn-compatible for a concrete family (the only generic on the method is
43/// a lifetime). Zero emissions is valid (tombstones, empty envelopes). An
44/// error is subject to the stage's `ErrorPolicy`, except
45/// [`DeserError::NotReady`], which holds the batch for a replay, and
46/// [`DeserError::Fatal`], which stops the pipeline.
47pub trait Deserializer<F: RecFamily>: Send {
48    /// Decode `raw`, emitting each resulting record with the payload's
49    /// metadata and a clone of `ack`.
50    fn deserialize<'buf>(
51        &mut self,
52        raw: &RawPayload<'buf>,
53        ack: &AckRef,
54        out: &mut dyn EmitRecord<'buf, F::Rec<'buf>>,
55    ) -> Result<(), DeserError>;
56}
57
58/// Passthrough deserializer: yields the raw bytes as owned `Vec<u8>`
59/// records. Useful for byte-oriented pipelines and tests.
60#[derive(Clone, Copy, Debug, Default)]
61pub struct BytesPassthrough;
62
63impl Deserializer<Owned<Vec<u8>>> for BytesPassthrough {
64    fn deserialize<'buf>(
65        &mut self,
66        raw: &RawPayload<'buf>,
67        ack: &AckRef,
68        out: &mut dyn EmitRecord<'buf, Vec<u8>>,
69    ) -> Result<(), DeserError> {
70        let _ = out.emit(Record {
71            payload: raw.bytes.to_vec(),
72            meta: raw.meta(),
73            ack: ack.clone(),
74        });
75        Ok(())
76    }
77}
78
79#[cfg(all(test, not(loom)))]
80mod tests {
81    use super::*;
82    use crate::record::PartitionId;
83
84    struct Sink(Vec<Vec<u8>>);
85    impl EmitRecord<'_, Vec<u8>> for Sink {
86        fn emit(&mut self, rec: Record<Vec<u8>>) -> Flow {
87            self.0.push(rec.payload);
88            Flow::Continue
89        }
90    }
91
92    #[test]
93    fn passthrough_emits_one_owned_record() {
94        let (ack, _rx) = AckRef::test_pair();
95        let raw = RawPayload {
96            bytes: b"abc",
97            key: None,
98            partition: PartitionId(0),
99            offset: 1,
100            timestamp_ms: 2,
101        };
102        let mut sink = Sink(Vec::new());
103        BytesPassthrough.deserialize(&raw, &ack, &mut sink).unwrap();
104        assert_eq!(sink.0, vec![b"abc".to_vec()]);
105    }
106}
spate_core/deser.rs

spate_core/
deser.rs
