tinyflow_framework/runtime/
chain_job.rs

1//! Linear operator chain built at DSL time (no per-step graph nodes).
2//!
3//! **Build-time errors panic** (duplicate source, frozen chain after terminal sink, missing `add_source`).
4//! Runtime failures use [`crate::api::error::StreamResult`] in [`crate::stream::chain_execute::execute`].
5//! Parallelism is set once on [`crate::env::stream_env::StreamEnv`] and stored on [`ChainJob`].
6
7use std::any::TypeId;
8
9use crate::runtime::chain_segment::ChainSegmentFactory;
10
11#[derive(Clone, Copy, Debug, PartialEq, Eq)]
12pub enum ChainOperatorKind {
13    Source,
14    Map,
15    Filter,
16    FlatMap,
17    Sink,
18}
19
20pub struct ChainStep {
21    pub kind: ChainOperatorKind,
22    pub user_name: Option<String>,
23    pub factory: ChainSegmentFactory,
24}
25
26pub struct ChainJob {
27    pub parallelism: usize,
28    pub steps: Vec<ChainStep>,
29    pub frozen: bool,
30    /// Output type of the tail step (for debug/test build-time checks).
31    pub output_type: Option<TypeId>,
32}
33
34impl ChainJob {
35    pub fn new(parallelism: usize) -> Self {
36        Self {
37            parallelism,
38            steps: Vec::new(),
39            frozen: false,
40            output_type: None,
41        }
42    }
43
44    pub fn push_step_with_types(
45        &mut self,
46        step: ChainStep,
47        input_type: Option<TypeId>,
48        output_type: TypeId,
49    ) {
50        if self.frozen {
51            panic!("chain is frozen after terminal sink");
52        }
53        if cfg!(debug_assertions)
54            && let (Some(expected_in), Some(prev_out)) = (input_type, self.output_type)
55        {
56            assert_eq!(
57                prev_out, expected_in,
58                "chain type mismatch at step {:?}: previous output {:?} != expected input {:?}",
59                step.kind, prev_out, expected_in
60            );
61        }
62        self.output_type = Some(output_type);
63        let is_sink = step.kind == ChainOperatorKind::Sink;
64        self.steps.push(step);
65        if is_sink {
66            self.frozen = true;
67        }
68    }
69}
tinyflow_framework/runtime/chain_job.rs

tinyflow_framework/runtime/
chain_job.rs
