camel-core 0.47.0

//! Splitting step compilers: Split, DeclarativeSplit, DeclarativeStreamSplit, Aggregate, Multicast.
//!
//! These steps split exchanges into fragments or aggregate them back together.

use camel_api::SpanKindHint;
use std::sync::Arc;

use bytes::Bytes;
use futures::{Stream, StreamExt};
use std::pin::Pin;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

use camel_api::{
    Body, BoxProcessor, CamelError, Exchange, MulticastStrategy, StepLifecycle, StreamSplitFormat,
    Value,
};

use super::{
    CompilationContext, CompileOutcome, CompiledStep, StepCompiler, StepCompilerRegistry,
    pack_lifecycles,
};
use crate::lifecycle::adapters::route_compiler::compose_outcome_segment;
use crate::lifecycle::adapters::route_controller::SharedLanguageRegistry;
use crate::lifecycle::adapters::step_resolution::{await_eval, compile_language_expression};
use crate::lifecycle::application::route_definition::BuilderStep;

/// Collect a byte stream into a single `Bytes` value, enforcing a size limit.
///
/// `archive_label` names the archive in the cap-violation error (e.g.
/// "ZIP archive", "TAR archive").
async fn collect_stream_with_limit(
    stream: futures::stream::BoxStream<'static, Result<Bytes, CamelError>>,
    max_bytes: u64,
    archive_label: &str,
) -> Result<Bytes, CamelError> {
    let mut stream = stream;
    let mut buf = Vec::new();
    while let Some(chunk) = stream.next().await {
        let chunk = chunk?;
        let new_len = buf.len().saturating_add(chunk.len());
        if new_len as u64 > max_bytes {
            return Err(CamelError::TypeConversionFailed(format!(
                "{archive_label} exceeds max compressed size: {max_bytes}"
            )));
        }
        buf.extend_from_slice(&chunk);
    }
    Ok(Bytes::from(buf))
}

/// Extract bytes from a materialized body (Bytes or Text), creating a parent exchange
/// with an empty body.
///
/// `archive_label` names the archive in the unsupported-body error (e.g. "ZIP", "TAR").
fn extract_bytes_from_body(
    exchange: &Exchange,
    archive_label: &str,
) -> Result<(Bytes, Exchange), CamelError> {
    match &exchange.input.body {
        Body::Bytes(b) => {
            let bytes = b.clone();
            let mut parent = exchange.clone();
            parent.input.body = Body::Empty;
            Ok((bytes, parent))
        }
        Body::Text(s) => {
            let bytes = Bytes::copy_from_slice(s.as_bytes());
            let mut parent = exchange.clone();
            parent.input.body = Body::Empty;
            Ok((bytes, parent))
        }
        _ => Err(CamelError::ProcessorError(format!(
            "{archive_label} split requires Body::Bytes, Body::Text, or Body::Stream"
        ))),
    }
}

type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, CamelError>> + Send>>;

/// Extract the byte stream from a StreamBody (single-consumption).
fn take_stream(stream_body: &camel_api::StreamBody) -> Result<ByteStream, CamelError> {
    match stream_body.stream.try_lock() {
        Ok(mut guard) => match guard.take() {
            Some(s) => Ok(s),
            None => Err(CamelError::ProcessorError(
                "stream body already consumed".into(),
            )),
        },
        Err(_) => Err(CamelError::ProcessorError("stream body locked".into())),
    }
}

pub(crate) struct SplittingCompiler;

impl StepCompiler for SplittingCompiler {
    fn compile(
        &self,
        step: BuilderStep,
        _step_index: usize,
        ctx: &CompilationContext,
        registry: &StepCompilerRegistry,
    ) -> Result<CompileOutcome, CamelError> {
        match step {
            // ── Split (programmatic) ──
            BuilderStep::Split { config, steps } => {
                let (sub_segments, lifecycles) = ctx.compile_children_segments(steps, registry)?;
                let body_segment = compose_outcome_segment(sub_segments);
                let split_segment = camel_processor::SplitSegment {
                    splitter: config.expression,
                    body: body_segment,
                    parallel: config.parallel,
                    parallel_limit: config.parallel_limit,
                    stop_on_exception: config.stop_on_exception,
                    aggregation: config.aggregation,
                };
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(split_segment)),
                    body_contract: None,
                    lifecycle: pack_lifecycles(lifecycles),
                    label: None,
                }))
            }

            // ── DeclarativeSplit ──
            BuilderStep::DeclarativeSplit {
                expression,
                aggregation,
                parallel,
                parallel_limit,
                stop_on_exception,
                steps,
            } => {
                let lang_expr = compile_language_expression(ctx.languages, &expression)?;
                let split_fn: camel_api::splitter::SplitExpression = Arc::new(
                    move |exchange: &Exchange| {
                        let value = await_eval(&lang_expr, exchange);
                        match value {
                            Value::String(s) => Ok(s
                                .lines()
                                .filter(|line| !line.is_empty())
                                .map(|line| {
                                    let mut fragment = exchange.clone();
                                    fragment.input.body = Body::from(line.to_string());
                                    fragment
                                })
                                .collect()),
                            Value::Array(arr) => Ok(arr
                                .into_iter()
                                .map(|v| {
                                    let mut fragment = exchange.clone();
                                    fragment.input.body = Body::from(v);
                                    fragment
                                })
                                .collect()),
                            _ => Err(CamelError::TypeConversionFailed(format!(
                                "declarative split requires a text or array value, got {received}; add an unmarshal step before split",
                                received = camel_api::value_type_name(&value)
                            ))),
                        }
                    },
                );

                let (sub_segments, lifecycles) = ctx.compile_children_segments(steps, registry)?;
                let body_segment = compose_outcome_segment(sub_segments);
                let split_segment = camel_processor::SplitSegment {
                    splitter: split_fn,
                    body: body_segment,
                    parallel,
                    parallel_limit,
                    stop_on_exception,
                    aggregation,
                };
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(split_segment)),
                    body_contract: None,
                    lifecycle: pack_lifecycles(lifecycles),
                    label: None,
                }))
            }

            // ── DeclarativeStreamSplit ──
            BuilderStep::DeclarativeStreamSplit {
                stream_config,
                aggregation,
                stop_on_exception,
                steps,
            } => {
                stream_config
                    .validate()
                    .map_err(|e| CamelError::Config(format!("invalid stream config: {e}")))?;
                let (sub_segments, lifecycles) = ctx.compile_children_segments(steps, registry)?;
                let body_segment = compose_outcome_segment(sub_segments);

                let config_clone = stream_config.clone();
                let zip_config = camel_processor::zip_splitter::ZipSplitConfig::default();
                let tar_config = camel_processor::tar_splitter::TarSplitConfig::default();
                let expression: camel_api::StreamingSplitExpression = Arc::new(
                    move |exchange: Exchange| {
                        let config = config_clone.clone();
                        match config.format {
                            StreamSplitFormat::Zip => {
                                let zip_config = zip_config.clone();
                                match &exchange.input.body {
                                    Body::Stream(sb) => {
                                        let sb = sb.clone();
                                        let mut parent = exchange.clone();
                                        parent.input.body = Body::Empty;
                                        let stream = match take_stream(&sb) {
                                            Ok(s) => s,
                                            Err(e) => {
                                                return Box::pin(futures::stream::once(
                                                    async move { Err(e) },
                                                ));
                                            }
                                        };
                                        Box::pin(async_stream::stream! {
                                            let collected =
                                                match collect_stream_with_limit(
                                                    stream,
                                                    zip_config.max_compressed_size,
                                                    "ZIP archive",
                                                )
                                                .await
                                                {
                                                    Ok(b) => b,
                                                    Err(e) => {
                                                        yield Err(e);
                                                        return;
                                                    }
                                                };
                                            parent.set_property(
                                                "CamelSplitMaterialized",
                                                Value::Bool(true),
                                            );
                                            parent.set_property(
                                                "CamelSplitMaterializedBytes",
                                                Value::from(collected.len() as u64),
                                            );
                                            let mut result_stream =
                                                camel_processor::zip_splitter::split_zip_bytes(
                                                    parent,
                                                    collected,
                                                    zip_config,
                                                );
                                            while let Some(item) = result_stream.next().await {
                                                yield item;
                                            }
                                        })
                                    }
                                    _ => match extract_bytes_from_body(&exchange, "ZIP") {
                                        Ok((bytes, mut parent)) => {
                                            parent.set_property(
                                                "CamelSplitMaterialized",
                                                Value::Bool(true),
                                            );
                                            parent.set_property(
                                                "CamelSplitMaterializedBytes",
                                                Value::from(bytes.len() as u64),
                                            );
                                            camel_processor::zip_splitter::split_zip_bytes(
                                                parent, bytes, zip_config,
                                            )
                                        }
                                        Err(e) => {
                                            Box::pin(futures::stream::once(async move { Err(e) }))
                                        }
                                    },
                                }
                            }
                            StreamSplitFormat::Tar | StreamSplitFormat::TarGz => {
                                let tar_config = tar_config.clone();
                                let gz = matches!(config.format, StreamSplitFormat::TarGz);
                                match &exchange.input.body {
                                    Body::Stream(sb) => {
                                        let sb = sb.clone();
                                        let mut parent = exchange.clone();
                                        parent.input.body = Body::Empty;
                                        let stream = match take_stream(&sb) {
                                            Ok(s) => s,
                                            Err(e) => {
                                                return Box::pin(futures::stream::once(
                                                    async move { Err(e) },
                                                ));
                                            }
                                        };
                                        Box::pin(async_stream::stream! {
                                            let label = if gz {
                                                "TAR.GZ archive"
                                            } else {
                                                "TAR archive"
                                            };
                                            let collected =
                                                match collect_stream_with_limit(
                                                    stream,
                                                    tar_config.max_compressed_size,
                                                    label,
                                                )
                                                .await
                                                {
                                                    Ok(b) => b,
                                                    Err(e) => {
                                                        yield Err(e);
                                                        return;
                                                    }
                                                };
                                            parent.set_property(
                                                "CamelSplitMaterialized",
                                                Value::Bool(true),
                                            );
                                            parent.set_property(
                                                "CamelSplitMaterializedBytes",
                                                Value::from(collected.len() as u64),
                                            );
                                            let mut result_stream = if gz {
                                                camel_processor::tar_splitter::split_tar_gz_bytes(
                                                    parent,
                                                    collected,
                                                    tar_config,
                                                )
                                            } else {
                                                camel_processor::tar_splitter::split_tar_bytes(
                                                    parent,
                                                    collected,
                                                    tar_config,
                                                )
                                            };
                                            while let Some(item) = result_stream.next().await {
                                                yield item;
                                            }
                                        })
                                    }
                                    _ => {
                                        let label = if gz { "TAR.GZ" } else { "TAR" };
                                        match extract_bytes_from_body(&exchange, label) {
                                            Ok((bytes, mut parent)) => {
                                                parent.set_property(
                                                    "CamelSplitMaterialized",
                                                    Value::Bool(true),
                                                );
                                                parent.set_property(
                                                    "CamelSplitMaterializedBytes",
                                                    Value::from(bytes.len() as u64),
                                                );
                                                if gz {
                                                    camel_processor::tar_splitter::split_tar_gz_bytes(
                                                        parent, bytes, tar_config,
                                                    )
                                                } else {
                                                    camel_processor::tar_splitter::split_tar_bytes(
                                                        parent, bytes, tar_config,
                                                    )
                                                }
                                            }
                                            Err(e) => {
                                                Box::pin(futures::stream::once(
                                                    async move { Err(e) },
                                                ))
                                            }
                                        }
                                    }
                                }
                            }
                            _ => {
                                // Incremental codec path (ndjson, lines, chunks)
                                let (stream_body, parent) = match &exchange.input.body {
                                    Body::Stream(sb) => (sb.clone(), {
                                        let mut p = exchange.clone();
                                        p.input.body = Body::Empty;
                                        p
                                    }),
                                    _ => {
                                        let err = camel_api::streaming_split_type_error(
                                            &exchange.input.body,
                                        );
                                        return Box::pin(futures::stream::once(
                                            async move { Err(err) },
                                        ));
                                    }
                                };
                                let stream = match take_stream(&stream_body) {
                                    Ok(s) => s,
                                    Err(e) => {
                                        return Box::pin(futures::stream::once(
                                            async move { Err(e) },
                                        ));
                                    }
                                };
                                let input = camel_processor::stream_codec::StreamSplitInput {
                                    parent,
                                    stream,
                                    metadata: stream_body.metadata,
                                };
                                match camel_processor::stream_codec::resolve_split(
                                    &config.format,
                                    &input.metadata,
                                ) {
                                    Ok(
                                        camel_processor::stream_codec::ResolvedStreamSplit::Incremental(
                                            codec,
                                        ),
                                    ) => codec.split(input, config),
                                    Ok(
                                        camel_processor::stream_codec::ResolvedStreamSplit::MaterializedArchive(
                                            _,
                                        ),
                                    ) => Box::pin(futures::stream::once(async {
                                        Err(CamelError::ProcessorError(
                                            "materialized archive formats require Body::Bytes, Body::Text, or Body::Stream"
                                                .into(),
                                        ))
                                    })),
                                    Err(e) => {
                                        Box::pin(futures::stream::once(async move { Err(e) }))
                                    }
                                }
                            }
                        }
                    },
                );

                let segment = camel_processor::StreamingSplitSegment {
                    expression,
                    body: body_segment,
                    aggregation,
                    stop_on_exception,
                };
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(segment)),
                    body_contract: None,
                    lifecycle: pack_lifecycles(lifecycles),
                    label: None,
                }))
            }

            // ── Aggregate ──
            BuilderStep::Aggregate { config } => {
                let (late_tx, _late_rx) = mpsc::channel(256);
                let registry: SharedLanguageRegistry =
                    Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
                let cancel = CancellationToken::new();
                let mut svc =
                    camel_processor::AggregatorService::new(config, late_tx, registry, cancel);
                // Queue-depth visibility: the TTL-sweep pass reports the
                // buffered group count under `aggregator:<route>`.
                if let Some(route_id) = ctx.route_id {
                    svc =
                        svc.with_queue_metrics(ctx.rt.metrics(), format!("aggregator:{route_id}"));
                }
                let lifecycle: Arc<dyn StepLifecycle> = Arc::new(svc.clone());
                Ok(CompileOutcome::Matched(CompiledStep::Process {
                    kind_hint: SpanKindHint::Internal,
                    processor: BoxProcessor::new(svc),
                    body_contract: None,
                    lifecycle: Some(lifecycle),
                    label: None,
                    to_uri: None,
                }))
            }

            // ── Multicast ──
            BuilderStep::Multicast { config, steps } => {
                let mut branch_segments: Vec<camel_api::OutcomeSegment> = Vec::new();
                let mut all_lifecycles = Vec::new();
                for step in steps {
                    let (sub_segments, lifecycles) =
                        ctx.compile_children_segments(vec![step], registry)?;
                    all_lifecycles.extend(lifecycles);
                    branch_segments.push(compose_outcome_segment(sub_segments));
                }
                let strategy = config.aggregation.clone();
                let aggregator: Arc<dyn Fn(Vec<Exchange>) -> Exchange + Send + Sync> = Arc::new(
                    move |outputs| match &strategy {
                        MulticastStrategy::CollectAll => {
                            let bodies: Vec<Value> = outputs
                                .iter()
                                .map(|ex| match &ex.input.body {
                                    Body::Text(s) => Value::String(s.clone()),
                                    Body::Json(v) => v.clone(),
                                    Body::Xml(s) => Value::String(s.clone()),
                                    Body::Bytes(b) => {
                                        Value::String(String::from_utf8_lossy(b).into_owned())
                                    }
                                    Body::Stream(s) => serde_json::json!({
                                        "_stream": {
                                            "origin": s.metadata.origin,
                                            "placeholder": true,
                                            "hint": "Materialize exchange body with .into_bytes() before multicast aggregation"
                                        }
                                    }),
                                    // Empty and future variants contribute no extractable value.
                                    _ => Value::Null,
                                })
                                .collect();
                            let mut out = Exchange::default();
                            out.input.body = Body::Json(Value::Array(bodies));
                            out
                        }
                        MulticastStrategy::Original => {
                            outputs.into_iter().next().unwrap_or_default()
                        }
                        MulticastStrategy::Custom(fold_fn) => {
                            let mut iter = outputs.into_iter();
                            let first = iter.next().unwrap_or_default();
                            iter.fold(first, |acc, next| fold_fn(acc, next))
                        }
                        // LastWins and any future variant return the last output.
                        _ => outputs.into_iter().last().unwrap_or_default(),
                    },
                );
                let segment = camel_processor::MulticastSegment {
                    branches: branch_segments,
                    parallel: config.parallel,
                    parallel_limit: config.parallel_limit,
                    stop_on_exception: config.stop_on_exception,
                    timeout: config.timeout,
                    aggregator,
                };
                Ok(CompileOutcome::Matched(CompiledStep::Segment {
                    segment: camel_api::OutcomeSegment::new(Box::new(segment)),
                    body_contract: None,
                    lifecycle: pack_lifecycles(all_lifecycles),
                    label: None,
                }))
            }

            _ => Ok(CompileOutcome::NotHandled(step)),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use camel_api::{AggregatorConfig, ProducerContext, StepShutdownReason};
    use camel_bean::BeanRegistry;
    use camel_component_api::{
        ComponentContext, NoOpComponentContext, RuntimeObservability,
        test_support::NoopRuntimeObservability,
    };
    use std::collections::HashMap;
    use std::sync::Mutex;

    use crate::lifecycle::adapters::step_compilers::build_registry;
    use crate::lifecycle::adapters::step_resolution::FunctionStagingMode;
    use crate::lifecycle::application::route_definition::LanguageExpressionDef;
    use camel_api::{BoxProcessorExt, OpaqueProcessor, PipelineOutcome};
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// Language registry with `SimpleLanguage` under key `"simple"` —
    /// follows the `languages_with_simple` pattern from `step_resolution.rs`.
    fn languages_with_simple() -> SharedLanguageRegistry {
        let mut map: HashMap<String, Arc<dyn camel_language_api::Language>> = HashMap::new();
        map.insert(
            "simple".to_string(),
            Arc::new(camel_language_simple::SimpleLanguage::new()),
        );
        Arc::new(std::sync::Mutex::new(map))
    }

    /// Shared context builder — follows the pattern from `mod.rs::dispatch_tests::ctx`.
    #[allow(clippy::too_many_arguments)]
    fn test_ctx<'a>(
        pc: &'a ProducerContext,
        rt: Arc<dyn RuntimeObservability>,
        languages: &'a SharedLanguageRegistry,
        beans: &'a Arc<Mutex<BeanRegistry>>,
        component_ctx: Arc<dyn ComponentContext>,
        staging: &'a FunctionStagingMode,
        idempotent_repositories: &'a crate::IdempotentRegistry,
        claim_check_repositories: &'a crate::ClaimCheckRegistry,
        cache_repositories: &'a crate::CacheRegistry,
    ) -> CompilationContext<'a> {
        CompilationContext {
            producer_ctx: pc,
            rt,
            languages,
            beans,
            function_invoker: None,
            component_ctx,
            route_id: None,
            staging_mode: staging,
            idempotent_repositories,
            claim_check_repositories,
            cache_repositories,
            intercept: crate::intercept::InterceptRules::default(),
        }
    }

    #[test]
    fn aggregate_step_registers_lifecycle_handle() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = test_ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        // Build a valid AggregatorConfig: correlate_by provides defaults
        // (max_buckets=10_000, bucket_ttl=300s); add a completion condition.
        let config = AggregatorConfig::correlate_by("id")
            .complete_when_size(1)
            .build()
            .expect("valid aggregator config");

        let aggregate_step = BuilderStep::Aggregate { config };

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::SplittingCompiler));

        let result = reg
            .compile_step(aggregate_step, 0, &ctx)
            .expect("compilation should succeed")
            .expect("should match Aggregate");

        match result {
            CompiledStep::Process { lifecycle, .. } => {
                assert!(
                    lifecycle.is_some(),
                    "Aggregate CompiledStep should have a lifecycle handle"
                );
            }
            other => panic!("Expected CompiledStep::Process, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn aggregate_dsl_shutdown_drives_lifecycle() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = test_ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        let config = AggregatorConfig::correlate_by("id")
            .complete_when_size(1)
            .build()
            .expect("valid aggregator config");

        let aggregate_step = BuilderStep::Aggregate { config };

        let mut reg = StepCompilerRegistry::new();
        reg.register(Box::new(super::SplittingCompiler));

        let result = reg
            .compile_step(aggregate_step, 0, &ctx)
            .expect("compilation should succeed")
            .expect("should match Aggregate");

        let lifecycle = match result {
            CompiledStep::Process { lifecycle, .. } => {
                lifecycle.expect("Aggregate should have a lifecycle handle")
            }
            other => panic!("Expected CompiledStep::Process, got {other:?}"),
        };

        let result = lifecycle.shutdown(StepShutdownReason::RouteStop).await;
        assert!(result.is_ok(), "shutdown should return Ok(())");
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_declarative_split_non_text_non_array_fails() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages = languages_with_simple();
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = test_ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        // Body segment records invocations; the count must stay at zero when
        // the split expression fails (no cloned-fragment fallback).
        let invocations = Arc::new(AtomicUsize::new(0));
        let counter = invocations.clone();
        let split_step = BuilderStep::DeclarativeSplit {
            expression: LanguageExpressionDef {
                language: "simple".into(),
                source: "${header.num}".into(),
            },
            aggregation: camel_api::splitter::AggregationStrategy::LastWins,
            parallel: false,
            parallel_limit: None,
            stop_on_exception: false,
            steps: vec![BuilderStep::Processor(OpaqueProcessor(
                BoxProcessor::from_fn(move |ex: Exchange| {
                    counter.fetch_add(1, Ordering::SeqCst);
                    Box::pin(async move { Ok(ex) })
                }),
            ))],
        };

        let reg = build_registry();

        let result = reg
            .compile_step(split_step, 0, &ctx)
            .expect("compilation should succeed")
            .expect("should match DeclarativeSplit");

        let mut segment = match result {
            CompiledStep::Segment { segment, .. } => segment,
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        };

        let mut exchange = Exchange::new(camel_api::Message::new("ignored"));
        exchange.input.set_header("num", Value::Number(1.into()));

        let outcome = segment.run(exchange).await;
        match outcome {
            PipelineOutcome::Failed(err) => {
                let msg = err.to_string();
                assert!(
                    matches!(err, CamelError::TypeConversionFailed(_)),
                    "expected TypeConversionFailed, got: {msg}"
                );
                assert!(msg.contains("declarative split"), "message: {msg}");
                assert!(msg.contains("number"), "message: {msg}");
                assert!(msg.contains("text or array"), "message: {msg}");
                assert!(
                    msg.contains("add an unmarshal step before split"),
                    "message: {msg}"
                );
            }
            other => panic!(
                "expected PipelineOutcome::Failed, got success={}",
                other.is_success()
            ),
        }
        assert_eq!(
            invocations.load(Ordering::SeqCst),
            0,
            "body segment must not run when the split expression fails"
        );
    }

    #[tokio::test]
    async fn test_declarative_stream_split_non_stream_body_typed_error() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = test_ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );

        let split_step = BuilderStep::DeclarativeStreamSplit {
            stream_config: camel_api::StreamSplitConfig {
                format: StreamSplitFormat::Ndjson,
                ..Default::default()
            },
            aggregation: camel_api::splitter::AggregationStrategy::LastWins,
            stop_on_exception: false,
            steps: vec![],
        };

        let reg = build_registry();

        let result = reg
            .compile_step(split_step, 0, &ctx)
            .expect("compilation should succeed")
            .expect("should match DeclarativeStreamSplit");

        let mut segment = match result {
            CompiledStep::Segment { segment, .. } => segment,
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        };

        let exchange = Exchange::new(camel_api::Message::new(Body::Text("x".into())));

        // The first pull of the compiled expression surfaces the typed error
        // (the segment translates it to PipelineOutcome::Failed).
        let outcome = segment.run(exchange).await;
        match outcome {
            PipelineOutcome::Failed(err) => {
                let msg = err.to_string();
                assert!(
                    matches!(err, CamelError::TypeConversionFailed(_)),
                    "expected TypeConversionFailed, got: {msg}"
                );
                assert!(msg.contains("streaming split"), "message: {msg}");
                assert!(msg.contains("text"), "message: {msg}");
                assert!(msg.contains("stream"), "message: {msg}");
                assert!(
                    msg.contains("add an unmarshal step before split"),
                    "message: {msg}"
                );
            }
            other => panic!(
                "expected PipelineOutcome::Failed, got success={}",
                other.is_success()
            ),
        }
    }

    /// Build one 512-byte ustar header with a valid checksum for a regular
    /// file entry, mirroring the shape produced by the `tar` crate.
    fn tar_header(name: &str, size: u64) -> [u8; 512] {
        let mut h = [0u8; 512];
        h[..name.len()].copy_from_slice(name.as_bytes());
        h[124..136].copy_from_slice(format!("{size:011o}\0").as_bytes());
        // Checksum field must be spaces while the checksum is computed.
        h[148..156].copy_from_slice(b"        ");
        h[156] = b'0'; // regular file
        h[257..263].copy_from_slice(b"ustar\0");
        h[263..265].copy_from_slice(b"00");
        let sum: u32 = h.iter().map(|&b| u32::from(b)).sum();
        h[148..156].copy_from_slice(format!("{sum:06o}\0 ").as_bytes());
        h
    }

    /// Assemble regular-file entries and the two-block end marker into an
    /// in-memory TAR archive.
    fn tar_archive(entries: &[(&str, &[u8])]) -> Bytes {
        let mut buf = Vec::new();
        for (name, data) in entries {
            buf.extend_from_slice(&tar_header(name, data.len() as u64));
            buf.extend_from_slice(data);
            let pad = (512 - data.len() % 512) % 512;
            buf.resize(buf.len() + pad, 0);
        }
        buf.extend_from_slice(&[0u8; 1024]);
        Bytes::from(buf)
    }

    /// Compile a `DeclarativeStreamSplit` step for `format` through the full
    /// step-compiler registry and return the resulting segment.
    fn compile_stream_split(
        ctx: &CompilationContext<'_>,
        registry: &StepCompilerRegistry,
        format: StreamSplitFormat,
    ) -> camel_api::OutcomeSegment {
        let split_step = BuilderStep::DeclarativeStreamSplit {
            stream_config: camel_api::StreamSplitConfig {
                format,
                ..Default::default()
            },
            aggregation: camel_api::splitter::AggregationStrategy::LastWins,
            stop_on_exception: false,
            steps: vec![],
        };
        match registry
            .compile_step(split_step, 0, ctx)
            .expect("compilation should succeed")
            .expect("should match DeclarativeStreamSplit")
        {
            CompiledStep::Segment { segment, .. } => segment,
            other => panic!("Expected CompiledStep::Segment, got {other:?}"),
        }
    }

    /// Assert the aggregated outcome is the last TAR fragment with the
    /// splitter's entry metadata and the materialized marker property.
    fn assert_last_tar_fragment(ex: &Exchange) {
        assert_eq!(
            ex.input
                .headers
                .get(camel_processor::tar_splitter::CAMEL_TAR_ENTRY_NAME),
            Some(&Value::String("b.txt".to_string()))
        );
        assert_eq!(
            ex.input
                .headers
                .get(camel_processor::tar_splitter::CAMEL_TAR_ENTRY_INDEX),
            Some(&Value::from(1u64))
        );
        match &ex.input.body {
            Body::Bytes(b) => assert_eq!(b.as_ref(), b"world"),
            other => panic!("expected Body::Bytes, got {other:?}"),
        }
        assert_eq!(
            ex.property("CamelSplitMaterialized"),
            Some(&Value::Bool(true))
        );
    }

    #[tokio::test]
    async fn tar_stream_split_compiles_from_bytes_and_stream() {
        let pc = ProducerContext::default();
        let rt: Arc<dyn RuntimeObservability> = Arc::new(NoopRuntimeObservability);
        let languages: SharedLanguageRegistry = Arc::new(Mutex::new(HashMap::new()));
        let beans: Arc<Mutex<BeanRegistry>> = Arc::new(Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent_repositories = crate::IdempotentRegistry::new();
        let claim_check_repositories = crate::ClaimCheckRegistry::new();
        let cache_repositories = crate::CacheRegistry::new();

        let ctx = test_ctx(
            &pc,
            rt,
            &languages,
            &beans,
            component_ctx,
            &staging,
            &idempotent_repositories,
            &claim_check_repositories,
            &cache_repositories,
        );
        let reg = build_registry();

        let tar_bytes = tar_archive(&[("a.txt", b"hello"), ("b.txt", b"world")]);

        // -- Byte body: TAR splitter selected on the materialized path. --
        let mut segment = compile_stream_split(&ctx, &reg, StreamSplitFormat::Tar);
        let exchange = Exchange::new(camel_api::Message::new(Body::Bytes(tar_bytes.clone())));
        let outcome = segment.run(exchange).await;
        match outcome {
            PipelineOutcome::Completed(ex) => assert_last_tar_fragment(&ex),
            other => panic!(
                "expected PipelineOutcome::Completed, got success={}",
                other.is_success()
            ),
        }

        // -- Stream body: TAR splitter selected on the bounded stream path.
        //    Chunks are misaligned with the 512-byte record size so the
        //    collection must reassemble them before splitting. --
        let chunks = futures::stream::iter(vec![
            Ok(tar_bytes.slice(..1000)),
            Ok(tar_bytes.slice(1000..)),
        ]);
        let mut segment = compile_stream_split(&ctx, &reg, StreamSplitFormat::Tar);
        let exchange = Exchange::new(camel_api::Message::new(Body::Stream(
            camel_api::StreamBody {
                stream: Arc::new(tokio::sync::Mutex::new(Some(Box::pin(chunks)))),
                metadata: camel_api::StreamMetadata::default(),
            },
        )));
        let outcome = segment.run(exchange).await;
        match outcome {
            PipelineOutcome::Completed(ex) => assert_last_tar_fragment(&ex),
            other => panic!(
                "expected PipelineOutcome::Completed, got success={}",
                other.is_success()
            ),
        }

        // -- Unknown format remains rejected: TAR is a materialized archive
        //    format, so the incremental codec path must not accept it. --
        let err =
            match camel_processor::stream_codec::resolve_incremental_codec(&StreamSplitFormat::Tar)
            {
                Err(e) => e,
                Ok(_) => panic!("Tar is a materialized archive format, not an incremental codec"),
            };
        assert!(
            err.to_string().contains("materialized archive format"),
            "message: {err}"
        );

        // -- TAR.GZ with an unsupported body fails closed instead of
        //    silently dispatching. --
        let mut segment = compile_stream_split(&ctx, &reg, StreamSplitFormat::TarGz);
        let exchange = Exchange::new(camel_api::Message::new(Body::Json(Value::Null)));
        let outcome = segment.run(exchange).await;
        match outcome {
            PipelineOutcome::Failed(err) => {
                let msg = err.to_string();
                assert!(
                    msg.contains("TAR.GZ split requires Body::Bytes, Body::Text, or Body::Stream"),
                    "message: {msg}"
                );
            }
            other => panic!(
                "expected PipelineOutcome::Failed, got success={}",
                other.is_success()
            ),
        }
    }
}