camel_processor/

splitter.rs

use futures::future::join_all;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use tokio::sync::Semaphore;
use tokio_util::sync::CancellationToken;
use tower::Service;

use camel_api::{
    AggregationStrategy, Body, BoxProcessor, CamelError, Exchange, SplitSource, SplitterConfig,
    Value,
};

// ── Metadata property keys ─────────────────────────────────────────────

/// Property key for the zero-based index of a fragment within the split.
pub const CAMEL_SPLIT_INDEX: &str = "CamelSplitIndex";
/// Property key for the total number of fragments produced by the split.
pub const CAMEL_SPLIT_SIZE: &str = "CamelSplitSize";
/// Property key indicating whether this fragment is the last one.
pub const CAMEL_SPLIT_COMPLETE: &str = "CamelSplitComplete";

// ── SplitterService ────────────────────────────────────────────────────

/// Tower Service implementing the Splitter EIP.
///
/// Splits an incoming exchange into fragments via a configurable expression,
/// processes each fragment through a sub-pipeline, and aggregates the results.
///
/// **DoS bound (R3-M4):** the eager splitter materializes the whole fragment
/// `Vec` before processing. `SplitterConfig::max_fragments` (default 100_000)
/// rejects a split that would explode memory. For unbounded or lazy byte-stream
/// input, prefer `StreamingSplitterService`, which processes fragments as they
/// arrive and never materializes the full set.
///
/// **Note:** In parallel mode, `stop_on_exception` only affects the aggregation
/// phase. All spawned fragments run to completion because `join_all` cannot
/// cancel in-flight futures. Sequential mode stops processing immediately.
#[derive(Clone)]
pub struct SplitterService {
    expression: SplitSource,
    sub_pipeline: BoxProcessor,
    aggregation: AggregationStrategy,
    parallel: bool,
    parallel_limit: Option<usize>,
    stop_on_exception: bool,
    max_fragments: usize,
    cancel_token: CancellationToken,
}

impl SplitterService {
    /// Create a new `SplitterService` from a [`SplitterConfig`] and a sub-pipeline.
    pub fn new(config: SplitterConfig, sub_pipeline: BoxProcessor) -> Result<Self, CamelError> {
        config.validate()?;
        Ok(Self {
            expression: config.expression,
            sub_pipeline,
            aggregation: config.aggregation,
            parallel: config.parallel,
            parallel_limit: config.parallel_limit,
            stop_on_exception: config.stop_on_exception,
            max_fragments: config.max_fragments,
            cancel_token: CancellationToken::new(),
        })
    }

    /// Cancel all in-flight parallel tasks and prevent new ones from starting.
    pub fn cancel(&self) {
        self.cancel_token.cancel();
    }

    /// Check whether the splitter has been cancelled.
    pub fn is_cancelled(&self) -> bool {
        self.cancel_token.is_cancelled()
    }
}

impl Service<Exchange> for SplitterService {
    type Response = Exchange;
    type Error = CamelError;
    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.sub_pipeline.poll_ready(cx)
    }

    fn call(&mut self, exchange: Exchange) -> Self::Future {
        let original = exchange.clone();
        let expression = self.expression.clone();
        let sub_pipeline = self.sub_pipeline.clone();
        let aggregation = self.aggregation.clone();
        let parallel = self.parallel;
        let parallel_limit = self.parallel_limit;
        let stop_on_exception = self.stop_on_exception;
        let max_fragments = self.max_fragments;
        let cancel_token = self.cancel_token.clone();

        Box::pin(async move {
            // Split the exchange into fragments. A typed expression error
            // (e.g. wrong body type) fails loud instead of degrading to an
            // empty-fragment pass-through.
            let mut fragments = expression.split(&exchange).await?;

            // If no fragments were produced, return the original exchange.
            if fragments.is_empty() {
                return Ok(original);
            }

            // R3-M4: the eager splitter materializes the whole Vec before
            // processing — cap the fragment count to bound memory.
            if fragments.len() > max_fragments {
                return Err(CamelError::ProcessorError(format!(
                    "Splitter produced {} fragments, exceeding max_fragments {}",
                    fragments.len(),
                    max_fragments
                )));
            }

            let total = fragments.len();

            // Set metadata on each fragment.
            for (i, frag) in fragments.iter_mut().enumerate() {
                frag.set_property(CAMEL_SPLIT_INDEX, Value::from(i as u64));
                frag.set_property(CAMEL_SPLIT_SIZE, Value::from(total as u64));
                frag.set_property(CAMEL_SPLIT_COMPLETE, Value::Bool(i == total - 1));
            }

            // Check cancellation before processing.
            if cancel_token.is_cancelled() {
                return Err(CamelError::ProcessorError(
                    "Splitter cancelled, dropping exchange".to_string(),
                ));
            }

            // Process fragments through the sub-pipeline.
            let results = if parallel {
                process_parallel(
                    fragments,
                    sub_pipeline,
                    parallel_limit,
                    stop_on_exception,
                    cancel_token,
                )
                .await
            } else {
                process_sequential(fragments, sub_pipeline, stop_on_exception).await
            };

            // Aggregate the results.
            aggregate(results, original, aggregation)
        })
    }
}

// ── Sequential processing ──────────────────────────────────────────────

async fn process_sequential(
    fragments: Vec<Exchange>,
    sub_pipeline: BoxProcessor,
    stop_on_exception: bool,
) -> Vec<Result<Exchange, CamelError>> {
    let mut results = Vec::with_capacity(fragments.len());

    for fragment in fragments {
        let mut pipeline = sub_pipeline.clone();
        match tower::ServiceExt::ready(&mut pipeline).await {
            Err(e) => {
                results.push(Err(e));
                if stop_on_exception {
                    break;
                }
            }
            Ok(svc) => {
                let result = svc.call(fragment).await;
                let is_err = result.is_err();
                results.push(result);
                if stop_on_exception && is_err {
                    break;
                }
            }
        }
    }

    results
}

// ── Parallel processing ────────────────────────────────────────────────

async fn process_parallel(
    fragments: Vec<Exchange>,
    sub_pipeline: BoxProcessor,
    parallel_limit: Option<usize>,
    _stop_on_exception: bool,
    cancel_token: CancellationToken,
) -> Vec<Result<Exchange, CamelError>> {
    let semaphore = parallel_limit.map(|limit| Arc::new(Semaphore::new(limit)));

    let futures: Vec<_> = fragments
        .into_iter()
        .map(|fragment| {
            let mut pipeline = sub_pipeline.clone();
            let sem = semaphore.clone();
            let cancel = cancel_token.clone();
            async move {
                // Check cancellation before acquiring semaphore.
                if cancel.is_cancelled() {
                    return Err(CamelError::ProcessorError("Splitter cancelled".to_string()));
                }

                // Acquire semaphore permit if a limit is set.
                let _permit = match &sem {
                    Some(s) => {
                        tokio::select! {
                            result = s.acquire() => {
                                Some(result.map_err(|e| {
                                    CamelError::ProcessorError(format!("semaphore error: {e}"))
                                })?)
                            }
                            _ = cancel.cancelled() => {
                                return Err(CamelError::ProcessorError(
                                    "Splitter cancelled while waiting for semaphore".to_string(),
                                ));
                            }
                        }
                    }
                    None => None,
                };

                // Check cancellation again after acquiring.
                if cancel.is_cancelled() {
                    return Err(CamelError::ProcessorError("Splitter cancelled".to_string()));
                }

                tokio::select! {
                    result = async {
                        tower::ServiceExt::ready(&mut pipeline).await?;
                        pipeline.call(fragment).await
                    } => result,
                    _ = cancel.cancelled() => {
                        Err(CamelError::ProcessorError(
                            "Splitter cancelled during processing".to_string(),
                        ))
                    }
                }
            }
        })
        .collect();

    join_all(futures).await
}

// ── Aggregation ────────────────────────────────────────────────────────

fn aggregate(
    results: Vec<Result<Exchange, CamelError>>,
    original: Exchange,
    strategy: AggregationStrategy,
) -> Result<Exchange, CamelError> {
    match strategy {
        AggregationStrategy::LastWins => {
            // Return the last result (error or success).
            results.into_iter().last().unwrap_or_else(|| Ok(original))
        }
        AggregationStrategy::CollectAll => {
            // Collect all bodies into a JSON array. Errors propagate.
            let mut bodies = Vec::new();
            for result in results {
                let ex = result?;
                let value = 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 aggregation if content needed"
                        }
                    }),
                    // Empty and future variants contribute no extractable value.
                    _ => Value::Null,
                };
                bodies.push(value);
            }
            let mut out = original;
            out.input.body = Body::Json(Value::Array(bodies));
            Ok(out)
        }
        AggregationStrategy::Custom(fold_fn) => {
            // Fold using the custom function, starting from the first result.
            let mut iter = results.into_iter();
            let first = iter.next().unwrap_or_else(|| Ok(original.clone()))?;
            iter.try_fold(first, |acc, next_result| {
                let next = next_result?;
                Ok(fold_fn(acc, next))
            })
        }
        // Original and any future variant return the original exchange.
        _ => Ok(original),
    }
}

// ── Tests ──────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use camel_api::{BoxProcessorExt, Message};
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use tower::ServiceExt;

    // ── Test helpers ───────────────────────────────────────────────────

    fn passthrough_pipeline() -> BoxProcessor {
        BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }))
    }

    fn uppercase_pipeline() -> BoxProcessor {
        BoxProcessor::from_fn(|mut ex: Exchange| {
            Box::pin(async move {
                if let Body::Text(s) = &ex.input.body {
                    ex.input.body = Body::Text(s.to_uppercase());
                }
                Ok(ex)
            })
        })
    }

    fn failing_pipeline() -> BoxProcessor {
        BoxProcessor::from_fn(|_ex| {
            Box::pin(async { Err(CamelError::ProcessorError("boom".into())) })
        })
    }

    fn fail_on_nth(n: usize) -> BoxProcessor {
        let count = Arc::new(AtomicUsize::new(0));
        BoxProcessor::from_fn(move |ex: Exchange| {
            let count = Arc::clone(&count);
            Box::pin(async move {
                let c = count.fetch_add(1, Ordering::SeqCst);
                if c == n {
                    Err(CamelError::ProcessorError(format!("fail on {c}")))
                } else {
                    Ok(ex)
                }
            })
        })
    }

    fn make_exchange(text: &str) -> Exchange {
        Exchange::new(Message::new(text))
    }

    #[tokio::test]
    async fn split_sync_builtin_still_works() {
        // The Sync arm of SplitSource must keep the built-in expressions
        // compiling and behaving unchanged.
        let config = SplitterConfig::new(camel_api::split_body_lines());
        let mut svc = SplitterService::new(config, uppercase_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await
            .unwrap();
        assert_eq!(result.input.body.as_text(), Some("C"));
    }

    #[tokio::test]
    async fn split_async_error_propagates_before_fragments() {
        use camel_api::{BoxValueFuture, SplitSource};

        let pipeline_calls = Arc::new(AtomicUsize::new(0));
        let calls_clone = pipeline_calls.clone();
        let pipeline = BoxProcessor::from_fn(move |ex: Exchange| {
            calls_clone.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move { Ok(ex) })
        });

        let config = SplitterConfig::new(SplitSource::Async(Arc::new(|_: &Exchange| {
            Box::pin(async { Err(CamelError::ProcessorError("split boom".into())) })
                as BoxValueFuture
        })));
        let mut svc = SplitterService::new(config, pipeline).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await;

        assert!(
            matches!(result, Err(CamelError::ProcessorError(ref e)) if e.contains("split boom")),
            "an async split-expression failure must propagate"
        );
        assert_eq!(
            pipeline_calls.load(Ordering::SeqCst),
            0,
            "no fragment may reach the sub-pipeline when the split expression fails"
        );
    }

    #[test]
    fn test_splitter_zero_parallel_limit_rejected() {
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .parallel(true)
            .parallel_limit(0);
        let result = SplitterService::new(config, passthrough_pipeline());
        assert!(result.is_err(), "zero parallel_limit should return Err");
    }

    // ── 1. Sequential + LastWins ───────────────────────────────────────

    #[tokio::test]
    async fn test_split_sequential_last_wins() {
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .aggregation(AggregationStrategy::LastWins);
        let mut svc = SplitterService::new(config, uppercase_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await
            .unwrap();
        assert_eq!(result.input.body.as_text(), Some("C"));
    }

    // ── 2. Sequential + CollectAll ─────────────────────────────────────

    #[tokio::test]
    async fn test_split_sequential_collect_all() {
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .aggregation(AggregationStrategy::CollectAll);
        let mut svc = SplitterService::new(config, uppercase_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await
            .unwrap();
        let expected = serde_json::json!(["A", "B", "C"]);
        match &result.input.body {
            Body::Json(v) => assert_eq!(*v, expected),
            other => panic!("expected JSON body, got {other:?}"),
        }
    }

    // ── 3. Sequential + Original ───────────────────────────────────────

    #[tokio::test]
    async fn test_split_sequential_original() {
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .aggregation(AggregationStrategy::Original);
        let mut svc = SplitterService::new(config, uppercase_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await
            .unwrap();
        // Original body should be unchanged.
        assert_eq!(result.input.body.as_text(), Some("a\nb\nc"));
    }

    // ── 4. Sequential + Custom aggregation ─────────────────────────────

    #[tokio::test]
    async fn test_split_sequential_custom_aggregation() {
        let joiner: Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync> =
            Arc::new(|mut acc: Exchange, next: Exchange| {
                let acc_text = acc.input.body.as_text().unwrap_or("").to_string();
                let next_text = next.input.body.as_text().unwrap_or("").to_string();
                acc.input.body = Body::Text(format!("{acc_text}+{next_text}"));
                acc
            });

        let config = SplitterConfig::new(camel_api::split_body_lines())
            .aggregation(AggregationStrategy::Custom(joiner));
        let mut svc = SplitterService::new(config, uppercase_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await
            .unwrap();
        assert_eq!(result.input.body.as_text(), Some("A+B+C"));
    }

    // ── 5. Stop on exception ───────────────────────────────────────────

    #[tokio::test]
    async fn test_split_stop_on_exception() {
        // 5 fragments, fail on the 2nd (index 1), stop=true
        let config = SplitterConfig::new(camel_api::split_body_lines()).stop_on_exception(true);
        let mut svc = SplitterService::new(config, fail_on_nth(1)).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc\nd\ne"))
            .await;

        // LastWins is default, the last result should be the error from fragment 1.
        assert!(result.is_err(), "expected error due to stop_on_exception");
    }

    // ── 6. Continue on exception ───────────────────────────────────────

    #[tokio::test]
    async fn test_split_continue_on_exception() {
        // 3 fragments, fail on 2nd (index 1), stop=false, LastWins.
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .stop_on_exception(false)
            .aggregation(AggregationStrategy::LastWins);
        let mut svc = SplitterService::new(config, fail_on_nth(1)).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await;

        // LastWins: last fragment (index 2) succeeded.
        assert!(result.is_ok(), "last fragment should succeed");
    }

    // ── 7. Empty fragments ─────────────────────────────────────────────

    #[tokio::test]
    async fn test_split_empty_fragments() {
        // Body::Empty → no fragments → return original unchanged.
        let config = SplitterConfig::new(camel_api::split_body_lines());
        let mut svc = SplitterService::new(config, passthrough_pipeline()).unwrap();

        let mut ex = Exchange::new(Message::default()); // Body::Empty
        ex.set_property("marker", Value::Bool(true));

        let result = svc.ready().await.unwrap().call(ex).await.unwrap();
        assert!(result.input.body.is_empty());
        assert_eq!(result.property("marker"), Some(&Value::Bool(true)));
    }

    // ── 8. Metadata properties ─────────────────────────────────────────

    #[tokio::test]
    async fn test_split_metadata_properties() {
        // Use passthrough so we can inspect metadata on returned fragments.
        // CollectAll won't preserve metadata, so use a pipeline that records
        // the metadata into the body as JSON.
        let recorder = BoxProcessor::from_fn(|ex: Exchange| {
            Box::pin(async move {
                let idx = ex.property(CAMEL_SPLIT_INDEX).cloned();
                let size = ex.property(CAMEL_SPLIT_SIZE).cloned();
                let complete = ex.property(CAMEL_SPLIT_COMPLETE).cloned();
                let body = serde_json::json!({
                    "index": idx,
                    "size": size,
                    "complete": complete,
                });
                let mut out = ex;
                out.input.body = Body::Json(body);
                Ok(out)
            })
        });

        let config = SplitterConfig::new(camel_api::split_body_lines())
            .aggregation(AggregationStrategy::CollectAll);
        let mut svc = SplitterService::new(config, recorder).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("x\ny\nz"))
            .await
            .unwrap();

        let expected = serde_json::json!([
            {"index": 0, "size": 3, "complete": false},
            {"index": 1, "size": 3, "complete": false},
            {"index": 2, "size": 3, "complete": true},
        ]);
        match &result.input.body {
            Body::Json(v) => assert_eq!(*v, expected),
            other => panic!("expected JSON body, got {other:?}"),
        }
    }

    // ── 9. poll_ready delegates to sub-pipeline ────────────────────────

    #[tokio::test]
    async fn test_poll_ready_delegates_to_sub_pipeline() {
        use std::sync::atomic::AtomicBool;

        // A service that is initially not ready, then becomes ready.
        #[derive(Clone)]
        struct DelayedReady {
            ready: Arc<AtomicBool>,
        }

        impl Service<Exchange> for DelayedReady {
            type Response = Exchange;
            type Error = CamelError;
            type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

            fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
                if self.ready.load(Ordering::SeqCst) {
                    Poll::Ready(Ok(()))
                } else {
                    cx.waker().wake_by_ref();
                    Poll::Pending
                }
            }

            fn call(&mut self, exchange: Exchange) -> Self::Future {
                Box::pin(async move { Ok(exchange) })
            }
        }

        let ready_flag = Arc::new(AtomicBool::new(false));
        let inner = DelayedReady {
            ready: Arc::clone(&ready_flag),
        };
        let boxed: BoxProcessor = BoxProcessor::new(inner);

        let config = SplitterConfig::new(camel_api::split_body_lines());
        let mut svc = SplitterService::new(config, boxed).unwrap();

        // First poll should be Pending.
        let waker = futures::task::noop_waker();
        let mut cx = Context::from_waker(&waker);
        let poll = Pin::new(&mut svc).poll_ready(&mut cx);
        assert!(
            poll.is_pending(),
            "expected Pending when sub_pipeline not ready"
        );

        // Mark inner as ready.
        ready_flag.store(true, Ordering::SeqCst);

        let poll = Pin::new(&mut svc).poll_ready(&mut cx);
        assert!(
            matches!(poll, Poll::Ready(Ok(()))),
            "expected Ready after sub_pipeline becomes ready"
        );
    }

    // ── 10. Parallel basic ─────────────────────────────────────────────

    #[tokio::test]
    async fn test_split_parallel_basic() {
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .parallel(true)
            .aggregation(AggregationStrategy::CollectAll);
        let mut svc = SplitterService::new(config, uppercase_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await
            .unwrap();

        let expected = serde_json::json!(["A", "B", "C"]);
        match &result.input.body {
            Body::Json(v) => assert_eq!(*v, expected),
            other => panic!("expected JSON body, got {other:?}"),
        }
    }

    // ── 11. Parallel with limit ────────────────────────────────────────

    #[tokio::test]
    async fn test_split_parallel_with_limit() {
        use std::sync::atomic::AtomicUsize;

        let concurrent = Arc::new(AtomicUsize::new(0));
        let max_concurrent = Arc::new(AtomicUsize::new(0));

        let c = Arc::clone(&concurrent);
        let mc = Arc::clone(&max_concurrent);
        let pipeline = BoxProcessor::from_fn(move |ex: Exchange| {
            let c = Arc::clone(&c);
            let mc = Arc::clone(&mc);
            Box::pin(async move {
                let current = c.fetch_add(1, Ordering::SeqCst) + 1;
                // Record the high-water mark.
                mc.fetch_max(current, Ordering::SeqCst);
                // Yield to let other tasks run.
                tokio::task::yield_now().await;
                c.fetch_sub(1, Ordering::SeqCst);
                Ok(ex)
            })
        });

        let config = SplitterConfig::new(camel_api::split_body_lines())
            .parallel(true)
            .parallel_limit(2)
            .aggregation(AggregationStrategy::CollectAll);
        let mut svc = SplitterService::new(config, pipeline).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc\nd"))
            .await;
        assert!(result.is_ok());

        let observed_max = max_concurrent.load(Ordering::SeqCst);
        assert!(
            observed_max <= 2,
            "max concurrency was {observed_max}, expected <= 2"
        );
    }

    // ── 12. Parallel stop on exception ─────────────────────────────────

    #[tokio::test]
    async fn test_split_parallel_stop_on_exception() {
        let config = SplitterConfig::new(camel_api::split_body_lines())
            .parallel(true)
            .stop_on_exception(true);
        let mut svc = SplitterService::new(config, failing_pipeline()).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await;

        // All fragments fail; LastWins returns the last error.
        assert!(result.is_err(), "expected error when all fragments fail");
    }

    // ── 13. Stream body aggregation creates valid JSON ───────────────────

    #[tokio::test]
    async fn test_splitter_stream_bodies_creates_valid_json() {
        use bytes::Bytes;
        use camel_api::{StreamBody, StreamMetadata};
        use futures::stream;
        use tokio::sync::Mutex;

        let chunks = vec![Ok(Bytes::from("test"))];
        let stream_body = StreamBody {
            stream: Arc::new(Mutex::new(Some(Box::pin(stream::iter(chunks))))),
            metadata: StreamMetadata {
                origin: Some("kafka://topic/partition".to_string()),
                ..Default::default()
            },
        };

        let original = Exchange::new(Message {
            headers: Default::default(),
            body: Body::Empty,
        });

        let results = vec![Ok(Exchange::new(Message {
            headers: Default::default(),
            body: Body::Stream(stream_body),
        }))];

        let result = aggregate(results, original, AggregationStrategy::CollectAll);

        let exchange = result.expect("Expected Ok result");
        assert!(
            matches!(exchange.input.body, Body::Json(_)),
            "Expected Json body"
        );

        if let Body::Json(value) = exchange.input.body {
            let json_str = serde_json::to_string(&value).unwrap();
            let parsed: serde_json::Value = serde_json::from_str(&json_str).unwrap();

            assert!(parsed.is_array());
            let arr = parsed.as_array().unwrap();
            assert!(arr[0].is_object());
            assert!(arr[0]["_stream"].is_object());
            assert_eq!(arr[0]["_stream"]["origin"], "kafka://topic/partition");
            assert_eq!(arr[0]["_stream"]["placeholder"], true);
        }
    }

    #[tokio::test]
    async fn test_splitter_stream_with_none_origin_creates_valid_json() {
        use bytes::Bytes;
        use camel_api::{StreamBody, StreamMetadata};
        use futures::stream;
        use tokio::sync::Mutex;

        let chunks = vec![Ok(Bytes::from("test"))];
        let stream_body = StreamBody {
            stream: Arc::new(Mutex::new(Some(Box::pin(stream::iter(chunks))))),
            metadata: StreamMetadata {
                origin: None,
                ..Default::default()
            },
        };

        let original = Exchange::new(Message {
            headers: Default::default(),
            body: Body::Empty,
        });

        let results = vec![Ok(Exchange::new(Message {
            headers: Default::default(),
            body: Body::Stream(stream_body),
        }))];

        let result = aggregate(results, original, AggregationStrategy::CollectAll);

        let exchange = result.expect("Expected Ok result");
        assert!(
            matches!(exchange.input.body, Body::Json(_)),
            "Expected Json body"
        );

        if let Body::Json(value) = exchange.input.body {
            let json_str = serde_json::to_string(&value).unwrap();
            let parsed: serde_json::Value = serde_json::from_str(&json_str).unwrap();

            assert!(parsed.is_array());
            let arr = parsed.as_array().unwrap();
            assert!(arr[0].is_object());
            assert!(arr[0]["_stream"].is_object());
            assert_eq!(arr[0]["_stream"]["origin"], serde_json::Value::Null);
            assert_eq!(arr[0]["_stream"]["placeholder"], true);
        }
    }

    // ── 14. Parallel cancellation ──────────────────────────────────────

    #[tokio::test]
    async fn test_splitter_parallel_cancel_aborts_processing() {
        use std::sync::atomic::AtomicBool;

        let started = Arc::new(AtomicBool::new(false));

        let s = Arc::clone(&started);
        let pipeline = BoxProcessor::from_fn(move |ex: Exchange| {
            let s = Arc::clone(&s);
            Box::pin(async move {
                s.store(true, Ordering::SeqCst);
                // Long-running task that should be cancelled.
                tokio::time::sleep(std::time::Duration::from_secs(60)).await;
                Ok(ex)
            })
        });

        let config = SplitterConfig::new(camel_api::split_body_lines())
            .parallel(true)
            .aggregation(AggregationStrategy::LastWins);
        let svc = SplitterService::new(config, pipeline).unwrap();

        // Cancel before calling — call should return an error.
        svc.cancel();
        assert!(svc.is_cancelled());

        let mut svc_clone = svc.clone();
        let result = svc_clone
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc"))
            .await;

        assert!(result.is_err(), "cancelled splitter should return error");
    }

    // ── 15. Fragment count cap ─────────────────────────────────────────

    #[tokio::test]
    async fn test_splitter_rejects_fragment_flood() {
        // Expression that produces 5 fragments; cap at 2.
        let expression: camel_api::SplitExpression = std::sync::Arc::new(|_| {
            Ok((0..5)
                .map(|i| Exchange::new(Message::new(Body::Text(i.to_string()))))
                .collect::<Vec<_>>())
        });
        let cfg = SplitterConfig::new(expression).max_fragments(2);
        let passthrough = BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }));
        let mut svc = SplitterService::new(cfg, passthrough).unwrap();

        let ex = Exchange::new(Message::new(Body::Text("parent".into())));
        let result = svc.ready().await.unwrap().call(ex).await;
        let err = result.unwrap_err();
        assert!(
            format!("{err}").contains("max_fragments"),
            "error should mention max_fragments: {err}"
        );
    }

    // ── 16. Fragments have unique correlation IDs ──────────────────────
    // Inspiration: Camel SplitterTest C1/C4 — fragments must be distinguishable
    // downstream (e.g. for idempotency keys). fragment_exchange (camel-api)
    // assigns a fresh UUID per fragment; this test pins the invariant.

    #[tokio::test]
    async fn test_splitter_each_fragment_has_unique_correlation_id() {
        // Pipeline records each fragment's correlation_id into the body.
        let recorder = BoxProcessor::from_fn(|ex: Exchange| {
            Box::pin(async move {
                let id = ex.correlation_id().to_string();
                let mut out = ex;
                out.input.body = Body::Text(id);
                Ok(out)
            })
        });

        let config = SplitterConfig::new(camel_api::split_body_lines())
            .aggregation(AggregationStrategy::CollectAll);
        let mut svc = SplitterService::new(config, recorder).unwrap();

        let result = svc
            .ready()
            .await
            .unwrap()
            .call(make_exchange("a\nb\nc\nd"))
            .await
            .unwrap();

        let ids: Vec<String> = match &result.input.body {
            Body::Json(serde_json::Value::Array(arr)) => arr
                .iter()
                .map(|v| v.as_str().unwrap_or("").to_string())
                .collect(),
            other => panic!("expected JSON array of ids, got {other:?}"),
        };
        assert_eq!(ids.len(), 4, "should have 4 fragments");
        let unique: std::collections::HashSet<&String> = ids.iter().collect();
        assert_eq!(unique.len(), 4, "fragment correlation_ids must be unique");
    }

    // ── 17. split(body()) on a JSON Array body ─────────────────────────
    // Inspiration: Camel SplitterTest C11 — when body is already a collection,
    // split iterates elements without tokenizing. camel_api::split_body_json_array
    // provides this; verify the wiring through SplitterService.

    #[tokio::test]
    async fn test_splitter_json_array_body() {
        let recorder = BoxProcessor::from_fn(|ex: Exchange| {
            Box::pin(async move {
                let v = match &ex.input.body {
                    Body::Json(v) => v.clone(),
                    other => panic!("expected JSON fragment, got {other:?}"),
                };
                let mut out = ex;
                out.input.body = Body::Json(v);
                Ok(out)
            })
        });

        let config = SplitterConfig::new(camel_api::split_body_json_array())
            .aggregation(AggregationStrategy::CollectAll);
        let mut svc = SplitterService::new(config, recorder).unwrap();

        let msg = Message::new(Body::Json(serde_json::json!([1, 2, 3])));
        let parent = Exchange::new(msg);

        let result = svc.ready().await.unwrap().call(parent).await.unwrap();
        match &result.input.body {
            Body::Json(serde_json::Value::Array(arr)) => {
                assert_eq!(arr.len(), 3, "should split array into 3 fragments");
                assert_eq!(arr[0], 1);
                assert_eq!(arr[1], 2);
                assert_eq!(arr[2], 3);
            }
            other => panic!("expected JSON array body, got {other:?}"),
        }
    }

    // ── 18. Wrong body type fails loud ─────────────────────────────────

    #[tokio::test]
    async fn test_splitter_wrong_type_body_fails_loud() {
        // Splitting a Text body with body_json_array must return Err with
        // the typed message, not silently pass the original through.
        let config = SplitterConfig::new(camel_api::split_body_json_array());
        let mut svc = SplitterService::new(config, passthrough_pipeline()).unwrap();

        let ex = Exchange::new(Message::new(Body::Text("a,b".to_string())));
        let result = svc.ready().await.unwrap().call(ex).await;

        let err = result.expect_err("wrong body type must fail loud, not pass through");
        let msg = err.to_string();
        for needle in [
            "body_json_array",
            "text",
            "json (array)",
            "add an unmarshal step before split",
        ] {
            assert!(msg.contains(needle), "message '{msg}' missing '{needle}'");
        }
    }
}