tinyflow-framework 0.1.1

//! Linear operator chain built at DSL time (no per-step graph nodes).
//!
//! **Build-time errors panic** (duplicate source, frozen chain after terminal sink, missing `add_source`).
//! Runtime failures use [`crate::api::error::StreamResult`] in [`crate::stream::chain_execute::execute`].
//! Parallelism is set once on [`crate::env::stream_env::StreamEnv`] and stored on [`ChainJob`].

use std::any::TypeId;

use crate::runtime::chain_segment::ChainSegmentFactory;

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChainOperatorKind {
    Source,
    Map,
    Filter,
    FlatMap,
    Sink,
}

pub struct ChainStep {
    pub kind: ChainOperatorKind,
    pub user_name: Option<String>,
    pub factory: ChainSegmentFactory,
}

pub struct ChainJob {
    pub parallelism: usize,
    pub steps: Vec<ChainStep>,
    pub frozen: bool,
    /// Output type of the tail step (for debug/test build-time checks).
    pub output_type: Option<TypeId>,
}

impl ChainJob {
    pub fn new(parallelism: usize) -> Self {
        Self {
            parallelism,
            steps: Vec::new(),
            frozen: false,
            output_type: None,
        }
    }

    pub fn push_step_with_types(
        &mut self,
        step: ChainStep,
        input_type: Option<TypeId>,
        output_type: TypeId,
    ) {
        if self.frozen {
            panic!("chain is frozen after terminal sink");
        }
        if cfg!(debug_assertions)
            && let (Some(expected_in), Some(prev_out)) = (input_type, self.output_type)
        {
            assert_eq!(
                prev_out, expected_in,
                "chain type mismatch at step {:?}: previous output {:?} != expected input {:?}",
                step.kind, prev_out, expected_in
            );
        }
        self.output_type = Some(output_type);
        let is_sink = step.kind == ChainOperatorKind::Sink;
        self.steps.push(step);
        if is_sink {
            self.frozen = true;
        }
    }
}