krabka_client_streams/processor/
erased.rs

1//! Type-erased records, plus the per-dispatch context that the graph driver
2//! hands to each node.
3//!
4//! Records flow between nodes erased, as `Box<dyn Any + Send>`. Only the source
5//! and sink boundaries serialize and deserialize.
6
7use std::{any::Any, collections::VecDeque};
8
9use bytes::Bytes;
10
11use super::record::RecordContext;
12
13/// An error raised while a node processes a record, such as an internal downcast
14/// mismatch. This is unreachable in practice, because `build()` validates the
15/// wiring.
16#[derive(Debug, thiserror::Error)]
17pub enum ProcessorError {
18    #[error("type mismatch in node `{node}`: expected {expected}, found a different type")]
19    Downcast {
20        node: String,
21        expected: &'static str,
22    },
23    #[error("serialization error in sink `{node}`: {message}")]
24    Serde { node: String, message: String },
25    /// A subtopology id the built topology does not have.
26    #[error("the topology has no subtopology `{id}`")]
27    UnknownSubtopology { id: String },
28}
29
30/// A record with erased key/value, as it flows between nodes.
31pub(crate) struct ErasedRecord {
32    pub key: Option<Box<dyn Any + Send>>,
33    pub value: Box<dyn Any + Send>,
34    pub timestamp: i64,
35}
36
37impl ErasedRecord {
38    pub fn new(
39        key: Option<Box<dyn Any + Send>>,
40        value: Box<dyn Any + Send>,
41        timestamp: i64,
42    ) -> Self {
43        Self {
44            key,
45            value,
46            timestamp,
47        }
48    }
49}
50
51/// A record that a sink node emits and the driver collects. The test driver
52/// collects it into an output queue, the runtime hands it to the producer, and
53/// an [`EmbeddedTask`](crate::EmbeddedTask) returns it to the embedder.
54#[derive(Debug, Clone, PartialEq, Eq)]
55pub struct OutputRecord {
56    /// The sink topic, which is a repartition topic for an internal sink.
57    pub topic: String,
58    /// The serialized key, or `None` for a record without one.
59    pub key: Option<Bytes>,
60    /// The serialized value, or `None` for a tombstone.
61    pub value: Option<Bytes>,
62    /// The record timestamp in epoch milliseconds.
63    pub timestamp: i64,
64}
65
66/// What the driver lends to a node for one `process` call: the forward buffer
67/// for the source and processor children, this node's child indices, the sink
68/// output collector, the source-record context, and the per-task store registry
69/// that `get_state_store` reads.
70pub(crate) struct Dispatch<'a> {
71    pub buffer: &'a mut VecDeque<(usize, ErasedRecord)>,
72    pub children: &'a [usize],
73    pub output: &'a mut Vec<OutputRecord>,
74    pub record_ctx: &'a RecordContext,
75    pub stores: &'a mut crate::store::registry::StoreRegistry,
76    /// The app-wide, fully-replicated global stores, shared across tasks.
77    /// Stream-globaltable join processors read them with
78    /// `ProcessorContext::global_get`.
79    pub globals: &'a crate::runtime::global::GlobalStateManager,
80    /// The graph node this dispatch is positioned at. `schedule` therefore tags
81    /// the owning node, and a punctuator forwards to this node's children.
82    pub node_idx: usize,
83    /// Sink for `ProcessorContext::schedule`: the newly-registered punctuation
84    /// schedules.
85    pub schedules: &'a mut Vec<crate::processor::punctuation::ScheduleEntry>,
86    /// The current stream-time or wall-clock time. This is the BASE that
87    /// `schedule` stamps a new entry's first-fire time from, as
88    /// `base + interval`.
89    pub sched_stream_time: i64,
90    pub sched_wall_clock: i64,
91}
92
93#[cfg(test)]
94mod tests {
95    use assert2::check;
96
97    use super::*;
98
99    #[test]
100    fn erase_then_downcast_roundtrips_value_and_key() {
101        let er = ErasedRecord::new(Some(Box::new(7i32)), Box::new("v".to_string()), 1);
102        let key = er.key.unwrap().downcast::<i32>().unwrap();
103        let val = er.value.downcast::<String>().unwrap();
104        check!(*key == 7);
105        check!(*val == "v");
106    }
107}
krabka_client_streams/processor/erased.rs

krabka_client_streams/processor/
erased.rs
