camel_component_api/
dispatch.rs

1//! Inline route dispatch capability — the fast-path seam between a consumer
2//! and its route pipeline.
3//!
4//! The camel-core runtime publishes an [`InlineRouteDispatcher`] on the
5//! [`ConsumerContext`](crate::consumer::ConsumerContext) when a route's
6//! concurrency model permits the inline fast path. Consumers (or the
7//! producers bound to them) may then hand Exchanges straight into the
8//! pipeline without a channel round-trip. The capability stays opaque:
9//! implementors live in camel-core and component crates never see pipeline
10//! internals (hexagonal boundary).
11
12use std::future::Future;
13use std::pin::Pin;
14
15use camel_api::{CamelError, Exchange};
16
17/// Opaque capability for dispatching an Exchange directly into a route
18/// pipeline (request-reply), bypassing the consumer submission channel.
19///
20/// Set once by the camel-core runtime before the consumer starts, via
21/// [`ConsumerContext::set_inline_dispatcher`](crate::consumer::ConsumerContext::set_inline_dispatcher).
22/// The trait exposes ONLY `dispatch` — no pipeline or processor accessors —
23/// so domain components stay framework-agnostic.
24pub trait InlineRouteDispatcher: Send + Sync + 'static {
25    /// Run `exchange` through the route pipeline and resolve with the
26    /// processed exchange, or with the pipeline error if no error handler
27    /// absorbed it.
28    fn dispatch(
29        &self,
30        exchange: Exchange,
31    ) -> Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send + 'static>>;
32}
33
34#[cfg(test)]
35mod tests {
36    use std::pin::Pin;
37    use std::sync::Arc;
38
39    use tokio::sync::mpsc;
40    use tokio_util::sync::CancellationToken;
41
42    use super::InlineRouteDispatcher;
43    use crate::consumer::{ConsumerContext, ExchangeEnvelope};
44    use camel_api::{Exchange, Message};
45
46    fn test_context() -> ConsumerContext {
47        let (tx, _rx) = mpsc::channel::<ExchangeEnvelope>(1);
48        ConsumerContext::new(tx, CancellationToken::new(), "test-route".to_string())
49    }
50
51    fn test_exchange() -> Exchange {
52        Exchange::new(Message::new("payload"))
53    }
54
55    /// No-op fake: dispatch resolves with the exchange unchanged.
56    struct IdentityDispatcher;
57
58    impl InlineRouteDispatcher for IdentityDispatcher {
59        fn dispatch(
60            &self,
61            exchange: Exchange,
62        ) -> Pin<Box<dyn Future<Output = Result<Exchange, camel_api::CamelError>> + Send + 'static>>
63        {
64            Box::pin(async move { Ok(exchange) })
65        }
66    }
67
68    /// Fake that tags the exchange with its name so tests can tell which
69    /// dispatcher answered.
70    struct TagDispatcher(&'static str);
71
72    impl InlineRouteDispatcher for TagDispatcher {
73        fn dispatch(
74            &self,
75            mut exchange: Exchange,
76        ) -> Pin<Box<dyn Future<Output = Result<Exchange, camel_api::CamelError>> + Send + 'static>>
77        {
78            let tag = self.0;
79            Box::pin(async move {
80                exchange.set_property("dispatcher", tag);
81                Ok(exchange)
82            })
83        }
84    }
85
86    #[test]
87    fn inline_dispatcher_defaults_to_none() {
88        let ctx = test_context();
89        assert!(ctx.inline_dispatcher().is_none());
90    }
91
92    #[tokio::test]
93    async fn inline_dispatcher_set_then_get_roundtrip() {
94        let ctx = test_context();
95        ctx.set_inline_dispatcher(Arc::new(IdentityDispatcher));
96
97        let dispatcher = ctx.inline_dispatcher().expect("dispatcher must be set");
98        // Clones observe the same capability slot.
99        let clone = ctx.clone();
100        assert!(clone.inline_dispatcher().is_some());
101
102        let sent = test_exchange();
103        let correlation_id = sent.correlation_id().to_string();
104        let returned = dispatcher
105            .dispatch(sent)
106            .await
107            .expect("dispatch must be Ok");
108        assert_eq!(returned.correlation_id(), correlation_id);
109        assert!(!returned.has_error());
110    }
111
112    #[tokio::test]
113    async fn inline_dispatcher_second_set_keeps_first() {
114        let ctx = test_context();
115        ctx.set_inline_dispatcher(Arc::new(TagDispatcher("A")));
116        ctx.set_inline_dispatcher(Arc::new(TagDispatcher("B")));
117
118        let dispatcher = ctx.inline_dispatcher().expect("dispatcher must be set");
119        let returned = dispatcher
120            .dispatch(test_exchange())
121            .await
122            .expect("dispatch must be Ok");
123        assert_eq!(
124            returned.property("dispatcher").and_then(|v| v.as_str()),
125            Some("A")
126        );
127    }
128}
camel_component_api/dispatch.rs

camel_component_api/
dispatch.rs
