camel_component_direct/
lib.rs

1//! In-memory direct component for rust-camel — synchronous point-to-point
2//! channel between routes sharing the same context with no serialization overhead.
3//!
4//! Main types: `DirectComponent`, `DirectEndpoint`, `DirectConsumer`, `DirectProducer`.
5
6use std::collections::HashMap;
7use std::future::Future;
8use std::pin::Pin;
9use std::sync::{Arc, Mutex};
10use std::task::{Context, Poll};
11use std::time::Duration;
12
13use async_trait::async_trait;
14use tokio::sync::{AcquireError, OwnedSemaphorePermit, Semaphore, mpsc, oneshot};
15use tokio::task::JoinHandle;
16use tokio_util::sync::CancellationToken;
17use tower::Service;
18
19use camel_component_api::parse_uri;
20use camel_component_api::{BoxProcessor, CamelError, Exchange};
21use camel_component_api::{Component, Consumer, ConsumerContext, Endpoint, ProducerContext};
22use tracing::{debug, error, info, warn};
23
24// ---------------------------------------------------------------------------
25// Shared state: maps endpoint names to senders that deliver exchanges to the
26// consumer side.  Each entry holds a sender of `(Exchange, oneshot::Sender)`
27// so the producer can wait for the consumer's pipeline to finish processing
28// and receive the (possibly transformed) exchange back.
29// ---------------------------------------------------------------------------
30
31type DirectSender = mpsc::Sender<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>;
32type DirectRegistry = Arc<Mutex<HashMap<String, DirectSender>>>;
33type AcquirePermitFut =
34    Pin<Box<dyn Future<Output = Result<OwnedSemaphorePermit, AcquireError>> + Send>>;
35
36// ---------------------------------------------------------------------------
37// Validation helpers
38// ---------------------------------------------------------------------------
39
40/// Validate the direct endpoint name (the part after `direct:`).
41fn validate_name(name: &str) -> Result<(), CamelError> {
42    if name.trim().is_empty() {
43        return Err(CamelError::InvalidUri(
44            "direct: endpoint name must not be empty".to_string(),
45        ));
46    }
47    if name.contains(char::is_whitespace) {
48        return Err(CamelError::InvalidUri(
49            "direct: endpoint name must not contain whitespace".to_string(),
50        ));
51    }
52    Ok(())
53}
54
55// ---------------------------------------------------------------------------
56// DirectConfig
57// ---------------------------------------------------------------------------
58
59/// Configuration for Direct endpoints parsed from URIs.
60///
61/// URI format: `direct:name[?timeout_ms=30000]`
62///
63/// Example: `direct:foo` creates an endpoint named "foo"
64#[derive(Debug, Clone)]
65pub struct DirectConfig {
66    /// Endpoint name (path portion).
67    pub name: String,
68    /// Timeout in milliseconds for producer `call()`. Defaults to 30 000 ms.
69    pub timeout_ms: Option<u64>,
70    /// When false, the producer returns immediately if no consumer is registered.
71    /// TODO(DIR-001): implement non-blocking send
72    pub block: Option<bool>,
73    /// When false, skip readiness error if no consumer registered.
74    pub fail_if_no_consumers: Option<bool>,
75    /// TODO(DIR-005): implement exchangePattern override
76    pub exchange_pattern: Option<String>,
77}
78
79impl DirectConfig {
80    pub fn from_uri(uri: &str) -> Result<Self, CamelError> {
81        let parts = parse_uri(uri)?;
82        if parts.scheme != "direct" {
83            return Err(CamelError::InvalidUri(format!(
84                "invalid scheme '{}', expected 'direct'",
85                parts.scheme
86            )));
87        }
88
89        let parse_bool = |name: &str, value: &str| -> Result<bool, CamelError> {
90            match value.to_ascii_lowercase().as_str() {
91                "true" | "1" | "yes" => Ok(true),
92                "false" | "0" | "no" => Ok(false),
93                _ => Err(CamelError::InvalidUri(format!(
94                    "invalid value for {}: invalid boolean value: '{}'",
95                    name, value
96                ))),
97            }
98        };
99
100        let timeout_ms = parts
101            .params
102            .get("timeout_ms")
103            .map(|v| {
104                v.parse::<u64>().map_err(|e| {
105                    CamelError::InvalidUri(format!("invalid value for timeout_ms: {}", e))
106                })
107            })
108            .transpose()?;
109
110        let block = parts
111            .params
112            .get("block")
113            .map(|v| parse_bool("block", v))
114            .transpose()?;
115
116        let fail_if_no_consumers = parts
117            .params
118            .get("fail_if_no_consumers")
119            .or_else(|| parts.params.get("failIfNoConsumers"))
120            .map(|v| parse_bool("fail_if_no_consumers", v))
121            .transpose()?;
122
123        let exchange_pattern = parts
124            .params
125            .get("exchange_pattern")
126            .or_else(|| parts.params.get("exchangePattern"))
127            .cloned();
128
129        Ok(Self {
130            name: parts.path,
131            timeout_ms,
132            block,
133            fail_if_no_consumers,
134            exchange_pattern,
135        })
136    }
137}
138
139// ---------------------------------------------------------------------------
140// DirectComponent
141// ---------------------------------------------------------------------------
142
143/// The Direct component provides in-memory synchronous communication between
144/// routes.
145///
146/// URI format: `direct:name`
147///
148/// A producer sending to `direct:foo` will block until the consumer on
149/// `direct:foo` has finished processing the exchange.
150pub struct DirectComponent {
151    registry: DirectRegistry,
152}
153
154impl DirectComponent {
155    pub fn new() -> Self {
156        Self {
157            registry: Arc::new(Mutex::new(HashMap::new())),
158        }
159    }
160}
161
162impl Default for DirectComponent {
163    fn default() -> Self {
164        Self::new()
165    }
166}
167
168impl Component for DirectComponent {
169    fn scheme(&self) -> &str {
170        "direct"
171    }
172
173    fn create_endpoint(
174        &self,
175        uri: &str,
176        _ctx: &dyn camel_component_api::ComponentContext,
177    ) -> Result<Box<dyn Endpoint>, CamelError> {
178        let config = DirectConfig::from_uri(uri)?;
179        validate_name(&config.name)?;
180        let name = config.name.clone();
181        debug!(endpoint_name = %name, "direct endpoint created");
182        Ok(Box::new(DirectEndpoint {
183            uri: uri.to_string(),
184            config,
185            registry: Arc::clone(&self.registry),
186        }))
187    }
188}
189
190// ---------------------------------------------------------------------------
191// DirectEndpoint
192// ---------------------------------------------------------------------------
193
194struct DirectEndpoint {
195    uri: String,
196    config: DirectConfig,
197    registry: DirectRegistry,
198}
199
200impl Endpoint for DirectEndpoint {
201    fn uri(&self) -> &str {
202        &self.uri
203    }
204
205    fn create_consumer(
206        &self,
207        rt: Arc<dyn camel_component_api::RuntimeObservability>,
208    ) -> Result<Box<dyn Consumer>, CamelError> {
209        Ok(Box::new(DirectConsumer::new(
210            self.config.name.clone(),
211            Arc::clone(&self.registry),
212            rt,
213        )))
214    }
215
216    fn create_producer(
217        &self,
218        _rt: Arc<dyn camel_component_api::RuntimeObservability>,
219        _ctx: &ProducerContext,
220    ) -> Result<BoxProcessor, CamelError> {
221        Ok(BoxProcessor::new(DirectProducer {
222            name: self.config.name.clone(),
223            registry: Arc::clone(&self.registry),
224            config: self.config.clone(),
225            semaphore: Arc::new(Semaphore::new(1)),
226            pending_permit: None,
227            acquire_fut: None,
228            fail_if_no_consumers: self.config.fail_if_no_consumers,
229        }))
230    }
231}
232
233// ---------------------------------------------------------------------------
234// DirectConsumer
235// ---------------------------------------------------------------------------
236
237/// The Direct consumer registers itself in the shared registry and forwards
238/// incoming exchanges to the route pipeline via `ConsumerContext`.
239struct DirectConsumer {
240    name: String,
241    registry: DirectRegistry,
242    cancel: Option<CancellationToken>,
243    handle: Option<JoinHandle<Result<(), CamelError>>>,
244    runtime: Arc<dyn camel_component_api::RuntimeObservability>,
245}
246
247impl DirectConsumer {
248    fn new(
249        name: String,
250        registry: DirectRegistry,
251        runtime: Arc<dyn camel_component_api::RuntimeObservability>,
252    ) -> Self {
253        Self {
254            name,
255            registry,
256            cancel: None,
257            handle: None,
258            runtime,
259        }
260    }
261}
262
263#[async_trait]
264impl Consumer for DirectConsumer {
265    async fn start(&mut self, context: ConsumerContext) -> Result<(), CamelError> {
266        // Create a channel for producers to send exchanges to this consumer.
267        let (tx, mut rx) =
268            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(32);
269
270        // Register ourselves so producers can find us.
271        {
272            let mut reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
273            if let Some(existing) = reg.get(&self.name)
274                && !existing.is_closed()
275            {
276                return Err(CamelError::EndpointCreationFailed(format!(
277                    "direct endpoint '{}' already has a registered consumer",
278                    self.name
279                )));
280            }
281            reg.insert(self.name.clone(), tx);
282        }
283
284        let name = self.name.clone();
285        let registry = Arc::clone(&self.registry);
286        let cancel = context.cancel_token();
287        let cancel_clone = cancel.clone();
288        let route_id = context.route_id().to_owned();
289        let runtime = Arc::clone(&self.runtime);
290
291        info!(endpoint_name = %self.name, "direct consumer started");
292
293        // Spawn the consumer loop so start() returns immediately.
294        let handle = tokio::spawn(async move {
295            loop {
296                tokio::select! {
297                    _ = cancel_clone.cancelled() => {
298                        debug!(endpoint_name = %name, "direct consumer received cancellation");
299                        break;
300                    }
301                    msg = rx.recv() => {
302                        match msg {
303                            Some((exchange, reply_tx)) => {
304                                debug!(
305                                    endpoint_name = %name,
306                                    exchange_id = %exchange.correlation_id,
307                                    "direct consumer received exchange"
308                                );
309                                let result = context.send_and_wait(exchange).await;
310                                if let Err(ref err) = result {
311                                    // (category b′: send_and_wait returned Err on a normal-data send,
312                                    // meaning the route handler did NOT absorb the failure —
313                                    // see ADR-0012 "b-bridged discriminator". This emitter is the
314                                    // only ERROR signal for the unhandled failure; must stay loud.)
315                                    runtime
316                                        .metrics()
317                                        .increment_errors(&route_id, "b-prime:direct:send-and-wait");
318                                    // log-policy: outside-contract
319                                    error!(
320                                        endpoint_name = %name,
321                                        error = %err,
322                                        "direct consumer pipeline error"
323                                    );
324                                }
325                                let _ = reply_tx.send(result);
326                            }
327                            None => break,
328                        }
329                    }
330                }
331            }
332
333            // Cleanup: remove from registry on exit
334            {
335                let mut reg = registry.lock().unwrap_or_else(|e| e.into_inner());
336                reg.remove(&name);
337            }
338
339            debug!(endpoint_name = %name, "direct consumer stopped");
340            Ok(())
341        });
342
343        self.cancel = Some(cancel);
344        self.handle = Some(handle);
345        Ok(())
346    }
347
348    async fn stop(&mut self) -> Result<(), CamelError> {
349        // Cancel the consumer loop if we have a cancellation token.
350        if let Some(cancel) = self.cancel.take() {
351            cancel.cancel();
352        }
353
354        // Wait for the spawned task to finish (with a 5s timeout).
355        if let Some(mut h) = self.handle.take() {
356            if tokio::time::timeout(Duration::from_secs(5), &mut h)
357                .await
358                .is_err()
359            {
360                h.abort();
361                let _ = h.await;
362                warn!(endpoint_name = %self.name, "consumer task did not stop in 5s; aborted");
363                // Aborted task cannot clean up its own registry entry — do it here.
364                let mut reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
365                reg.remove(&self.name);
366            }
367        } else {
368            // No join handle — just remove from registry.
369            let mut reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
370            reg.remove(&self.name);
371        }
372
373        debug!(endpoint_name = %self.name, "direct consumer stopped");
374        Ok(())
375    }
376
377    fn background_task_handle(&mut self) -> Option<JoinHandle<Result<(), CamelError>>> {
378        // Take the handle so spawn_consumer_task can monitor it for unexpected
379        // panics. The consumer loop exits cleanly via the CancellationToken, so
380        // stop() drives shutdown through cancel.take(); the task removes itself
381        // from the registry on exit.
382        self.handle.take()
383    }
384}
385
386// ---------------------------------------------------------------------------
387// DirectProducer
388// ---------------------------------------------------------------------------
389
390/// The Direct producer sends an exchange to the named direct endpoint and
391/// waits for a reply (synchronous in-memory call).
392struct DirectProducer {
393    name: String,
394    registry: DirectRegistry,
395    config: DirectConfig,
396    semaphore: Arc<Semaphore>,
397    pending_permit: Option<OwnedSemaphorePermit>,
398    acquire_fut: Option<AcquirePermitFut>,
399    fail_if_no_consumers: Option<bool>,
400}
401
402impl Clone for DirectProducer {
403    fn clone(&self) -> Self {
404        Self {
405            name: self.name.clone(),
406            registry: self.registry.clone(),
407            config: self.config.clone(),
408            semaphore: self.semaphore.clone(),
409            pending_permit: None,
410            acquire_fut: None,
411            fail_if_no_consumers: self.fail_if_no_consumers,
412        }
413    }
414}
415
416impl Service<Exchange> for DirectProducer {
417    type Response = Exchange;
418    type Error = CamelError;
419    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
420
421    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
422        // If we already hold a permit we are ready.
423        if self.pending_permit.is_some() {
424            return Poll::Ready(Ok(()));
425        }
426
427        // Check that the endpoint is registered.
428        {
429            let reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
430            match reg.get(&self.name) {
431                None => {
432                    if self.fail_if_no_consumers != Some(false) {
433                        return Poll::Ready(Err(CamelError::EndpointCreationFailed(format!(
434                            "direct endpoint '{}' not registered",
435                            self.name
436                        ))));
437                    }
438                }
439                Some(sender) if sender.is_closed() => {
440                    return Poll::Ready(Err(CamelError::EndpointCreationFailed(format!(
441                        "direct endpoint '{}' channel closed",
442                        self.name
443                    ))));
444                }
445                Some(_) => {}
446            }
447        }
448
449        // Acquire a semaphore permit (bounded concurrency).
450        let fut = self
451            .acquire_fut
452            .get_or_insert_with(|| Box::pin(Arc::clone(&self.semaphore).acquire_owned()));
453        match fut.as_mut().poll(cx) {
454            Poll::Ready(Ok(permit)) => {
455                self.acquire_fut = None;
456                self.pending_permit = Some(permit);
457                Poll::Ready(Ok(()))
458            }
459            Poll::Ready(Err(_)) => Poll::Ready(Err(CamelError::ChannelClosed)),
460            Poll::Pending => Poll::Pending,
461        }
462    }
463
464    fn call(&mut self, exchange: Exchange) -> Self::Future {
465        let _permit = match self.pending_permit.take() {
466            Some(p) => p,
467            None => {
468                return Box::pin(async {
469                    Err(CamelError::ProcessorError(
470                        "call() invoked without poll_ready()".into(),
471                    ))
472                });
473            }
474        };
475
476        let name = self.name.clone();
477        let registry = Arc::clone(&self.registry);
478        let timeout = Duration::from_millis(self.config.timeout_ms.unwrap_or(30_000));
479        let exchange_id = exchange.correlation_id.clone();
480
481        debug!(
482            endpoint_name = %name,
483            exchange_id = %exchange_id,
484            "direct producer call entry"
485        );
486
487        Box::pin(async move {
488            tokio::time::timeout(timeout, async {
489                let sender = {
490                    let reg = registry.lock().unwrap_or_else(|e| e.into_inner());
491                    reg.get(&name)
492                        .ok_or_else(|| {
493                            let err = CamelError::EndpointCreationFailed(format!(
494                                "no consumer registered for direct:{name}"
495                            ));
496                            // log-policy: handler-owned
497                            // (category a: producer send failure inside the route pipeline)
498                            warn!(endpoint_name = %name, error = %err, "direct send failed");
499                            err
500                        })?
501                        .clone()
502                };
503
504                let (reply_tx, reply_rx) = oneshot::channel();
505                sender.send((exchange, reply_tx)).await.map_err(|err| {
506                    // log-policy: handler-owned
507                    // (category a: producer send failure inside the route pipeline)
508                    warn!(endpoint_name = %name, error = %err, "direct send failed");
509                    CamelError::ChannelClosed
510                })?;
511
512                let result = reply_rx.await.map_err(|err| {
513                    // log-policy: handler-owned
514                    // (category a: producer send failure inside the route pipeline)
515                    warn!(endpoint_name = %name, error = %err, "direct send failed");
516                    CamelError::ChannelClosed
517                })?;
518
519                debug!(endpoint_name = %name, "direct message sent");
520                result
521            })
522            .await
523            .map_err(|_| CamelError::ProcessorError(format!("direct:{name} call timed out")))?
524        })
525    }
526}
527
528// ---------------------------------------------------------------------------
529// Tests
530// ---------------------------------------------------------------------------
531
532#[cfg(test)]
533mod tests {
534    use std::sync::Mutex;
535
536    use camel_api::MetricsCollector;
537    use camel_component_api::HealthCheckRegistry;
538    use std::time::Duration;
539    fn rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
540        // NoOpComponentContext implements RuntimeObservability via blanket
541        // impl and returns a no-op metrics collector — avoids panicking now
542        // that direct consumer calls runtime.metrics() on send_and_wait errors.
543        std::sync::Arc::new(NoOpComponentContext)
544    }
545
546    use super::*;
547    use camel_component_api::ExchangeEnvelope;
548    use camel_component_api::Message;
549    use camel_component_api::NoOpComponentContext;
550    use camel_component_api::RuntimeObservability;
551    use std::task::RawWakerVTable;
552    use tower::ServiceExt;
553
554    // -----------------------------------------------------------------------
555    // Recording metrics collector for testing increment_errors calls
556    // -----------------------------------------------------------------------
557
558    struct RecordingMetrics {
559        errors: Arc<Mutex<Vec<(String, String)>>>,
560    }
561
562    impl MetricsCollector for RecordingMetrics {
563        fn record_exchange_duration(&self, _: &str, _: Duration) {}
564        fn increment_errors(&self, route_id: &str, error_type: &str) {
565            self.errors
566                .lock()
567                .unwrap()
568                .push((route_id.to_string(), error_type.to_string()));
569        }
570        fn increment_exchanges(&self, _: &str) {}
571        fn set_queue_depth(&self, _: &str, _: usize) {}
572        fn record_circuit_breaker_change(&self, _: &str, _: &str, _: &str) {}
573    }
574
575    struct RecordingRuntime {
576        metrics_collector: Arc<RecordingMetrics>,
577    }
578
579    impl RecordingRuntime {
580        fn new(errors: Arc<Mutex<Vec<(String, String)>>>) -> Self {
581            Self {
582                metrics_collector: Arc::new(RecordingMetrics { errors }),
583            }
584        }
585    }
586
587    impl RuntimeObservability for RecordingRuntime {
588        fn metrics(&self) -> Arc<dyn MetricsCollector> {
589            self.metrics_collector.clone() as Arc<dyn MetricsCollector>
590        }
591        fn health(&self) -> Arc<dyn HealthCheckRegistry> {
592            panic!("RecordingRuntime::health not used in this test")
593        }
594    }
595
596    fn noop_waker() -> std::task::Waker {
597        const VTABLE: RawWakerVTable = RawWakerVTable::new(|_| RAW, |_| {}, |_| {}, |_| {});
598        const RAW: std::task::RawWaker = std::task::RawWaker::new(std::ptr::null(), &VTABLE);
599        unsafe { std::task::Waker::from_raw(RAW) }
600    }
601
602    fn test_producer_ctx() -> ProducerContext {
603        ProducerContext::new()
604    }
605
606    #[test]
607    fn test_direct_component_scheme() {
608        let component = DirectComponent::new();
609        assert_eq!(component.scheme(), "direct");
610    }
611
612    #[test]
613    fn test_direct_component_default() {
614        let component = DirectComponent::default();
615        assert_eq!(component.scheme(), "direct");
616    }
617
618    #[test]
619    fn test_direct_config_from_uri() {
620        let config = DirectConfig::from_uri("direct:orders").unwrap();
621        assert_eq!(config.name, "orders");
622    }
623
624    #[test]
625    fn test_direct_endpoint_uri() {
626        let component = DirectComponent::new();
627        let endpoint = component
628            .create_endpoint("direct:uri-check", &NoOpComponentContext)
629            .unwrap();
630        assert_eq!(endpoint.uri(), "direct:uri-check");
631    }
632
633    #[test]
634    fn test_direct_creates_endpoint() {
635        let component = DirectComponent::new();
636        let endpoint = component.create_endpoint("direct:foo", &NoOpComponentContext);
637        assert!(endpoint.is_ok());
638    }
639
640    #[test]
641    fn test_direct_wrong_scheme() {
642        let component = DirectComponent::new();
643        let result = component.create_endpoint("timer:tick", &NoOpComponentContext);
644        assert!(result.is_err());
645    }
646
647    #[test]
648    fn test_direct_endpoint_creates_consumer() {
649        let component = DirectComponent::new();
650        let endpoint = component
651            .create_endpoint("direct:foo", &NoOpComponentContext)
652            .unwrap();
653        assert!(endpoint.create_consumer(rt()).is_ok());
654    }
655
656    #[test]
657    fn test_direct_endpoint_creates_producer() {
658        let ctx = test_producer_ctx();
659        let component = DirectComponent::new();
660        let endpoint = component
661            .create_endpoint("direct:foo", &NoOpComponentContext)
662            .unwrap();
663        assert!(endpoint.create_producer(rt(), &ctx).is_ok());
664    }
665
666    #[test]
667    fn test_direct_empty_name_rejected() {
668        let component = DirectComponent::new();
669        match component.create_endpoint("direct:", &NoOpComponentContext) {
670            Err(e) => assert!(
671                e.to_string().contains("must not be empty"),
672                "unexpected error: {e}"
673            ),
674            Ok(_) => panic!("expected error for empty name"),
675        }
676    }
677
678    #[tokio::test]
679    async fn test_direct_producer_no_consumer_registered() {
680        let ctx = test_producer_ctx();
681        let component = DirectComponent::new();
682        let endpoint = component
683            .create_endpoint("direct:missing", &NoOpComponentContext)
684            .unwrap();
685        let producer = endpoint.create_producer(rt(), &ctx).unwrap();
686
687        let exchange = Exchange::new(Message::new("test"));
688        let result = producer.oneshot(exchange).await;
689        assert!(result.is_err());
690    }
691
692    #[tokio::test]
693    async fn test_direct_duplicate_consumer_returns_error() {
694        let component = DirectComponent::new();
695        let endpoint = component
696            .create_endpoint("direct:dup", &NoOpComponentContext)
697            .unwrap();
698
699        let mut consumer_a = endpoint.create_consumer(rt()).unwrap();
700        let mut consumer_b = endpoint.create_consumer(rt()).unwrap();
701
702        let (route_tx_a, _route_rx_a) = mpsc::channel::<ExchangeEnvelope>(16);
703        let ctx_a = ConsumerContext::new(
704            route_tx_a,
705            tokio_util::sync::CancellationToken::new(),
706            "direct-test-route-a".to_string(),
707        );
708        consumer_a.start(ctx_a).await.unwrap();
709
710        let (route_tx_b, _route_rx_b) = mpsc::channel::<ExchangeEnvelope>(16);
711        let ctx_b = ConsumerContext::new(
712            route_tx_b,
713            tokio_util::sync::CancellationToken::new(),
714            "direct-test-route-b".to_string(),
715        );
716        let result = consumer_b.start(ctx_b).await;
717
718        assert!(matches!(
719            result,
720            Err(CamelError::EndpointCreationFailed(msg))
721                if msg.contains("already has a registered consumer")
722        ));
723
724        consumer_a.stop().await.unwrap();
725    }
726
727    #[tokio::test]
728    async fn test_direct_producer_consumer_roundtrip() {
729        let component = DirectComponent::new();
730
731        // Create consumer endpoint and start it
732        let consumer_endpoint = component
733            .create_endpoint("direct:test", &NoOpComponentContext)
734            .unwrap();
735        let mut consumer = consumer_endpoint.create_consumer(rt()).unwrap();
736
737        // The route channel now carries ExchangeEnvelope (request-reply support).
738        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
739        let ctx = ConsumerContext::new(
740            route_tx,
741            tokio_util::sync::CancellationToken::new(),
742            "direct-test-route".to_string(),
743        );
744
745        // Start the consumer in a background task
746        tokio::spawn(async move {
747            consumer.start(ctx).await.unwrap();
748        });
749
750        // Give the consumer a moment to register
751        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
752
753        // Spawn a pipeline simulator that reads envelopes and replies Ok.
754        tokio::spawn(async move {
755            while let Some(envelope) = route_rx.recv().await {
756                let ExchangeEnvelope { exchange, reply_tx } = envelope;
757                if let Some(tx) = reply_tx {
758                    let _ = tx.send(Ok(exchange));
759                }
760            }
761        });
762
763        // Now send an exchange via the producer
764        let ctx = test_producer_ctx();
765        let producer_endpoint = component
766            .create_endpoint("direct:test", &NoOpComponentContext)
767            .unwrap();
768        let producer = producer_endpoint.create_producer(rt(), &ctx).unwrap();
769
770        let exchange = Exchange::new(Message::new("hello direct"));
771        let result = producer.oneshot(exchange).await;
772
773        assert!(result.is_ok());
774        let reply = result.unwrap();
775        assert_eq!(reply.input.body.as_text(), Some("hello direct"));
776    }
777
778    #[tokio::test]
779    async fn test_direct_propagates_error_when_no_handler() {
780        let component = DirectComponent::new();
781
782        let consumer_endpoint = component
783            .create_endpoint("direct:err-test", &NoOpComponentContext)
784            .unwrap();
785        let mut consumer = consumer_endpoint.create_consumer(rt()).unwrap();
786
787        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
788        let ctx = ConsumerContext::new(
789            route_tx,
790            tokio_util::sync::CancellationToken::new(),
791            "direct-test-route".to_string(),
792        );
793
794        tokio::spawn(async move {
795            consumer.start(ctx).await.unwrap();
796        });
797
798        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
799
800        // Pipeline simulator that replies with Err (simulates no error handler).
801        tokio::spawn(async move {
802            while let Some(envelope) = route_rx.recv().await {
803                if let Some(tx) = envelope.reply_tx {
804                    let _ = tx.send(Err(CamelError::ProcessorError("subroute failed".into())));
805                }
806            }
807        });
808
809        let ctx = test_producer_ctx();
810        let producer_endpoint = component
811            .create_endpoint("direct:err-test", &NoOpComponentContext)
812            .unwrap();
813        let producer = producer_endpoint.create_producer(rt(), &ctx).unwrap();
814
815        let exchange = Exchange::new(Message::new("test"));
816        let result = producer.oneshot(exchange).await;
817        assert!(result.is_err());
818        assert!(matches!(result.unwrap_err(), CamelError::ProcessorError(_)));
819    }
820
821    #[tokio::test]
822    async fn test_direct_consumer_stop_unregisters() {
823        let component = DirectComponent::new();
824        let endpoint = component
825            .create_endpoint("direct:cleanup", &NoOpComponentContext)
826            .unwrap();
827
828        // We need a consumer to register
829        let mut consumer = endpoint.create_consumer(rt()).unwrap();
830
831        let (route_tx, _route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
832        let ctx = ConsumerContext::new(
833            route_tx,
834            tokio_util::sync::CancellationToken::new(),
835            "direct-test-route".to_string(),
836        );
837
838        // Start consumer in background
839        let handle = tokio::spawn(async move {
840            consumer.start(ctx).await.unwrap();
841        });
842
843        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
844
845        // Verify the name is registered
846        {
847            let reg = component.registry.lock().unwrap_or_else(|e| e.into_inner());
848            assert!(reg.contains_key("cleanup"));
849        }
850
851        // Create a new consumer just to call stop (stop removes from registry)
852        let mut stop_consumer = DirectConsumer {
853            name: "cleanup".to_string(),
854            registry: Arc::clone(&component.registry),
855            cancel: None,
856            handle: None,
857            runtime: rt(),
858        };
859        stop_consumer.stop().await.unwrap();
860
861        // Verify removed from registry
862        {
863            let reg = component.registry.lock().unwrap_or_else(|e| e.into_inner());
864            assert!(!reg.contains_key("cleanup"));
865        }
866
867        handle.abort();
868    }
869
870    #[tokio::test]
871    async fn test_direct_consumer_respects_cancellation() {
872        use tokio_util::sync::CancellationToken;
873
874        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
875        let token = CancellationToken::new();
876        let (tx, _rx) = mpsc::channel(16);
877        let ctx = ConsumerContext::new(tx, token.clone(), "direct-test-route".to_string());
878
879        let mut consumer = DirectConsumer {
880            name: "cancel-test".to_string(),
881            registry: registry.clone(),
882            cancel: None,
883            handle: None,
884            runtime: rt(),
885        };
886
887        let handle = tokio::spawn(async move {
888            consumer.start(ctx).await.unwrap();
889        });
890
891        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
892        assert!(
893            registry
894                .lock()
895                .unwrap_or_else(|e| e.into_inner())
896                .contains_key("cancel-test")
897        );
898
899        token.cancel();
900        // start() now spawns the loop internally and returns immediately,
901        // so the outer handle completes right away. Give the inner task time
902        // to react to the cancellation and clean up the registry.
903        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
904
905        // After cancellation, the consumer should have cleaned up the registry
906        assert!(
907            !registry
908                .lock()
909                .unwrap_or_else(|e| e.into_inner())
910                .contains_key("cancel-test")
911        );
912
913        let _ = handle.await;
914    }
915
916    #[tokio::test]
917    async fn test_direct_consumer_stop_missing_entry_is_ok() {
918        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
919        let mut consumer = DirectConsumer {
920            name: "never-registered".to_string(),
921            registry,
922            cancel: None,
923            handle: None,
924            runtime: rt(),
925        };
926        let result = consumer.stop().await;
927        assert!(result.is_ok());
928    }
929
930    #[test]
931    fn test_poll_ready_endpoint_not_registered() {
932        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
933        let producer = DirectProducer {
934            name: "missing".to_string(),
935            registry,
936            config: DirectConfig {
937                name: "missing".to_string(),
938                timeout_ms: None,
939                block: None,
940                fail_if_no_consumers: None,
941                exchange_pattern: None,
942            },
943            semaphore: Arc::new(Semaphore::new(1)),
944            pending_permit: None,
945            acquire_fut: None,
946            fail_if_no_consumers: None,
947        };
948        let waker = noop_waker();
949        let mut cx = Context::from_waker(&waker);
950        let mut producer = producer;
951        let result = producer.poll_ready(&mut cx);
952        assert!(matches!(
953            result,
954            Poll::Ready(Err(CamelError::EndpointCreationFailed(_)))
955        ));
956    }
957
958    #[test]
959    fn test_poll_ready_endpoint_registered() {
960        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
961        let (tx, _rx) =
962            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(1);
963        registry.lock().unwrap().insert("active".to_string(), tx);
964        let producer = DirectProducer {
965            name: "active".to_string(),
966            registry,
967            config: DirectConfig {
968                name: "active".to_string(),
969                timeout_ms: None,
970                block: None,
971                fail_if_no_consumers: None,
972                exchange_pattern: None,
973            },
974            semaphore: Arc::new(Semaphore::new(1)),
975            pending_permit: None,
976            acquire_fut: None,
977            fail_if_no_consumers: None,
978        };
979        let waker = noop_waker();
980        let mut cx = Context::from_waker(&waker);
981        let mut producer = producer;
982        let result = producer.poll_ready(&mut cx);
983        assert!(matches!(result, Poll::Ready(Ok(()))));
984    }
985
986    #[test]
987    fn test_poll_ready_allows_missing_consumer_when_fail_if_no_consumers_false() {
988        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
989        let producer = DirectProducer {
990            name: "missing-ok".to_string(),
991            registry,
992            config: DirectConfig {
993                name: "missing-ok".to_string(),
994                timeout_ms: None,
995                block: None,
996                fail_if_no_consumers: Some(false),
997                exchange_pattern: None,
998            },
999            semaphore: Arc::new(Semaphore::new(1)),
1000            pending_permit: None,
1001            acquire_fut: None,
1002            fail_if_no_consumers: Some(false),
1003        };
1004
1005        let waker = noop_waker();
1006        let mut cx = Context::from_waker(&waker);
1007        let mut producer = producer;
1008        let result = producer.poll_ready(&mut cx);
1009        assert!(matches!(result, Poll::Ready(Ok(()))));
1010    }
1011
1012    #[test]
1013    fn test_poll_ready_channel_closed() {
1014        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
1015        let (tx, rx) =
1016            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(1);
1017        drop(rx);
1018        registry.lock().unwrap().insert("closed".to_string(), tx);
1019        let producer = DirectProducer {
1020            name: "closed".to_string(),
1021            registry,
1022            config: DirectConfig {
1023                name: "closed".to_string(),
1024                timeout_ms: None,
1025                block: None,
1026                fail_if_no_consumers: None,
1027                exchange_pattern: None,
1028            },
1029            semaphore: Arc::new(Semaphore::new(1)),
1030            pending_permit: None,
1031            acquire_fut: None,
1032            fail_if_no_consumers: None,
1033        };
1034        let waker = noop_waker();
1035        let mut cx = Context::from_waker(&waker);
1036        let mut producer = producer;
1037        let result = producer.poll_ready(&mut cx);
1038        assert!(matches!(
1039            result,
1040            Poll::Ready(Err(CamelError::EndpointCreationFailed(_)))
1041        ));
1042    }
1043
1044    #[tokio::test]
1045    async fn test_direct_stop_cancels_loop() {
1046        use tokio_util::sync::CancellationToken;
1047
1048        let component = DirectComponent::new();
1049        let endpoint = component
1050            .create_endpoint("direct:stop-test", &NoOpComponentContext)
1051            .unwrap();
1052        let mut consumer = endpoint.create_consumer(rt()).unwrap();
1053
1054        let token = CancellationToken::new();
1055        let (route_tx, _route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
1056        let ctx = ConsumerContext::new(route_tx, token.clone(), "direct-test-route".to_string());
1057
1058        // start() blocks — it should run the consumer loop on a JoinHandle
1059        // and return immediately. But currently start() IS the loop.
1060        // We test stop() cancels the loop.
1061        let handle = tokio::spawn(async move {
1062            consumer.start(ctx).await.unwrap();
1063        });
1064
1065        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1066        assert!(
1067            component
1068                .registry
1069                .lock()
1070                .unwrap_or_else(|e| e.into_inner())
1071                .contains_key("stop-test")
1072        );
1073
1074        // Create a new consumer just for stop
1075        let mut stop_consumer = DirectConsumer {
1076            name: "stop-test".to_string(),
1077            registry: Arc::clone(&component.registry),
1078            cancel: Some(token.clone()),
1079            handle: None,
1080            runtime: rt(),
1081        };
1082        stop_consumer.stop().await.unwrap();
1083
1084        // The consumer loop should finish within 2s after stop
1085        let result = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
1086        assert!(result.is_ok(), "Consumer loop did not stop within 2s");
1087
1088        // Registry should be cleaned up
1089        assert!(
1090            !component
1091                .registry
1092                .lock()
1093                .unwrap_or_else(|e| e.into_inner())
1094                .contains_key("stop-test")
1095        );
1096    }
1097
1098    #[tokio::test]
1099    async fn test_direct_producer_timeout() {
1100        let component = DirectComponent::new();
1101        let endpoint = component
1102            .create_endpoint("direct:timeout-test", &NoOpComponentContext)
1103            .unwrap();
1104        let mut consumer = endpoint.create_consumer(rt()).unwrap();
1105
1106        // Consumer that never replies (simulates a stuck pipeline)
1107        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
1108        let token = tokio_util::sync::CancellationToken::new();
1109        let ctx = ConsumerContext::new(route_tx, token.clone(), "direct-test-route".to_string());
1110        tokio::spawn(async move {
1111            consumer.start(ctx).await.unwrap();
1112        });
1113
1114        // Drain envelopes but hold onto reply_tx so the producer never gets a reply
1115        tokio::spawn(async move {
1116            let mut held_reply: Vec<oneshot::Sender<Result<Exchange, CamelError>>> = Vec::new();
1117            while let Some(envelope) = route_rx.recv().await {
1118                held_reply.push(envelope.reply_tx.unwrap());
1119            }
1120            drop(held_reply);
1121        });
1122
1123        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1124
1125        // Create producer with a short timeout
1126        let _ = test_producer_ctx();
1127        let _producer_endpoint = component
1128            .create_endpoint("direct:timeout-test", &NoOpComponentContext)
1129            .unwrap();
1130        let producer = DirectProducer {
1131            name: "timeout-test".to_string(),
1132            registry: Arc::clone(&component.registry),
1133            config: DirectConfig {
1134                name: "timeout-test".to_string(),
1135                timeout_ms: Some(100), // 100ms timeout
1136                block: None,
1137                fail_if_no_consumers: None,
1138                exchange_pattern: None,
1139            },
1140            semaphore: Arc::new(Semaphore::new(1)),
1141            pending_permit: None,
1142            acquire_fut: None,
1143            fail_if_no_consumers: None,
1144        };
1145
1146        let exchange = Exchange::new(Message::new("test"));
1147        let mut svc = producer;
1148        let _ = svc.poll_ready(&mut Context::from_waker(&noop_waker()));
1149        let result = svc.call(exchange).await;
1150        assert!(result.is_err(), "Expected timeout error");
1151        assert!(
1152            result.unwrap_err().to_string().contains("timed out"),
1153            "Expected timeout message"
1154        );
1155
1156        token.cancel();
1157    }
1158
1159    #[tokio::test]
1160    async fn test_send_and_wait_error_increments_errors_metric() {
1161        let errors: Arc<Mutex<Vec<(String, String)>>> = Arc::new(Mutex::new(Vec::new()));
1162        let runtime = Arc::new(RecordingRuntime::new(Arc::clone(&errors)));
1163
1164        let component = DirectComponent::new();
1165        let endpoint = component
1166            .create_endpoint("direct:metrics-error", &NoOpComponentContext)
1167            .unwrap();
1168        let mut consumer = endpoint.create_consumer(runtime).unwrap();
1169
1170        // Route that returns Err — simulates unhandled pipeline failure
1171        let cancel = tokio_util::sync::CancellationToken::new();
1172        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
1173        let ctx = ConsumerContext::new(route_tx, cancel.clone(), "test-route-id".to_string());
1174
1175        tokio::spawn(async move {
1176            consumer.start(ctx).await.unwrap();
1177        });
1178
1179        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1180
1181        // Route pipeline: reply with Err for every incoming exchange
1182        tokio::spawn(async move {
1183            while let Some(envelope) = route_rx.recv().await {
1184                if let Some(tx) = envelope.reply_tx {
1185                    let _ = tx.send(Err(CamelError::ProcessorError(
1186                        "pipeline failure".to_string(),
1187                    )));
1188                }
1189            }
1190        });
1191
1192        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1193
1194        // Send an exchange via the producer so the consumer's send_and_wait
1195        // returns Err, triggering the metrics call.
1196        let ctx = test_producer_ctx();
1197        let producer_endpoint = component
1198            .create_endpoint("direct:metrics-error", &NoOpComponentContext)
1199            .unwrap();
1200        let producer = producer_endpoint.create_producer(rt(), &ctx).unwrap();
1201
1202        let exchange = Exchange::new(Message::new("test"));
1203        // The producer oneshot wraps the call — it returns the Err from
1204        // send_and_wait back to us.
1205        let _result = producer.oneshot(exchange).await;
1206
1207        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
1208
1209        cancel.cancel();
1210
1211        // Verify MetricsCollector::increment_errors was called
1212        let recorded = errors.lock().unwrap();
1213        assert_eq!(
1214            recorded.len(),
1215            1,
1216            "expected 1 increment_errors call, got {}: {:?}",
1217            recorded.len(),
1218            *recorded
1219        );
1220        assert_eq!(recorded[0].0, "test-route-id");
1221        assert_eq!(recorded[0].1, "b-prime:direct:send-and-wait");
1222    }
1223
1224    #[test]
1225    fn test_empty_endpoint_name_rejected() {
1226        let result = DirectConfig::from_uri("direct:");
1227        // from_uri may parse empty name — validate catches it
1228        if let Ok(config) = result {
1229            assert!(
1230                validate_name(&config.name).is_err(),
1231                "expected validation error for empty name"
1232            );
1233        }
1234        // Also verify via Component (the main entry point)
1235        let component = DirectComponent::new();
1236        let result = component.create_endpoint("direct:", &NoOpComponentContext);
1237        assert!(result.is_err(), "empty endpoint name must be rejected");
1238    }
1239
1240    #[test]
1241    fn test_whitespace_endpoint_name_rejected() {
1242        let result = DirectConfig::from_uri("direct:my endpoint");
1243        if let Ok(config) = result {
1244            assert!(
1245                validate_name(&config.name).is_err(),
1246                "expected validation error for whitespace in name"
1247            );
1248        }
1249        let component = DirectComponent::new();
1250        let result = component.create_endpoint("direct:my endpoint", &NoOpComponentContext);
1251        assert!(result.is_err(), "whitespace endpoint name must be rejected");
1252    }
1253
1254    #[test]
1255    fn test_valid_endpoint_name_accepted() {
1256        let component = DirectComponent::new();
1257        let result = component.create_endpoint("direct:my-endpoint", &NoOpComponentContext);
1258        assert!(result.is_ok(), "valid endpoint name should be accepted");
1259    }
1260}
camel_component_direct/lib.rs

camel_component_direct/
lib.rs
