krabka_client_streams/dsl/
config.rs

1//! DSL config objects that mirror the JVM `Grouped`, `Materialized`, and
2//! `Repartitioned`.
3//!
4//! `Consumed` and `Produced` come from `crate::processor::serde`.
5
6use krabka_units::prelude::*;
7
8/// The serdes, and an optional repartition name, for `groupBy` and
9/// `groupByKey`.
10#[derive(Debug, Clone)]
11pub struct Grouped<KS, VS> {
12    #[allow(dead_code)]
13    pub(crate) key_serde: KS,
14    #[allow(dead_code)]
15    pub(crate) value_serde: VS,
16    pub(crate) name: Option<String>,
17}
18impl<KS, VS> Grouped<KS, VS> {
19    pub fn with(key_serde: KS, value_serde: VS) -> Self {
20        Self {
21            key_serde,
22            value_serde,
23            name: None,
24        }
25    }
26    #[must_use]
27    pub fn with_name(mut self, name: impl Into<String>) -> Self {
28        self.name = Some(name.into());
29        self
30    }
31}
32
33/// Versioned-store settings for a `Materialized` (KIP-889).
34///
35/// `segment_interval` affects only the JVM eviction granularity, which is not
36/// observable here. This type accepts it for API parity. A `None` segment
37/// interval uses the JVM default segment heuristic.
38#[derive(Debug, Clone, Copy)]
39pub struct VersionedConfig {
40    pub history_retention: Time,
41    pub segment_interval: Option<Time>,
42}
43
44/// The store name, the serdes, and the logging flag for a materialized
45/// `KTable`.
46#[derive(Debug, Clone)]
47pub struct Materialized<KS, VS> {
48    #[allow(dead_code)]
49    pub(crate) key_serde: KS,
50    #[allow(dead_code)]
51    pub(crate) value_serde: VS,
52    pub(crate) store_name: Option<String>,
53    pub(crate) logging: bool,
54    pub(crate) caching: bool,
55    pub(crate) versioned: Option<VersionedConfig>,
56}
57impl<KS, VS> Materialized<KS, VS> {
58    pub fn with(key_serde: KS, value_serde: VS) -> Self {
59        Self {
60            key_serde,
61            value_serde,
62            store_name: None,
63            logging: true,
64            caching: true,
65            versioned: None,
66        }
67    }
68    #[must_use]
69    pub fn as_store(mut self, name: impl Into<String>) -> Self {
70        self.store_name = Some(name.into());
71        self
72    }
73    #[must_use]
74    pub fn with_logging(mut self, on: bool) -> Self {
75        self.logging = on;
76        self
77    }
78    /// Turn the record cache for this store on or off. This is the JVM
79    /// `Materialized.withCachingEnabled` and
80    /// `Materialized.withCachingDisabled`. Default: on.
81    #[must_use]
82    pub fn with_caching(mut self, on: bool) -> Self {
83        self.caching = on;
84        self
85    }
86    #[must_use]
87    pub fn caching_enabled(&self) -> bool {
88        self.caching
89    }
90    /// Materialize this table into a versioned key-value store (KIP-889) named
91    /// `name`, retaining `history_retention` of version history.
92    #[must_use]
93    pub fn as_versioned(mut self, name: impl Into<String>, history_retention: Time) -> Self {
94        self.store_name = Some(name.into());
95        self.versioned = Some(VersionedConfig {
96            history_retention,
97            segment_interval: None,
98        });
99        self
100    }
101}
102
103impl<KS, VS> From<(KS, VS)> for Materialized<KS, VS> {
104    fn from((key_serde, value_serde): (KS, VS)) -> Self {
105        Self::with(key_serde, value_serde)
106    }
107}
108
109impl<KS, VS> From<(KS, VS)> for Grouped<KS, VS> {
110    fn from((key_serde, value_serde): (KS, VS)) -> Self {
111        Self::with(key_serde, value_serde)
112    }
113}
114
115impl<KS, VS> From<(KS, VS)> for Repartitioned<KS, VS> {
116    fn from((key_serde, value_serde): (KS, VS)) -> Self {
117        Self::with(key_serde, value_serde)
118    }
119}
120
121impl<KS, V1S, V2S> From<(KS, V1S, V2S)> for StreamJoined<KS, V1S, V2S> {
122    fn from((key_serde, value1_serde, value2_serde): (KS, V1S, V2S)) -> Self {
123        Self::with(key_serde, value1_serde, value2_serde)
124    }
125}
126
127/// The serdes for a windowed `KStream`-`KStream` join, the `StreamJoined`.
128///
129/// It holds the key serde and the per-side value serdes: `value1` for this side,
130/// the left side, and `value2` for the other side, the right side. The two
131/// join-window stores are registered with `key_serde` and the matching value
132/// serde, which mirrors the JVM `StreamJoined.with`.
133#[derive(Debug, Clone, Copy)]
134pub struct StreamJoined<KS, V1S, V2S> {
135    pub(crate) key: KS,
136    pub(crate) left_value: V1S,
137    pub(crate) right_value: V2S,
138}
139impl<KS, V1S, V2S> StreamJoined<KS, V1S, V2S> {
140    pub fn with(key_serde: KS, value1_serde: V1S, value2_serde: V2S) -> Self {
141        Self {
142            key: key_serde,
143            left_value: value1_serde,
144            right_value: value2_serde,
145        }
146    }
147}
148
149/// Stream-table join config (KIP-923).
150///
151/// It carries an optional grace period that buffers the stream records, so an
152/// out-of-order record still joins the table value as of its own timestamp.
153/// `grace` needs the joined table to be versioned, and it must be
154/// `< history_retention`. The join DSL asserts that at build time. `name`
155/// optionally names the grace buffer store.
156#[derive(Debug, Clone, Default)]
157pub struct Joined {
158    pub(crate) grace: Option<Time>,
159    pub(crate) name: Option<String>,
160}
161impl Joined {
162    #[must_use]
163    pub fn with_grace_period(grace: Time) -> Self {
164        Self {
165            grace: Some(grace),
166            name: None,
167        }
168    }
169    #[must_use]
170    pub fn as_named(mut self, name: impl Into<String>) -> Self {
171        self.name = Some(name.into());
172        self
173    }
174}
175
176/// The serdes, and an optional name and partition count, for an explicit
177/// `repartition()`.
178#[derive(Debug, Clone)]
179pub struct Repartitioned<KS, VS> {
180    pub(crate) key_serde: KS,
181    pub(crate) value_serde: VS,
182    pub(crate) name: Option<String>,
183    pub(crate) partitions: Option<i32>,
184}
185impl<KS, VS> Repartitioned<KS, VS> {
186    pub fn with(key_serde: KS, value_serde: VS) -> Self {
187        Self {
188            key_serde,
189            value_serde,
190            name: None,
191            partitions: None,
192        }
193    }
194    #[must_use]
195    pub fn with_name(mut self, name: impl Into<String>) -> Self {
196        self.name = Some(name.into());
197        self
198    }
199    #[must_use]
200    pub fn num_partitions(mut self, n: i32) -> Self {
201        self.partitions = Some(n);
202        self
203    }
204}
205
206#[cfg(test)]
207mod tests {
208    use assert2::check;
209
210    use super::*;
211    use crate::processor::serde::{I64Serde, StringSerde};
212
213    #[test]
214    fn grouped_materialized_repartitioned_carry_serdes_and_names() {
215        let g = Grouped::with(StringSerde, I64Serde).with_name("g1");
216        check!(g.name.as_deref() == Some("g1"));
217        let m = Materialized::with(StringSerde, I64Serde).as_store("counts");
218        check!(m.store_name.as_deref() == Some("counts"));
219        check!(m.logging);
220        let r = Repartitioned::with(StringSerde, I64Serde)
221            .with_name("rp")
222            .num_partitions(4);
223        check!(r.name.as_deref() == Some("rp"));
224        check!(r.partitions == Some(4));
225    }
226
227    #[test]
228    fn materialized_caching_defaults_on_and_toggles() {
229        let m = Materialized::with(StringSerde, I64Serde);
230        check!(m.caching_enabled()); // default true
231        let m = m.with_caching(false);
232        check!(!m.caching_enabled());
233    }
234
235    #[test]
236    fn joined_carries_grace_and_name() {
237        let j = Joined::with_grace_period(secs(5)).as_named("jb");
238        check!(j.grace == Some(secs(5)));
239        check!(j.name.as_deref() == Some("jb"));
240        check!(Joined::default().grace == None);
241    }
242
243    #[test]
244    fn materialized_as_versioned_sets_config() {
245        let m = Materialized::with(StringSerde, I64Serde).as_versioned("vstore", minutes(10));
246        check!(m.store_name.as_deref() == Some("vstore"));
247        let vc = m.versioned.expect("versioned config");
248        check!(vc.history_retention == minutes(10));
249    }
250}
krabka_client_streams/dsl/config.rs

krabka_client_streams/dsl/
config.rs
