camel_processor/
streaming_split_segment.rs

1//! ## Stop semantics (ADR-0025)
2//!
3//! This segment implements `OutcomePipeline` and propagates `PipelineOutcome::Stopped(ex)` with the exchange state intact (including mutations made inside the segment body before Stop fired). See ADR-0025 §3 (stopped-exchange-state-preservation invariant).
4
5use std::future::Future;
6use std::pin::Pin;
7use std::sync::Arc;
8
9use futures::{StreamExt, pin_mut};
10
11use camel_api::{
12    AggregationStrategy, Body, CamelError, Exchange, OutcomePipeline, OutcomeSegment,
13    PipelineOutcome, StreamingSplitExpression, Value,
14};
15
16use crate::split_segment::aggregate_completed;
17use crate::splitter::{CAMEL_SPLIT_COMPLETE, CAMEL_SPLIT_INDEX, CAMEL_SPLIT_SIZE};
18
19// ── StreamingSplitSegment (ADR-0025 OutcomePipeline) ─────────────────
20
21/// Outcome-aware StreamingSplit segment. Consumes a stream of fragment
22/// Exchanges via `expression(exchange)`, runs `body` on each sequentially.
23/// On Stop: return `Stopped(fragment_ex)` immediately; the `stream` local is
24/// dropped when this future returns, closing the underlying stream resource.
25/// Aggregation is SKIPPED when Stop fires (spec §5.3).
26///
27/// NOTE: No `CancellationToken` — the stream-drop path (route
28/// shutdown) is handled by Drop semantics on the source stream future.
29/// Adding an internal cancel token + `tokio::select!` adds dead complexity:
30/// nothing external can fire the token (it's private), and the same-future
31/// cancel-after-Stop branch can never win (we've already returned).
32pub struct StreamingSplitSegment {
33    /// Produces a stream of fragment exchanges from the incoming exchange.
34    pub expression: StreamingSplitExpression,
35    /// The sub-pipeline executed for each fragment.
36    pub body: OutcomeSegment,
37    /// Strategy for aggregating fragment results.
38    pub aggregation: AggregationStrategy,
39    /// Whether to stop processing on the first exception.
40    ///
41    /// When `true`, a `Failed` outcome from any fragment halts processing
42    /// immediately (matching legacy behavior). When `false`, the error is
43    /// collected and processing continues; the last error is propagated
44    /// after the stream is exhausted.
45    ///
46    /// `Stopped` outcomes always propagate immediately regardless of this
47    /// flag (per ADR-0025 §7 — Stop is successful control flow).
48    pub stop_on_exception: bool,
49}
50
51impl Clone for StreamingSplitSegment {
52    fn clone(&self) -> Self {
53        Self {
54            expression: Arc::clone(&self.expression),
55            body: self.body.clone(),
56            aggregation: self.aggregation.clone(),
57            stop_on_exception: self.stop_on_exception,
58        }
59    }
60}
61
62impl OutcomePipeline for StreamingSplitSegment {
63    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
64        Box::new(self.clone())
65    }
66
67    fn run<'a>(
68        &'a mut self,
69        exchange: Exchange,
70    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
71        let expression = Arc::clone(&self.expression);
72        let mut body = self.body.clone();
73        let aggregation = self.aggregation.clone();
74        let stop_on_exception = self.stop_on_exception;
75
76        Box::pin(async move {
77            let mut original = exchange.branch_clone();
78            if matches!(original.input.body, Body::Stream(_)) {
79                original.input.body = Body::Empty;
80            }
81            let stream = expression(exchange);
82            pin_mut!(stream);
83
84            let mut outputs: Vec<Exchange> = Vec::new();
85            let mut last_error: Option<CamelError> = None;
86            let mut index: u64 = 0;
87
88            // One-entry lookahead: the last fragment is only known once the
89            // following stream item arrives, which is what allows
90            // CamelSplitComplete and CamelSplitSize to be stamped on it.
91            let mut current = stream.next().await;
92
93            while let Some(frag_result) = current.take() {
94                let mut frag = match frag_result {
95                    Ok(f) => f,
96                    // Mid-stream expression errors hard-fail here regardless of
97                    // stop_on_exception; covered by the first-item type-error test
98                    // (stream_split_non_stream_body_surfaces_expression_error). If
99                    // this arm ever becomes index-aware, port the deleted service's
100                    // test_streaming_error_in_expression.
101                    Err(e) => return PipelineOutcome::Failed(e),
102                };
103
104                let next = stream.next().await;
105                let is_last = next.is_none();
106
107                frag.set_property(CAMEL_SPLIT_INDEX, Value::from(index));
108                frag.set_property(CAMEL_SPLIT_COMPLETE, Value::Bool(is_last));
109                // SPLIT_SIZE is unknown until end-of-stream, so it is only
110                // present on the last fragment.
111                if is_last {
112                    frag.set_property(CAMEL_SPLIT_SIZE, Value::from(index + 1));
113                }
114
115                match body.run(frag).await {
116                    PipelineOutcome::Completed(ex) => {
117                        outputs.push(ex);
118                        index += 1;
119                    }
120                    PipelineOutcome::Stopped(stopped_ex) => {
121                        // Stop: return immediately. The `stream` local is
122                        // dropped when this future returns, closing the
123                        // underlying stream resource.
124                        return PipelineOutcome::Stopped(stopped_ex);
125                    }
126                    PipelineOutcome::Failed(err) => {
127                        if stop_on_exception {
128                            return PipelineOutcome::Failed(err);
129                        }
130                        // stop_on_exception=false: collect error, continue
131                        // processing remaining fragments.
132                        last_error = Some(err);
133                        index += 1;
134                    }
135                }
136
137                current = next;
138            }
139
140            // Stream exhausted — if any errors were collected, propagate last.
141            if let Some(err) = last_error {
142                return PipelineOutcome::Failed(err);
143            }
144            PipelineOutcome::Completed(aggregate_completed(outputs, original, aggregation))
145        })
146    }
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use camel_api::{Body, CamelError, Message, Value};
153    use std::task::{Context as TaskContext, Poll};
154    use tokio::sync::Mutex;
155
156    type BoxedOutcomeFuture = Pin<Box<dyn Future<Output = PipelineOutcome> + Send>>;
157    type RecordedMetadata = Arc<Mutex<Vec<(Option<Value>, Option<Value>)>>>;
158
159    /// Closure-backed `OutcomePipeline` used by the ported segment tests.
160    #[derive(Clone)]
161    struct FnBody {
162        f: Arc<dyn Fn(Exchange) -> BoxedOutcomeFuture + Send + Sync>,
163    }
164
165    impl OutcomePipeline for FnBody {
166        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
167            Box::new(self.clone())
168        }
169        fn run<'a>(
170            &'a mut self,
171            exchange: Exchange,
172        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
173            let f = Arc::clone(&self.f);
174            Box::pin(async move { f(exchange).await })
175        }
176    }
177
178    fn body_from_fn<F>(f: F) -> OutcomeSegment
179    where
180        F: Fn(Exchange) -> BoxedOutcomeFuture + Send + Sync + 'static,
181    {
182        OutcomeSegment::new(Box::new(FnBody { f: Arc::new(f) }))
183    }
184
185    fn completed_body() -> OutcomeSegment {
186        body_from_fn(|ex| Box::pin(async move { PipelineOutcome::Completed(ex) }))
187    }
188
189    fn uppercase_body() -> OutcomeSegment {
190        body_from_fn(|mut ex: Exchange| {
191            Box::pin(async move {
192                if let Body::Text(s) = &ex.input.body {
193                    ex.input.body = Body::Text(s.to_uppercase());
194                }
195                PipelineOutcome::Completed(ex)
196            })
197        })
198    }
199
200    fn make_exchange(text: &str) -> Exchange {
201        Exchange::new(Message::new(text))
202    }
203
204    fn fragments_expression(fragments: Vec<Exchange>) -> StreamingSplitExpression {
205        Arc::new(move |_| {
206            let frags = fragments.clone();
207            Box::pin(futures::stream::iter(frags.into_iter().map(Ok)))
208        })
209    }
210
211    fn empty_expression() -> StreamingSplitExpression {
212        Arc::new(|_| Box::pin(futures::stream::empty()))
213    }
214
215    fn stream_body_exchange(chunks: Vec<&str>) -> Exchange {
216        let chunks: Vec<Result<bytes::Bytes, CamelError>> = chunks
217            .into_iter()
218            .map(|chunk| Ok(bytes::Bytes::from(chunk.to_string())))
219            .collect();
220        let body = camel_api::StreamBody {
221            stream: Arc::new(Mutex::new(Some(Box::pin(futures::stream::iter(chunks))))),
222            metadata: camel_api::StreamMetadata::default(),
223        };
224        Exchange::new(Message::new(Body::Stream(body)))
225    }
226
227    /// Consumes `Body::Stream` and yields one text fragment per chunk, mirroring
228    /// the incremental codec expressions the compiler builds.
229    fn chunk_stream_expression() -> StreamingSplitExpression {
230        Arc::new(|exchange: Exchange| {
231            let (stream_body, parent) = match &exchange.input.body {
232                Body::Stream(sb) => (sb.clone(), {
233                    let mut p = exchange.branch_clone();
234                    p.input.body = Body::Empty;
235                    p
236                }),
237                _ => {
238                    return Box::pin(futures::stream::once(async move {
239                        Err(CamelError::TypeConversionFailed(
240                            "expected Body::Stream".to_string(),
241                        ))
242                    }));
243                }
244            };
245            let taken = match stream_body.stream.try_lock() {
246                Ok(mut guard) => guard.take(),
247                Err(_) => None,
248            };
249            let Some(stream) = taken else {
250                return Box::pin(futures::stream::once(async {
251                    Err(CamelError::ProcessorError(
252                        "stream body already consumed".to_string(),
253                    ))
254                }));
255            };
256            Box::pin(async_stream::stream! {
257                let mut stream = stream;
258                while let Some(item) = stream.next().await {
259                    match item {
260                        Ok(bytes) => {
261                            let mut fragment = parent.branch_clone();
262                            fragment.input.body =
263                                Body::Text(String::from_utf8_lossy(&bytes).into_owned());
264                            yield Ok(fragment);
265                        }
266                        Err(e) => {
267                            yield Err(e);
268                            return;
269                        }
270                    }
271                }
272            })
273        })
274    }
275
276    /// Stream wrapper that records when the underlying source is dropped.
277    struct DropTrackingStream<S> {
278        inner: Pin<Box<S>>,
279        dropped: Arc<std::sync::atomic::AtomicBool>,
280    }
281
282    impl<S: futures::Stream> futures::Stream for DropTrackingStream<S> {
283        type Item = S::Item;
284
285        fn poll_next(
286            mut self: Pin<&mut Self>,
287            cx: &mut TaskContext<'_>,
288        ) -> Poll<Option<Self::Item>> {
289            self.inner.as_mut().poll_next(cx)
290        }
291    }
292
293    impl<S> Drop for DropTrackingStream<S> {
294        fn drop(&mut self) {
295            self.dropped
296                .store(true, std::sync::atomic::Ordering::SeqCst);
297        }
298    }
299
300    /// Helper: OutcomePipeline body that counts invocations and stops on the
301    /// nth invocation (0-indexed).
302    #[derive(Clone)]
303    struct StopOnNthBody {
304        counter: Arc<std::sync::atomic::AtomicUsize>,
305        stop_at: usize,
306    }
307    impl OutcomePipeline for StopOnNthBody {
308        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
309            Box::new(self.clone())
310        }
311        fn run<'a>(
312            &'a mut self,
313            exchange: Exchange,
314        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
315            let count = self
316                .counter
317                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
318            let stop_at = self.stop_at;
319            Box::pin(async move {
320                if count >= stop_at {
321                    PipelineOutcome::Stopped(exchange)
322                } else {
323                    PipelineOutcome::Completed(exchange)
324                }
325            })
326        }
327    }
328
329    /// Helper: OutcomePipeline body that mutates the exchange body then stops.
330    #[derive(Clone)]
331    struct MutateAndStopBody;
332    impl OutcomePipeline for MutateAndStopBody {
333        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
334            Box::new(MutateAndStopBody)
335        }
336        fn run<'a>(
337            &'a mut self,
338            mut exchange: Exchange,
339        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
340            Box::pin(async move {
341                exchange.input.body = Body::Text("mutated-by-body".to_string());
342                PipelineOutcome::Stopped(exchange)
343            })
344        }
345    }
346
347    // ── Test 1: Stop halts stream consumption ──────────────────────────
348
349    #[tokio::test]
350    async fn streaming_split_stop_halts_stream_consumption() {
351        // Stream produces 5 fragments. Body stops on fragment 3 (index 2).
352        // Remaining 2 fragments are NOT consumed from the stream.
353        let fragments: Vec<Exchange> = (0..5)
354            .map(|i| Exchange::new(Message::new(format!("frag-{i}"))))
355            .collect();
356        let stored_frags = fragments.clone();
357
358        let expression: StreamingSplitExpression = Arc::new(move |_| {
359            let frags = stored_frags.clone();
360            Box::pin(futures::stream::iter(frags.into_iter().map(Ok)))
361        });
362
363        let invoke_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
364        let body = StopOnNthBody {
365            counter: Arc::clone(&invoke_count),
366            stop_at: 2, // stop on 3rd invocation (count=2 before fetch_add → 0,1 pass, 2 stops)
367        };
368
369        let mut seg = StreamingSplitSegment {
370            expression,
371            body: OutcomeSegment::new(Box::new(body)),
372            aggregation: AggregationStrategy::LastWins,
373            stop_on_exception: true,
374        };
375
376        let ex = Exchange::new(Message::new("trigger"));
377        let result = OutcomePipeline::run(&mut seg, ex).await;
378
379        assert!(
380            matches!(result, PipelineOutcome::Stopped(_)),
381            "Expected Stopped, got {result:?}"
382        );
383        // Only 3 fragments processed (indices 0, 1, 2) — fragments 3,4 never touched.
384        assert_eq!(invoke_count.load(std::sync::atomic::Ordering::SeqCst), 3);
385    }
386
387    // ── Test 2: Stop with NO aggregation ───────────────────────────────
388
389    #[tokio::test]
390    async fn streaming_split_stop_no_aggregate() {
391        // Stop fires on the first fragment. Result MUST be Stopped (NOT Completed).
392        // Aggregation is skipped entirely.
393        let fragments: Vec<Exchange> = (0..3)
394            .map(|i| Exchange::new(Message::new(format!("frag-{i}"))))
395            .collect();
396        let stored_frags = fragments.clone();
397
398        let expression: StreamingSplitExpression = Arc::new(move |_| {
399            let frags = stored_frags.clone();
400            Box::pin(futures::stream::iter(frags.into_iter().map(Ok)))
401        });
402
403        let mut seg = StreamingSplitSegment {
404            expression,
405            body: OutcomeSegment::new(Box::new(MutateAndStopBody)),
406            aggregation: AggregationStrategy::CollectAll,
407            stop_on_exception: true,
408        };
409
410        let ex = Exchange::new(Message::new("original"));
411        let result = OutcomePipeline::run(&mut seg, ex).await;
412
413        match result {
414            PipelineOutcome::Stopped(ex) => {
415                // Body mutation is preserved
416                assert_eq!(
417                    ex.input.body.as_text(),
418                    Some("mutated-by-body"),
419                    "Stopped exchange should carry body mutation"
420                );
421            }
422            other => panic!(
423                "Expected Stopped with mutated body, got {other:?} — aggregation should NOT fire"
424            ),
425        }
426    }
427
428    // ── Test 3: Stop preserves exchange mutations ──────────────────────
429
430    #[tokio::test]
431    async fn streaming_split_stop_preserves_exchange_mutations() {
432        // Verify that when a body segment mutates the exchange AND stops,
433        // those mutations are visible in the resulting Stopped exchange.
434        let fragments: Vec<Exchange> = (0..5)
435            .map(|i| Exchange::new(Message::new(format!("frag-{i}"))))
436            .collect();
437        let stored_frags = fragments.clone();
438
439        let expression: StreamingSplitExpression = Arc::new(move |_| {
440            let frags = stored_frags.clone();
441            Box::pin(futures::stream::iter(frags.into_iter().map(Ok)))
442        });
443
444        let mut seg = StreamingSplitSegment {
445            expression,
446            body: OutcomeSegment::new(Box::new(MutateAndStopBody)),
447            aggregation: AggregationStrategy::CollectAll,
448            stop_on_exception: true,
449        };
450
451        let ex = Exchange::new(Message::new("original"));
452        let result = OutcomePipeline::run(&mut seg, ex).await;
453
454        match result {
455            PipelineOutcome::Stopped(ex) => {
456                assert_eq!(
457                    ex.input.body.as_text(),
458                    Some("mutated-by-body"),
459                    "Stopped exchange should carry the body mutation from the segment"
460                );
461            }
462            other => panic!("Expected Stopped with mutation, got {other:?}"),
463        }
464    }
465
466    // ── Test 4: stop_on_exception=true — stream halts on failure ─────
467
468    #[tokio::test]
469    async fn streaming_split_stop_on_exception_true() {
470        let fragments: Vec<Exchange> = (0..3)
471            .map(|i| Exchange::new(Message::new(format!("frag-{i}"))))
472            .collect();
473        let stored_frags = fragments.clone();
474
475        let expression: StreamingSplitExpression = Arc::new(move |_| {
476            let frags = stored_frags.clone();
477            Box::pin(futures::stream::iter(frags.into_iter().map(Ok)))
478        });
479
480        let invoke_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
481        // Body that fails on the 2nd fragment (index 1), passes on others.
482        #[derive(Clone)]
483        struct FailOnSecondBody {
484            counter: Arc<std::sync::atomic::AtomicUsize>,
485        }
486        impl OutcomePipeline for FailOnSecondBody {
487            fn clone_box(&self) -> Box<dyn OutcomePipeline> {
488                Box::new(self.clone())
489            }
490            fn run<'a>(
491                &'a mut self,
492                exchange: Exchange,
493            ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
494                let count = self
495                    .counter
496                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
497                Box::pin(async move {
498                    if count == 1 {
499                        PipelineOutcome::Failed(CamelError::ProcessorError("fail".into()))
500                    } else {
501                        PipelineOutcome::Completed(exchange)
502                    }
503                })
504            }
505        }
506
507        let body = FailOnSecondBody {
508            counter: Arc::clone(&invoke_count),
509        };
510
511        let mut seg = StreamingSplitSegment {
512            expression,
513            body: OutcomeSegment::new(Box::new(body)),
514            aggregation: AggregationStrategy::LastWins,
515            stop_on_exception: true,
516        };
517
518        let ex = Exchange::new(Message::new("trigger"));
519        let result = OutcomePipeline::run(&mut seg, ex).await;
520
521        assert!(
522            matches!(result, PipelineOutcome::Failed(_)),
523            "stop_on_exception=true should propagate failure immediately"
524        );
525        // Only 2 fragments processed (index 0 passed, index 1 failed); fragment 2 never processed.
526        assert_eq!(invoke_count.load(std::sync::atomic::Ordering::SeqCst), 2);
527    }
528
529    // ── Test 5: stop_on_exception=false — stream continues on failure ─
530
531    #[tokio::test]
532    async fn streaming_split_stop_on_exception_false() {
533        let fragments: Vec<Exchange> = (0..3)
534            .map(|i| Exchange::new(Message::new(format!("frag-{i}"))))
535            .collect();
536        let stored_frags = fragments.clone();
537
538        let expression: StreamingSplitExpression = Arc::new(move |_| {
539            let frags = stored_frags.clone();
540            Box::pin(futures::stream::iter(frags.into_iter().map(Ok)))
541        });
542
543        let invoke_count = Arc::new(std::sync::atomic::AtomicUsize::new(0));
544        // Body that fails on the 2nd fragment (index 1), passes on others.
545        #[derive(Clone)]
546        struct FailOnSecondBody {
547            counter: Arc<std::sync::atomic::AtomicUsize>,
548        }
549        impl OutcomePipeline for FailOnSecondBody {
550            fn clone_box(&self) -> Box<dyn OutcomePipeline> {
551                Box::new(self.clone())
552            }
553            fn run<'a>(
554                &'a mut self,
555                exchange: Exchange,
556            ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
557                let count = self
558                    .counter
559                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
560                Box::pin(async move {
561                    if count == 1 {
562                        PipelineOutcome::Failed(CamelError::ProcessorError("fail".into()))
563                    } else {
564                        PipelineOutcome::Completed(exchange)
565                    }
566                })
567            }
568        }
569
570        let body = FailOnSecondBody {
571            counter: Arc::clone(&invoke_count),
572        };
573
574        let mut seg = StreamingSplitSegment {
575            expression,
576            body: OutcomeSegment::new(Box::new(body)),
577            aggregation: AggregationStrategy::LastWins,
578            stop_on_exception: false,
579        };
580
581        let ex = Exchange::new(Message::new("trigger"));
582        let result = OutcomePipeline::run(&mut seg, ex).await;
583
584        // stop_on_exception=false → error propagated at end, but all fragments processed.
585        assert!(
586            matches!(result, PipelineOutcome::Failed(_)),
587            "stop_on_exception=false should still propagate error at end"
588        );
589        assert_eq!(
590            invoke_count.load(std::sync::atomic::Ordering::SeqCst),
591            3,
592            "all 3 fragments should be processed"
593        );
594    }
595
596    // ------------------------------------------------------------------
597    // Coverage ported from the legacy streaming splitter tests.
598    // ------------------------------------------------------------------
599
600    #[tokio::test]
601    async fn streaming_split_last_wins_aggregation() {
602        let mut seg = StreamingSplitSegment {
603            expression: fragments_expression(vec![
604                make_exchange("a"),
605                make_exchange("b"),
606                make_exchange("c"),
607            ]),
608            body: uppercase_body(),
609            aggregation: AggregationStrategy::LastWins,
610            stop_on_exception: true,
611        };
612
613        let result = OutcomePipeline::run(&mut seg, make_exchange("original")).await;
614        match result {
615            PipelineOutcome::Completed(ex) => {
616                assert_eq!(ex.input.body.as_text(), Some("C"));
617            }
618            other => panic!("expected Completed, got {other:?}"),
619        }
620    }
621
622    #[tokio::test]
623    async fn streaming_split_original_aggregation() {
624        let mut seg = StreamingSplitSegment {
625            expression: fragments_expression(vec![make_exchange("a"), make_exchange("b")]),
626            body: uppercase_body(),
627            aggregation: AggregationStrategy::Original,
628            stop_on_exception: true,
629        };
630
631        let result = OutcomePipeline::run(&mut seg, make_exchange("original")).await;
632        match result {
633            PipelineOutcome::Completed(ex) => {
634                assert_eq!(ex.input.body.as_text(), Some("original"));
635            }
636            other => panic!("expected Completed, got {other:?}"),
637        }
638    }
639
640    #[tokio::test]
641    async fn streaming_split_custom_aggregation() {
642        let joiner: Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync> =
643            Arc::new(|mut acc: Exchange, next: Exchange| {
644                let acc_text = acc.input.body.as_text().unwrap_or("").to_string();
645                let next_text = next.input.body.as_text().unwrap_or("").to_string();
646                acc.input.body = Body::Text(format!("{acc_text}+{next_text}"));
647                acc
648            });
649
650        let mut seg = StreamingSplitSegment {
651            expression: fragments_expression(vec![
652                make_exchange("a"),
653                make_exchange("b"),
654                make_exchange("c"),
655            ]),
656            body: uppercase_body(),
657            aggregation: AggregationStrategy::Custom(joiner),
658            stop_on_exception: true,
659        };
660
661        let result = OutcomePipeline::run(&mut seg, make_exchange("original")).await;
662        match result {
663            PipelineOutcome::Completed(ex) => {
664                assert_eq!(ex.input.body.as_text(), Some("A+B+C"));
665            }
666            other => panic!("expected Completed, got {other:?}"),
667        }
668    }
669
670    #[tokio::test]
671    async fn streaming_split_empty_stream_returns_original_last_wins() {
672        let mut ex = make_exchange("original");
673        ex.set_property("marker", Value::Bool(true));
674
675        let mut seg = StreamingSplitSegment {
676            expression: empty_expression(),
677            body: completed_body(),
678            aggregation: AggregationStrategy::LastWins,
679            stop_on_exception: true,
680        };
681
682        let result = OutcomePipeline::run(&mut seg, ex).await;
683        match result {
684            PipelineOutcome::Completed(ex) => {
685                assert_eq!(ex.input.body.as_text(), Some("original"));
686                assert_eq!(ex.property("marker"), Some(&Value::Bool(true)));
687            }
688            other => panic!("expected Completed, got {other:?}"),
689        }
690    }
691
692    #[tokio::test]
693    async fn streaming_split_empty_stream_collect_all() {
694        let mut ex = make_exchange("original");
695        ex.set_property("trace_id", Value::String("empty-test".into()));
696
697        let mut seg = StreamingSplitSegment {
698            expression: empty_expression(),
699            body: completed_body(),
700            aggregation: AggregationStrategy::CollectAll,
701            stop_on_exception: true,
702        };
703
704        let result = OutcomePipeline::run(&mut seg, ex).await;
705        match result {
706            PipelineOutcome::Completed(ex) => {
707                match &ex.input.body {
708                    Body::Json(Value::Array(items)) => {
709                        assert!(items.is_empty(), "expected empty array, got {items:?}");
710                    }
711                    other => panic!("expected Body::Json([]), got {other:?}"),
712                }
713                assert_eq!(
714                    ex.property("trace_id"),
715                    Some(&Value::String("empty-test".into()))
716                );
717            }
718            other => panic!("expected Completed, got {other:?}"),
719        }
720    }
721
722    /// When the body stops, the segment returns immediately and drops the
723    /// source stream future (closing the underlying resource).
724    #[tokio::test]
725    async fn streaming_split_stop_drops_source_stream() {
726        let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false));
727        let dropped_flag = Arc::clone(&dropped);
728        let expression: StreamingSplitExpression = Arc::new(move |_| {
729            let source = DropTrackingStream {
730                inner: Box::pin(futures::stream::iter(vec![
731                    Ok(make_exchange("a")),
732                    Ok(make_exchange("b")),
733                ])),
734                dropped: Arc::clone(&dropped_flag),
735            };
736            Box::pin(source)
737        });
738        let stop_body = body_from_fn(|ex| Box::pin(async move { PipelineOutcome::Stopped(ex) }));
739
740        let mut seg = StreamingSplitSegment {
741            expression,
742            body: stop_body,
743            aggregation: AggregationStrategy::LastWins,
744            stop_on_exception: true,
745        };
746
747        let result = OutcomePipeline::run(&mut seg, make_exchange("trigger")).await;
748        assert!(
749            matches!(result, PipelineOutcome::Stopped(_)),
750            "got {result:?}"
751        );
752        assert!(
753            dropped.load(std::sync::atomic::Ordering::SeqCst),
754            "source stream should be dropped when the segment returns on Stop"
755        );
756    }
757
758    #[tokio::test]
759    async fn streaming_split_metadata_index_and_complete() {
760        let records: RecordedMetadata = Arc::new(Mutex::new(Vec::new()));
761        let records_clone = Arc::clone(&records);
762        let recorder = body_from_fn(move |ex: Exchange| {
763            let records = Arc::clone(&records_clone);
764            Box::pin(async move {
765                let index = ex.property(crate::splitter::CAMEL_SPLIT_INDEX).cloned();
766                let complete = ex.property(crate::splitter::CAMEL_SPLIT_COMPLETE).cloned();
767                records.lock().await.push((index, complete));
768                PipelineOutcome::Completed(ex)
769            })
770        });
771
772        let mut seg = StreamingSplitSegment {
773            expression: fragments_expression(vec![
774                make_exchange("x"),
775                make_exchange("y"),
776                make_exchange("z"),
777            ]),
778            body: recorder,
779            aggregation: AggregationStrategy::LastWins,
780            stop_on_exception: true,
781        };
782
783        let result = OutcomePipeline::run(&mut seg, make_exchange("original")).await;
784        assert!(result.is_success(), "expected Completed, got {result:?}");
785
786        let guard = records.lock().await; // allow-test-wait: pipeline completed before lock; single holder, no contention
787        assert_eq!(guard.len(), 3);
788        for (i, (index, _complete)) in guard.iter().enumerate() {
789            assert_eq!(
790                *index,
791                Some(Value::from(i as u64)),
792                "fragment {i}: CAMEL_SPLIT_INDEX mismatch"
793            );
794        }
795        assert_eq!(guard[0].1, Some(Value::Bool(false)));
796        assert_eq!(guard[1].1, Some(Value::Bool(false)));
797        assert_eq!(guard[2].1, Some(Value::Bool(true)));
798    }
799
800    // Inspiration: Camel SplitterTest C6 — in streaming mode the total count
801    // is unknown until end-of-stream; SPLIT_SIZE is therefore absent on
802    // intermediate fragments and present (= index+1) on the last one.
803    #[tokio::test]
804    async fn streaming_split_size_set_only_on_last_fragment() {
805        let records: RecordedMetadata = Arc::new(Mutex::new(Vec::new()));
806        let records_clone = Arc::clone(&records);
807        let recorder = body_from_fn(move |ex: Exchange| {
808            let records = Arc::clone(&records_clone);
809            Box::pin(async move {
810                let index = ex.property(crate::splitter::CAMEL_SPLIT_INDEX).cloned();
811                let size = ex.property(crate::splitter::CAMEL_SPLIT_SIZE).cloned();
812                records.lock().await.push((index, size));
813                PipelineOutcome::Completed(ex)
814            })
815        });
816
817        let mut seg = StreamingSplitSegment {
818            expression: fragments_expression(vec![
819                make_exchange("x"),
820                make_exchange("y"),
821                make_exchange("z"),
822            ]),
823            body: recorder,
824            aggregation: AggregationStrategy::LastWins,
825            stop_on_exception: true,
826        };
827
828        let result = OutcomePipeline::run(&mut seg, make_exchange("original")).await;
829        assert!(result.is_success(), "expected Completed, got {result:?}");
830
831        let guard = records.lock().await; // allow-test-wait: pipeline completed before lock; single holder, no contention
832        assert_eq!(guard.len(), 3);
833        assert_eq!(guard[0].1, None, "size must be absent on first fragment");
834        assert_eq!(guard[1].1, None, "size must be absent on middle fragment");
835        assert_eq!(
836            guard[2].1,
837            Some(Value::from(3u64)),
838            "size must equal total on last fragment"
839        );
840    }
841
842    #[tokio::test]
843    async fn streaming_split_original_aggregation_sanitizes_stream_body() {
844        let ex = stream_body_exchange(vec!["line1\n"]);
845        let mut seg = StreamingSplitSegment {
846            expression: chunk_stream_expression(),
847            body: completed_body(),
848            aggregation: AggregationStrategy::Original,
849            stop_on_exception: true,
850        };
851
852        let result = OutcomePipeline::run(&mut seg, ex).await;
853        match result {
854            PipelineOutcome::Completed(ex) => {
855                assert!(
856                    matches!(ex.input.body, Body::Empty),
857                    "original body should be sanitized to Empty, got {:?}",
858                    ex.input.body
859                );
860            }
861            other => panic!("expected Completed, got {other:?}"),
862        }
863    }
864}
camel_processor/streaming_split_segment.rs

camel_processor/
streaming_split_segment.rs
