krabka_client_streams/processor/
punctuation.rs

1//! Punctuation through `ProcessorContext::schedule` in the KIP Processor API.
2//! These are periodic callbacks that fire on a stream-time or wall-clock
3//! boundary.
4//!
5//! A `Punctuator` is a trait object erased to the driver exactly like a
6//! `Processor`. `TypedPunctuator` rebuilds the typed `ProcessorContext` from the
7//! `Dispatch`.
8//!
9//! The schedules live in the `Graph`, tagged by node index. The driver fires each
10//! one positioned at that node, so a punctuator's `forward` flows downstream.
11//! Punctuation is invisible in the wire topology and is purely a runtime
12//! feature.
13use std::{
14    any::Any,
15    marker::PhantomData,
16    sync::{
17        Arc,
18        atomic::{AtomicBool, Ordering},
19    },
20};
21
22use async_trait::async_trait;
23use krabka_units::prelude::*;
24
25use crate::processor::{api::ProcessorContext, erased::Dispatch};
26
27/// Which clock drives a punctuation schedule. This is the JVM
28/// `PunctuationType`.
29#[derive(Clone, Copy, Debug, PartialEq, Eq)]
30pub enum PunctuationType {
31    /// Driven by the task's observed max record timestamp.
32    StreamTime,
33    /// Driven by the system wall clock, or by a mock wall clock.
34    WallClockTime,
35}
36
37/// A periodic callback. You implement it on a user struct, as you do with
38/// [`Processor`]. It shares mutable state with its owning processor through an
39/// `Arc<Mutex<_>>`.
40///
41/// [`Processor`]: crate::processor::api::Processor
42#[async_trait]
43pub trait Punctuator<K: Send, V: Send>: Send + 'static {
44    /// Fire at `timestamp`. For a stream-time schedule this is the scheduled
45    /// time. For a wall-clock schedule it is the clock's current time. The
46    /// callback may `forward` through `ctx` and may use the state stores.
47    async fn punctuate(&mut self, ctx: &mut ProcessorContext<'_, '_, K, V>, timestamp: i64);
48}
49
50/// The handle that `ProcessorContext::schedule` returns. `cancel()` stops the
51/// schedule, and the driver drops it on the next punctuate pass.
52#[derive(Clone)]
53pub struct Cancellable(Arc<AtomicBool>);
54impl Cancellable {
55    pub(crate) fn new(flag: Arc<AtomicBool>) -> Self {
56        Self(flag)
57    }
58    /// Stop this schedule from firing again.
59    pub fn cancel(&self) {
60        self.0.store(true, Ordering::SeqCst);
61    }
62}
63
64/// An internal punctuator erased to the driver's untyped surface. It mirrors
65/// `ErasedNode`.
66#[async_trait]
67pub(crate) trait ErasedPunctuator: Send {
68    async fn fire(&mut self, dispatch: &mut Dispatch<'_>, timestamp: i64);
69}
70
71/// Wraps a typed [`Punctuator`] into an [`ErasedPunctuator`] by rebuilding the
72/// typed `ProcessorContext` from the `Dispatch`.
73pub(crate) struct TypedPunctuator<K, V, P> {
74    inner: P,
75    _pd: PhantomData<fn(K, V)>,
76}
77impl<K, V, P> TypedPunctuator<K, V, P> {
78    pub(crate) fn new(inner: P) -> Self {
79        Self {
80            inner,
81            _pd: PhantomData,
82        }
83    }
84}
85#[async_trait]
86impl<K, V, P> ErasedPunctuator for TypedPunctuator<K, V, P>
87where
88    K: Any + Send + Clone,
89    V: Any + Send + Clone,
90    P: Punctuator<K, V>,
91{
92    async fn fire(&mut self, dispatch: &mut Dispatch<'_>, timestamp: i64) {
93        let mut ctx = ProcessorContext::<'_, '_, K, V>::new(dispatch);
94        self.inner.punctuate(&mut ctx, timestamp).await;
95    }
96}
97
98/// One live punctuation schedule, owned by the `Graph`.
99pub(crate) struct ScheduleEntry {
100    pub node_idx: usize,
101    pub interval: Time,
102    pub ty: PunctuationType,
103    /// The next time to fire, as an instant on the evaluating clock's timeline.
104    ///
105    /// `schedule()` stamps it as `base + interval`, where `base` is the clock
106    /// value at the moment the schedule is registered.
107    pub next_time: i64,
108    pub punctuator: Box<dyn ErasedPunctuator>,
109    pub cancel: Arc<AtomicBool>,
110}
111impl ScheduleEntry {
112    pub fn is_cancelled(&self) -> bool {
113        self.cancel.load(Ordering::SeqCst)
114    }
115}
116
117#[cfg(test)]
118mod tests {
119    use assert2::check;
120
121    use super::*;
122
123    #[test]
124    fn cancellable_flips_the_flag() {
125        let flag = Arc::new(AtomicBool::new(false));
126        let c = Cancellable::new(flag.clone());
127        check!(!flag.load(Ordering::SeqCst));
128        c.cancel();
129        check!(flag.load(Ordering::SeqCst));
130    }
131
132    #[test]
133    fn schedule_entry_reports_cancelled() {
134        struct NoOp;
135        #[async_trait]
136        impl ErasedPunctuator for NoOp {
137            async fn fire(&mut self, _d: &mut Dispatch<'_>, _ts: i64) {}
138        }
139        let flag = Arc::new(AtomicBool::new(false));
140        let e = ScheduleEntry {
141            node_idx: 0,
142            interval: millis(10),
143            ty: PunctuationType::StreamTime,
144            next_time: 0,
145            punctuator: Box::new(NoOp),
146            cancel: flag.clone(),
147        };
148        check!(!e.is_cancelled());
149        flag.store(true, Ordering::SeqCst);
150        check!(e.is_cancelled());
151    }
152}
krabka_client_streams/processor/punctuation.rs

krabka_client_streams/processor/
punctuation.rs
