krabka_client_streams/dsl/
builder.rs

1//! The public `StreamsBuilder` and the `InternalStreamsBuilder` that holds the
2//! graph and the name counter.
3use std::{cell::RefCell, rc::Rc};
4
5use crate::{
6    dsl::graph::{GraphNodeKind, LogicalGraph},
7    processor::serde::{Consumed, DefaultSerde, Serde, SerdeAssociate},
8};
9
10/// A serde-carrying thunk that registers a DSL-added state store and connects it
11/// to a processor by name during lowering.
12///
13/// `process` and `process_values` look up this thunk and call it.
14pub(crate) type StoreConnectThunk =
15    std::sync::Arc<dyn Fn(&mut crate::dsl::graph::LowerState, &str) + Send + Sync>;
16
17pub(crate) struct InternalStreamsBuilder {
18    pub graph: LogicalGraph,
19    index: usize,
20    /// Serde-carrying connect thunks for DSL-added stores, keyed by store name.
21    /// [`StreamsBuilder::add_state_store`] fills this map. A `process` or
22    /// `process_values` node looks up a thunk and calls it during lowering.
23    pub store_thunks: std::collections::HashMap<String, StoreConnectThunk>,
24}
25
26impl InternalStreamsBuilder {
27    pub fn new() -> Self {
28        Self {
29            graph: LogicalGraph::default(),
30            index: 0,
31            store_thunks: std::collections::HashMap::new(),
32        }
33    }
34
35    /// JVM `InternalStreamsBuilder.newProcessorName`. It returns
36    /// `prefix + %010d` and then increments the counter.
37    pub fn new_processor_name(&mut self, prefix: &str) -> String {
38        let n = format!("{prefix}{:010}", self.index);
39        self.index += 1;
40        n
41    }
42
43    /// The connect thunk for an added store, if one exists, as a cloned `Arc`.
44    /// `process` uses it.
45    pub fn store_thunk(&self, name: &str) -> Option<StoreConnectThunk> {
46        self.store_thunks.get(name).cloned()
47    }
48}
49
50/// The DSL entry point. Build a topology, then call `build` or
51/// `build_optimized`.
52pub struct StreamsBuilder {
53    pub(crate) internal: Rc<RefCell<InternalStreamsBuilder>>,
54}
55
56impl StreamsBuilder {
57    #[must_use]
58    pub fn new() -> Self {
59        Self {
60            internal: Rc::new(RefCell::new(InternalStreamsBuilder::new())),
61        }
62    }
63
64    /// Source a `KStream` from one or more topics with explicit serdes.
65    ///
66    /// Prefer [`StreamsBuilder::stream`], the default-serde form, for types that
67    /// implement [`DefaultSerde`]. Use this escape hatch when a type has no
68    /// default serde, or to override it. Examples are a hand-rolled `Serde<T>`,
69    /// a **key**-role schema serde (`AvroSerde::<T>::key(&cache)`), and a
70    /// validation-on JSON serde (`JsonSerde::value(&cache, true)`).
71    pub fn stream_explicit<KS, VS>(
72        &self,
73        topics: impl IntoIterator<Item = impl Into<String>>,
74        consumed: impl Into<Consumed<KS, VS>>,
75    ) -> crate::dsl::kstream::KStream<KS::Target, VS::Target, KS, VS>
76    where
77        KS: SerdeAssociate + Serde<KS::Target> + Clone,
78        VS: SerdeAssociate + Serde<VS::Target> + Clone,
79        KS::Target: std::any::Any + Send + Clone,
80        VS::Target: std::any::Any + Send + Clone,
81    {
82        let consumed = consumed.into();
83        let topics: Vec<String> = topics.into_iter().map(Into::into).collect();
84        // A stream sourced from exactly one topic carries that topic as its
85        // copartition-group member lineage (used by `KStream::join`). A multi-topic
86        // source has no single member, so the lineage is `None`.
87        let single_source_topic = match topics.as_slice() {
88            [only] => Some(only.clone()),
89            _ => None,
90        };
91        let mut g = self.internal.borrow_mut();
92        let name = g.new_processor_name(crate::dsl::names::SOURCE);
93        let id = g.graph.add(
94            name.clone(),
95            GraphNodeKind::StreamSource {
96                topics: topics.clone(),
97            },
98            vec![],
99        );
100        // Attach the lowering thunk: it performs the typed `Topology::add_source`
101        // call (capturing the source `Consumed` serdes + topics) and records the
102        // resulting node name so children can rebuild a typed parent handle.
103        let consumed_for_lower = consumed.clone();
104        g.graph.nodes[id].lower = Some(Box::new(
105            move |state: &mut crate::dsl::graph::LowerState| {
106                let h = state
107                    .topology
108                    .add_source_explicit::<KS::Target, VS::Target, KS, VS>(
109                        name,
110                        topics,
111                        consumed_for_lower,
112                    );
113                state.handle_name.insert(id, h.name().to_string());
114            },
115        ));
116        drop(g);
117        crate::dsl::kstream::KStream::new(
118            Rc::clone(&self.internal),
119            id,
120            consumed.key_serde.clone(),
121            consumed.value_serde.clone(),
122        )
123        .with_source_topic(single_source_topic)
124    }
125
126    /// Source a materialized `KTable` from a changelog-style topic with explicit
127    /// serdes.
128    ///
129    /// Prefer [`StreamsBuilder::table`], the default-serde form, for
130    /// [`DefaultSerde`] types. Use this method when a type has no default serde
131    /// or to override it, for example with a custom `Serde<T>`, a key-role
132    /// schema serde, or validated JSON.
133    ///
134    /// The method records a single `TableSource` logical node. Its thunk lowers a
135    /// source node, the table-source processor, and the materialized state store.
136    /// The store name comes from `Materialized`, or from a fresh
137    /// `KTABLE-SOURCE-STATE-STORE` counter. The changelog topic is
138    /// `<app>-<store>-changelog`, unless the `REUSE_KTABLE_SOURCE_TOPICS`
139    /// optimizer pass, which `build_optimized` runs, makes it reuse the source
140    /// topic.
141    pub fn table_explicit<KS, VS>(
142        &self,
143        topic: impl Into<String>,
144        consumed: impl Into<Consumed<KS, VS>>,
145        materialized: impl Into<crate::dsl::config::Materialized<KS, VS>>,
146    ) -> crate::dsl::ktable::KTable<KS::Target, VS::Target, KS, VS>
147    where
148        KS: SerdeAssociate + Serde<KS::Target> + Clone + 'static,
149        VS: SerdeAssociate + Serde<VS::Target> + Clone + 'static,
150        KS::Target: std::any::Any + Send + Sync + Clone,
151        VS::Target: std::any::Any + Send + Clone,
152    {
153        let consumed = consumed.into();
154        let materialized = materialized.into();
155        let topic: String = topic.into();
156        // Preserve a copy of the source topic to surface via `KTable::source_topic()`.
157        let topic_for_ktable = topic.clone();
158        // Factory letting a downstream `suppress` register a SuppressBytesStore<K,V>
159        // with this table's serdes (non-windowed). Built before the thunk moves them.
160        let suppress_factory =
161            crate::dsl::ktable::kv_suppress_factory::<KS::Target, VS::Target, KS, VS>(
162                materialized.key_serde.clone(),
163                materialized.value_serde.clone(),
164            );
165        let mut g = self.internal.borrow_mut();
166        // Store name at the JVM position (minted before the source/processor name).
167        let store_name = match &materialized.store_name {
168            Some(name) => name.clone(),
169            None => g.new_processor_name(crate::dsl::names::TABLE_SOURCE),
170        };
171        let source_name = g.new_processor_name(crate::dsl::names::SOURCE);
172        let proc_name = g.new_processor_name(crate::dsl::names::TABLE_SOURCE);
173        let id = g.graph.add(
174            source_name.clone(),
175            GraphNodeKind::TableSource {
176                topic: topic.clone(),
177                store_name: store_name.clone(),
178                reuse_source_for_changelog: false,
179            },
180            vec![],
181        );
182        // Capture versioned config before destructuring moves the serdes.
183        let versioned_cfg = materialized.versioned;
184        // Surface the versioned history-retention on the returned KTable handle
185        // (read by KIP-914 join routing), independent of the thunk's move.
186        let versioned_retention = versioned_cfg.map(|v| v.history_retention);
187        let crate::dsl::config::Materialized {
188            key_serde,
189            value_serde,
190            ..
191        } = materialized;
192        let store_for_thunk = store_name.clone();
193        let key_serde_for_lower = key_serde.clone();
194        let value_serde_for_lower = value_serde.clone();
195        g.graph.nodes[id].lower = Some(Box::new(
196            move |state: &mut crate::dsl::graph::LowerState| {
197                let src = state
198                    .topology
199                    .add_source_explicit::<KS::Target, VS::Target, KS, VS>(
200                        source_name,
201                        [topic],
202                        consumed,
203                    );
204                let store_for_proc = store_for_thunk.clone();
205                if let Some(vc) = versioned_cfg {
206                    // ── Versioned KTable branch (KIP-889/914) ──────────────────
207                    let h = state
208                        .topology
209                        .add_processor::<KS::Target, VS::Target, KS::Target, crate::dsl::processors::change::Change<VS::Target>, _, _, _>(
210                            proc_name,
211                            move || crate::dsl::processors::table::VersionedKTableSourceProcessor {
212                                store_name: store_for_proc.clone(),
213                                _pd: std::marker::PhantomData,
214                            },
215                            [&src],
216                        );
217                    // REUSE_KTABLE_SOURCE_TOPICS applies to versioned tables too:
218                    // under optimization the changelog is the source topic, else
219                    // the default `<app>-<store>-changelog`.
220                    match state.reuse_changelog.get(&id).cloned() {
221                        Some(changelog_topic) => {
222                            state
223                                .topology
224                                .add_versioned_store_with_changelog::<KS::Target, VS::Target, KS, VS>(
225                                    store_for_thunk.clone(),
226                                    key_serde_for_lower,
227                                    value_serde_for_lower,
228                                    vc.history_retention,
229                                    [h.name().to_string()],
230                                    changelog_topic,
231                                );
232                        }
233                        None => {
234                            state
235                                .topology
236                                .add_versioned_store::<KS::Target, VS::Target, KS, VS>(
237                                    store_for_thunk.clone(),
238                                    key_serde_for_lower,
239                                    value_serde_for_lower,
240                                    vc.history_retention,
241                                    [h.name().to_string()],
242                                );
243                        }
244                    }
245                    state.handle_name.insert(id, h.name().to_string());
246                } else {
247                    // ── Standard KV KTable branch ──────────────────────────────
248                    // The KTable source forwards Change<V> (prior store value as old).
249                    let h = state
250                        .topology
251                        .add_processor::<KS::Target, VS::Target, KS::Target, crate::dsl::processors::change::Change<VS::Target>, _, _, _>(
252                            proc_name,
253                            move || crate::dsl::processors::table::KTableSourceProcessor {
254                                store_name: store_for_proc.clone(),
255                                forwarder:
256                                    crate::dsl::processors::tuple_forwarder::TupleForwarder::default(
257                                    ),
258                                _pd: std::marker::PhantomData,
259                            },
260                            [&src],
261                        );
262                    // REUSE_KTABLE_SOURCE_TOPICS: if the optimizer flagged this
263                    // TableSource, register the store with the source topic as its
264                    // changelog; otherwise the default `<app>-<store>-changelog`.
265                    match state.reuse_changelog.get(&id).cloned() {
266                        Some(changelog_topic) => {
267                            state
268                                .topology
269                                .add_state_store_with_changelog::<KS::Target, VS::Target, KS, VS>(
270                                    store_for_thunk.clone(),
271                                    key_serde_for_lower,
272                                    value_serde_for_lower,
273                                    [h.name().to_string()],
274                                    changelog_topic,
275                                );
276                        }
277                        None => {
278                            state
279                                .topology
280                                .add_state_store::<KS::Target, VS::Target, KS, VS>(
281                                    store_for_thunk.clone(),
282                                    key_serde_for_lower,
283                                    value_serde_for_lower,
284                                    [h.name().to_string()],
285                                );
286                        }
287                    }
288                    // Children of the TableSource wire to the processor output.
289                    state.handle_name.insert(id, h.name().to_string());
290                }
291            },
292        ));
293        drop(g);
294        crate::dsl::ktable::KTable::new(
295            Rc::clone(&self.internal),
296            id,
297            Some(store_name),
298            Some(topic_for_ktable),
299            key_serde.clone(),
300            value_serde.clone(),
301        )
302        .with_suppress_factory(Some(suppress_factory))
303        .with_versioned_retention(versioned_retention)
304    }
305
306    /// Source a [`GlobalKTable`] from a topic with explicit serdes. A
307    /// `GlobalKTable` is a fully-replicated lookup table and works only as a
308    /// join target.
309    ///
310    /// Prefer [`StreamsBuilder::global_table`], the default-serde form, for
311    /// [`DefaultSerde`] types. Use this method when a type has no default serde
312    /// or to override it, for example with a custom `Serde<T>`, a key-role
313    /// schema serde, or validated JSON.
314    ///
315    /// The method records a single `GlobalSource` logical node. Its thunk calls
316    /// [`Topology::add_global_store`] to lower a source node, an
317    /// update-processor, and a global KV store. The store, the source, and the
318    /// processor are **invisible in the wire**, with no subtopology and no
319    /// changelog. The global source node still takes a node-group index during
320    /// grouping, so a `global_table` declared before `stream` shifts the stream
321    /// subtopology id, for example to `"1"`. The store name comes from
322    /// `materialized`, or from a fresh `KTABLE-SOURCE-STATE-STORE` counter, and
323    /// it is minted at the JVM position, before the source name and the
324    /// processor name.
325    ///
326    /// [`GlobalKTable`]: crate::dsl::global_table::GlobalKTable
327    /// [`Topology::add_global_store`]: crate::topology::Topology::add_global_store
328    pub fn global_table_explicit<KS, VS>(
329        &self,
330        topic: impl Into<String>,
331        consumed: impl Into<Consumed<KS, VS>>,
332        materialized: impl Into<crate::dsl::config::Materialized<KS, VS>>,
333    ) -> crate::dsl::global_table::GlobalKTable<KS::Target, VS::Target, KS, VS>
334    where
335        KS: SerdeAssociate + Serde<KS::Target> + Clone + 'static,
336        VS: SerdeAssociate + Serde<VS::Target> + Clone + 'static,
337        KS::Target: std::any::Any + Send + Sync + Clone,
338        VS::Target: std::any::Any + Send + Clone,
339    {
340        let consumed = consumed.into();
341        let materialized = materialized.into();
342        let topic: String = topic.into();
343        let topic_for_handle = topic.clone();
344        let mut g = self.internal.borrow_mut();
345        // Store name at the JVM position (minted before the source/processor name).
346        let store_name = match &materialized.store_name {
347            Some(name) => name.clone(),
348            None => g.new_processor_name(crate::dsl::names::TABLE_SOURCE),
349        };
350        let source_name = g.new_processor_name(crate::dsl::names::GLOBAL_SOURCE);
351        let proc_name = g.new_processor_name(crate::dsl::names::GLOBAL_PROCESSOR);
352        let id = g.graph.add(
353            source_name.clone(),
354            GraphNodeKind::GlobalSource {
355                topic: topic.clone(),
356                store_name: store_name.clone(),
357                source_name: source_name.clone(),
358                processor_name: proc_name.clone(),
359            },
360            vec![],
361        );
362        // `materialized` carries the store serdes; for a global table the store and
363        // source serdes coincide, so `add_global_store` (which uses one `Consumed`
364        // for both the source deser and the store factory) takes the source serdes.
365        // Drop `materialized` to keep its API surface symmetric with `table()` even
366        // though its serdes aren't separately threaded here.
367        let key_serde = materialized.key_serde.clone();
368        let value_serde = materialized.value_serde.clone();
369        drop(materialized);
370        let store_for_handle = store_name.clone();
371        g.graph.nodes[id].lower = Some(Box::new(
372            move |state: &mut crate::dsl::graph::LowerState| {
373                state
374                    .topology
375                    .add_global_store::<KS::Target, VS::Target, KS, VS>(
376                        store_name,
377                        source_name,
378                        topic,
379                        proc_name,
380                        consumed,
381                    );
382            },
383        ));
384        drop(g);
385        crate::dsl::global_table::GlobalKTable::new(
386            Rc::clone(&self.internal),
387            id,
388            store_for_handle,
389            topic_for_handle,
390            key_serde,
391            value_serde,
392        )
393    }
394
395    /// Source a `KStream` using each type's [`DefaultSerde`]. Use
396    /// [`StreamsBuilder::stream_explicit`] to supply custom serdes.
397    pub fn stream<K, V>(
398        &self,
399        topics: impl IntoIterator<Item = impl Into<String>>,
400    ) -> crate::dsl::kstream::KStream<K, V>
401    where
402        K: DefaultSerde,
403        V: DefaultSerde,
404        K::Serde: SerdeAssociate<Target = K> + Clone,
405        V::Serde: SerdeAssociate<Target = V> + Clone,
406        K: std::any::Any + Send + Clone,
407        V: std::any::Any + Send + Clone,
408    {
409        self.stream_explicit(
410            topics,
411            Consumed::with(K::Serde::default(), V::Serde::default()),
412        )
413    }
414
415    /// Source a materialized `KTable` using each type's [`DefaultSerde`]. Use
416    /// [`StreamsBuilder::table_explicit`] to supply custom serdes or a custom
417    /// `Materialized`.
418    pub fn table<K, V>(
419        &self,
420        topic: impl Into<String>,
421        store_name: impl Into<String>,
422    ) -> crate::dsl::ktable::KTable<K, V>
423    where
424        K: DefaultSerde,
425        V: DefaultSerde,
426        K::Serde: SerdeAssociate<Target = K> + Clone,
427        V::Serde: SerdeAssociate<Target = V> + Clone,
428        K: std::any::Any + Send + Sync + Clone,
429        V: std::any::Any + Send + Clone,
430    {
431        self.table_explicit(
432            topic,
433            Consumed::with(K::Serde::default(), V::Serde::default()),
434            crate::dsl::config::Materialized::with(K::Serde::default(), V::Serde::default())
435                .as_store(store_name),
436        )
437    }
438
439    /// Source a [`GlobalKTable`] using each type's [`DefaultSerde`]. Use
440    /// [`StreamsBuilder::global_table_explicit`] to supply custom serdes.
441    ///
442    /// [`GlobalKTable`]: crate::dsl::global_table::GlobalKTable
443    pub fn global_table<K, V>(
444        &self,
445        topic: impl Into<String>,
446        store_name: impl Into<String>,
447    ) -> crate::dsl::global_table::GlobalKTable<K, V>
448    where
449        K: DefaultSerde,
450        V: DefaultSerde,
451        K::Serde: SerdeAssociate<Target = K> + Clone,
452        V::Serde: SerdeAssociate<Target = V> + Clone,
453        K: std::any::Any + Send + Sync + Clone,
454        V: std::any::Any + Send + Clone,
455    {
456        self.global_table_explicit(
457            topic,
458            Consumed::with(K::Serde::default(), V::Serde::default()),
459            crate::dsl::config::Materialized::with(K::Serde::default(), V::Serde::default())
460                .as_store(store_name),
461        )
462    }
463
464    /// Register a state store that the DSL can connect by name to a `process` or
465    /// `process_values` node.
466    ///
467    /// The store is registered and its compact changelog is emitted when a
468    /// `process` call connects it. Call this method BEFORE the `process` that
469    /// names the store.
470    ///
471    /// The method captures the serdes into a connect thunk. During lowering the
472    /// thunk calls [`Topology::add_state_store`] with the named processor as the
473    /// store's connected processor, which gives the standard
474    /// `<app>-<name>-changelog` compact changelog. The thunk is recorded under
475    /// `name`, and `process` looks it up.
476    ///
477    /// [`Topology::add_state_store`]: crate::topology::Topology::add_state_store
478    pub fn add_state_store<KS, VS>(
479        &self,
480        name: impl Into<String>,
481        key_serde: KS,
482        value_serde: VS,
483    ) -> &Self
484    where
485        KS: SerdeAssociate + Serde<KS::Target> + Clone + 'static,
486        VS: SerdeAssociate + Serde<VS::Target> + Clone + 'static,
487        KS::Target: std::any::Any + Send + Sync + Clone,
488        VS::Target: std::any::Any + Send + Clone,
489    {
490        let name: String = name.into();
491        // `name` is moved into the thunk; keep a copy for the map key.
492        let key = name.clone();
493        let thunk: StoreConnectThunk = std::sync::Arc::new(
494            move |state: &mut crate::dsl::graph::LowerState, processor: &str| {
495                state
496                    .topology
497                    .add_state_store::<KS::Target, VS::Target, KS, VS>(
498                        name.clone(),
499                        key_serde.clone(),
500                        value_serde.clone(),
501                        [processor.to_string()],
502                    );
503            },
504        );
505        self.internal.borrow_mut().store_thunks.insert(key, thunk);
506        self
507    }
508
509    /// Build the topology with no optimizer, which is the JVM
510    /// `NO_OPTIMIZATION` default.
511    ///
512    /// The method lowers the logical graph straight to the Processor-API
513    /// [`crate::topology::Topology`] and then finalizes it into a
514    /// [`BuiltTopology`].
515    ///
516    /// It consumes the builder, so no [`KStream`] or [`KTable`] handle may still
517    /// be alive. Each handle holds an `Rc` clone of the internal builder, so an
518    /// outstanding handle makes `Rc::try_unwrap` fail and the method panics. The
519    /// fluent `stream(..).map_values(..)..to_explicit(..)` form drops every
520    /// intermediate handle before `build`, which meets this requirement.
521    ///
522    /// [`KStream`]: crate::dsl::kstream::KStream
523    /// [`KTable`]: crate::dsl::ktable::KTable
524    /// [`BuiltTopology`]: crate::topology::BuiltTopology
525    #[tracing::instrument(
526        name = "streams.dsl.build",
527        level = "info",
528        skip_all,
529        fields(app_id = %app_id),
530        err,
531    )]
532    /// # Errors
533    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
534    pub fn build(
535        self,
536        app_id: &str,
537    ) -> Result<crate::topology::BuiltTopology, crate::topology::TopologyError> {
538        let graph = self.into_graph("build");
539        let topology = crate::dsl::lower::lower(graph, app_id);
540        topology.build(app_id)
541    }
542
543    /// Build the topology with the DSL optimizations on, which matches the JVM
544    /// `optimization=all`.
545    ///
546    /// The method runs the optimizer passes over the logical graph, then lowers
547    /// the graph to the Processor-API [`crate::topology::Topology`] and
548    /// finalizes it.
549    ///
550    /// There are two passes. `MERGE_REPARTITION_TOPICS` makes two aggregations
551    /// off one key-changing op share a single repartition topic.
552    /// `REUSE_KTABLE_SOURCE_TOPICS` makes a `builder.table_explicit()` store
553    /// reuse its source topic as its changelog. The passes are independent, so
554    /// their order does not matter. The outstanding-handle requirement is the
555    /// same as for [`build`](Self::build).
556    #[tracing::instrument(
557        name = "streams.dsl.build_optimized",
558        level = "info",
559        skip_all,
560        fields(app_id = %app_id),
561        err,
562    )]
563    /// # Errors
564    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
565    pub fn build_optimized(
566        self,
567        app_id: &str,
568    ) -> Result<crate::topology::BuiltTopology, crate::topology::TopologyError> {
569        let mut graph = self.into_graph("build_optimized");
570        crate::dsl::optimizer::merge_repartition_topics(&mut graph);
571        crate::dsl::optimizer::reuse_ktable_source_topics(&mut graph);
572        let topology = crate::dsl::lower::lower(graph, app_id);
573        topology.build(app_id)
574    }
575
576    /// Unwrap the shared internal builder into its [`LogicalGraph`].
577    ///
578    /// No [`KStream`] or [`KTable`] handle may still be alive, because each one
579    /// holds an `Rc` clone of the internal builder. An outstanding handle makes
580    /// `Rc::try_unwrap` fail, and the method panics.
581    ///
582    /// [`KStream`]: crate::dsl::kstream::KStream
583    /// [`KTable`]: crate::dsl::ktable::KTable
584    /// [`LogicalGraph`]: crate::dsl::graph::LogicalGraph
585    fn into_graph(self, who: &str) -> LogicalGraph {
586        Rc::try_unwrap(self.internal)
587            .unwrap_or_else(|_| panic!("StreamsBuilder::{who}: outstanding KStream/KTable handles"))
588            .into_inner()
589            .graph
590    }
591}
592
593impl Default for StreamsBuilder {
594    fn default() -> Self {
595        Self::new()
596    }
597}
598
599#[cfg(test)]
600mod tests {
601    use assert2::check;
602
603    use super::*;
604
605    #[test]
606    fn counter_assigns_jvm_names_in_call_order() {
607        let mut b = InternalStreamsBuilder::new();
608        check!(b.new_processor_name(crate::dsl::names::SOURCE) == "KSTREAM-SOURCE-0000000000");
609        check!(
610            b.new_processor_name(crate::dsl::names::MAPVALUES) == "KSTREAM-MAPVALUES-0000000001"
611        );
612        check!(b.new_processor_name(crate::dsl::names::FILTER) == "KSTREAM-FILTER-0000000002");
613    }
614
615    #[test]
616    fn stream_records_a_source_node() {
617        let builder = StreamsBuilder::new();
618        let _s = builder.stream::<String, String>(["in"]);
619        let g = builder.internal.borrow();
620        check!(g.graph.nodes.len() == 1);
621        check!(matches!(
622            g.graph.nodes[0].kind,
623            GraphNodeKind::StreamSource { .. }
624        ));
625        check!(g.graph.nodes[0].name == "KSTREAM-SOURCE-0000000000");
626    }
627
628    #[test]
629    fn build_lowers_source_sink_to_wire_topology() {
630        let b = StreamsBuilder::new();
631        b.stream::<String, String>(["in"]).to("out");
632        let built = b.build("app").unwrap();
633        let wire = built.to_wire();
634        check!(wire.epoch == 0);
635        check!(wire.subtopologies.len() == 1);
636        check!(wire.subtopologies[0].source_topics == vec!["in".to_string()]);
637        // The sink topic surfaces as a list-able output topic.
638        check!(built.list_sink_topics() == vec!["out".to_string()]);
639    }
640
641    #[test]
642    fn build_optimized_matches_build_for_stateless_chain() {
643        let b = StreamsBuilder::new();
644        b.stream::<String, String>(["in"])
645            .map_values(|v: &String| v.clone())
646            .to("out");
647        let wire = b.build_optimized("app").unwrap().to_wire();
648        check!(wire.subtopologies.len() == 1);
649        check!(wire.subtopologies[0].source_topics == vec!["in".to_string()]);
650    }
651
652    #[test]
653    fn table_build_keeps_derived_changelog_name() {
654        // Without the optimizer (plain `build`), a `table()` store's changelog is
655        // the JVM-default `<app>-<store>-changelog` — REUSE_KTABLE_SOURCE_TOPICS
656        // must NOT fire.
657        let b = StreamsBuilder::new();
658        b.table::<String, String>("in", "store")
659            .to_stream()
660            .to("out");
661        let wire = b.build("app").unwrap().to_wire();
662        let cl: Vec<&str> = wire.subtopologies[0]
663            .state_changelog_topics
664            .iter()
665            .map(|t| t.name.as_str())
666            .collect();
667        check!(cl == vec!["app-store-changelog"]);
668    }
669
670    #[test]
671    fn table_optimized_build_reuses_source_topic_as_changelog() {
672        // With the optimizer (`build_optimized`), the `table()` store's changelog
673        // is the SOURCE topic ("in"), not "app-store-changelog".
674        let b = StreamsBuilder::new();
675        b.table::<String, String>("in", "store")
676            .to_stream()
677            .to("out");
678        let wire = b.build_optimized("app").unwrap().to_wire();
679        let cl: Vec<&str> = wire.subtopologies[0]
680            .state_changelog_topics
681            .iter()
682            .map(|t| t.name.as_str())
683            .collect();
684        check!(cl == vec!["in"]);
685    }
686
687    #[test]
688    fn global_table_returns_handle_with_store_name() {
689        let builder = StreamsBuilder::new();
690        let gt = builder.global_table::<String, String>("global", "g-store");
691        check!(gt.store_name() == "g-store");
692        check!(gt.source_topic == "global");
693        // The global source is the FIRST logical node (so it takes index 0 at
694        // grouping time, bumping a later stream's subtopology id).
695        let g = builder.internal.borrow();
696        check!(matches!(
697            g.graph.nodes[0].kind,
698            GraphNodeKind::GlobalSource { .. }
699        ));
700    }
701
702    #[test]
703    fn global_table_before_stream_bumps_stream_subtopology_to_one() {
704        let b = StreamsBuilder::new();
705        // Declared FIRST: the global source is registered before the stream source,
706        // so it consumes node-group index 0 and the stream emits as "1".
707        let gt = b.global_table::<String, String>("global", "g-store");
708        // The GlobalKTable handle holds an `Rc` clone of the internal builder; drop
709        // it before `build()` (which requires `Rc::try_unwrap` to succeed).
710        drop(gt);
711        b.stream::<String, String>(["in"]).to("out");
712        let wire = b.build("app").unwrap().to_wire();
713        check!(wire.subtopologies.len() == 1);
714        check!(wire.subtopologies[0].subtopology_id == "1");
715        check!(wire.subtopologies[0].source_topics == vec!["in".to_string()]);
716        // No changelog topic for the global store.
717        check!(
718            wire.subtopologies
719                .iter()
720                .all(|s| s.state_changelog_topics.is_empty())
721        );
722    }
723
724    #[test]
725    fn add_state_store_records_a_connect_thunk() {
726        use crate::processor::serde::{I64Serde, StringSerde};
727        let b = StreamsBuilder::new();
728        // Chains (returns &Self) and records a thunk under the given name.
729        b.add_state_store("counts", StringSerde, I64Serde);
730        check!(b.internal.borrow().store_thunk("counts").is_some());
731        check!(b.internal.borrow().store_thunk("missing").is_none());
732    }
733
734    #[test]
735    #[should_panic(expected = "outstanding KStream/KTable handles")]
736    fn build_panics_with_outstanding_handle() {
737        let b = StreamsBuilder::new();
738        // Hold a live KStream handle across the build call: it keeps an `Rc`
739        // clone of the internal builder alive, so `Rc::try_unwrap` fails.
740        let _held = b.stream::<String, String>(["in"]);
741        let _ = b.build("app");
742    }
743}
krabka_client_streams/dsl/builder.rs

krabka_client_streams/dsl/
builder.rs
