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);
44/// errors are subject to the stage's `ErrorPolicy`.
45pub trait Deserializer<F: RecFamily>: Send {
46    /// Decode `raw`, emitting each resulting record with the payload's
47    /// metadata and a clone of `ack`.
48    fn deserialize<'buf>(
49        &mut self,
50        raw: &RawPayload<'buf>,
51        ack: &AckRef,
52        out: &mut dyn EmitRecord<'buf, F::Rec<'buf>>,
53    ) -> Result<(), DeserError>;
54}
55
56/// Passthrough deserializer: yields the raw bytes as owned `Vec<u8>`
57/// records. Useful for byte-oriented pipelines and tests.
58#[derive(Clone, Copy, Debug, Default)]
59pub struct BytesPassthrough;
60
61impl Deserializer<Owned<Vec<u8>>> for BytesPassthrough {
62    fn deserialize<'buf>(
63        &mut self,
64        raw: &RawPayload<'buf>,
65        ack: &AckRef,
66        out: &mut dyn EmitRecord<'buf, Vec<u8>>,
67    ) -> Result<(), DeserError> {
68        let _ = out.emit(Record {
69            payload: raw.bytes.to_vec(),
70            meta: raw.meta(),
71            ack: ack.clone(),
72        });
73        Ok(())
74    }
75}
76
77#[cfg(all(test, not(loom)))]
78mod tests {
79    use super::*;
80    use crate::record::PartitionId;
81
82    struct Sink(Vec<Vec<u8>>);
83    impl EmitRecord<'_, Vec<u8>> for Sink {
84        fn emit(&mut self, rec: Record<Vec<u8>>) -> Flow {
85            self.0.push(rec.payload);
86            Flow::Continue
87        }
88    }
89
90    #[test]
91    fn passthrough_emits_one_owned_record() {
92        let (ack, _rx) = AckRef::test_pair();
93        let raw = RawPayload {
94            bytes: b"abc",
95            key: None,
96            partition: PartitionId(0),
97            offset: 1,
98            timestamp_ms: 2,
99        };
100        let mut sink = Sink(Vec::new());
101        BytesPassthrough.deserialize(&raw, &ack, &mut sink).unwrap();
102        assert_eq!(sink.0, vec![b"abc".to_vec()]);
103    }
104}
spate_core/deser.rs

spate_core/
deser.rs
