krabka_client_streams/dsl/
emit.rs

1//! `EmitStrategy` sets when a windowed aggregation forwards its results. It
2//! mirrors the JVM `org.apache.kafka.streams.kstream.EmitStrategy`.
3//! `on_window_update()` is the default and emits on every update.
4//! `on_window_close()` emits each window's final result once stream-time passes
5//! that window's close.
6//!
7//! The windowed handles carry it as a `Copy` field, and the lowering threads it
8//! into the aggregate processors. It changes ONLY the runtime forwarding
9//! behavior. The lowered topology, that is the node kind, the store
10//! registration, and the names, is identical for both strategies. This matches
11//! the JVM, which has one `KStreamWindowAggregate` class parameterized by
12//! `EmitStrategy`.
13
14/// When a windowed aggregation forwards its results downstream.
15#[derive(Clone, Copy, Debug, PartialEq, Eq)]
16pub struct EmitStrategy {
17    kind: EmitKind,
18}
19
20#[derive(Clone, Copy, Debug, PartialEq, Eq)]
21enum EmitKind {
22    OnWindowUpdate,
23    OnWindowClose,
24}
25
26impl EmitStrategy {
27    /// Emit on every update. This is the default.
28    #[must_use]
29    pub fn on_window_update() -> Self {
30        Self {
31            kind: EmitKind::OnWindowUpdate,
32        }
33    }
34
35    /// Emit each window's final result once stream-time passes its close.
36    #[must_use]
37    pub fn on_window_close() -> Self {
38        Self {
39            kind: EmitKind::OnWindowClose,
40        }
41    }
42
43    /// True for the emit-on-update strategy, which is the default. The aggregate
44    /// processors guard their per-update `ctx.forward` with this method.
45    pub(crate) fn is_on_update(self) -> bool {
46        matches!(self.kind, EmitKind::OnWindowUpdate)
47    }
48
49    /// True for the emit-on-close strategy.
50    pub(crate) fn is_on_close(self) -> bool {
51        matches!(self.kind, EmitKind::OnWindowClose)
52    }
53}
54
55impl Default for EmitStrategy {
56    fn default() -> Self {
57        Self::on_window_update()
58    }
59}
60
61#[cfg(test)]
62mod tests {
63    use super::*;
64
65    #[test]
66    fn defaults_to_on_update() {
67        assert!(EmitStrategy::default().is_on_update());
68        assert!(!EmitStrategy::default().is_on_close());
69    }
70
71    #[test]
72    fn on_window_close_is_close() {
73        let e = EmitStrategy::on_window_close();
74        assert!(e.is_on_close());
75        assert!(!e.is_on_update());
76    }
77}
krabka_client_streams/dsl/emit.rs

krabka_client_streams/dsl/
emit.rs
