camel_component_direct/

lib.rs

//! In-memory direct component for rust-camel — synchronous point-to-point
//! channel between routes sharing the same context with no serialization overhead.
//!
//! Main types: `DirectComponent`, `DirectEndpoint`, `DirectConsumer`, `DirectProducer`.

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::Duration;

use async_trait::async_trait;
use tokio::sync::{Semaphore, mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use tower::Service;

use camel_component_api::UriConfig;
use camel_component_api::parse_uri;
use camel_component_api::{BoxProcessor, CamelError, Exchange};
use camel_component_api::{
    Component, ComponentMetadata, Consumer, ConsumerContext, ConsumerStartupMode, Endpoint,
    ProducerContext,
};
use tracing::{debug, error, info, warn};

// ---------------------------------------------------------------------------
// Shared state: maps endpoint names to senders that deliver exchanges to the
// consumer side.  Each entry holds a sender of `(Exchange, oneshot::Sender)`
// so the producer can wait for the consumer's pipeline to finish processing
// and receive the (possibly transformed) exchange back.
// ---------------------------------------------------------------------------

type DirectSender = mpsc::Sender<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>;
type DirectRegistry = Arc<Mutex<HashMap<String, DirectSender>>>;

// ---------------------------------------------------------------------------
// Validation helpers
// ---------------------------------------------------------------------------

/// Validate the direct endpoint name (the part after `direct:`).
fn validate_name(name: &str) -> Result<(), CamelError> {
    if name.trim().is_empty() {
        return Err(CamelError::InvalidUri(
            "direct: endpoint name must not be empty".to_string(),
        ));
    }
    if name.contains(char::is_whitespace) {
        return Err(CamelError::InvalidUri(
            "direct: endpoint name must not contain whitespace".to_string(),
        ));
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// DirectConfig
// ---------------------------------------------------------------------------

/// Configuration for Direct endpoints parsed from URIs.
///
/// URI format: `direct:name[?timeout_ms=30000]`
///
/// Example: `direct:foo` creates an endpoint named "foo"
#[derive(Debug, Clone, UriConfig)]
#[uri_scheme = "direct"]
#[uri_config(
    skip_impl,
    metadata(
        scheme = "direct",
        description = "Synchronous in-memory direct invocation between routes",
        producer,
        consumer
    ),
    crate = "camel_component_api"
)]
pub struct DirectConfig {
    /// Endpoint name (path portion).
    pub name: String,
    /// Timeout in milliseconds for producer `call()`. Defaults to 30 000 ms.
    #[uri_param(
        name = "timeout_ms",
        default = "30000",
        desc = "Producer call timeout in milliseconds"
    )]
    pub timeout_ms: Option<u64>,
    /// When false, skip readiness error if no consumer registered.
    #[uri_param(
        name = "failIfNoConsumers",
        default = "true",
        desc = "Fail if no consumer registered for the name"
    )]
    pub fail_if_no_consumers: Option<bool>,
}

impl DirectConfig {
    pub fn from_uri(uri: &str) -> Result<Self, CamelError> {
        let parts = parse_uri(uri)?;
        if parts.scheme != "direct" {
            return Err(CamelError::InvalidUri(format!(
                "invalid scheme '{}', expected 'direct'",
                parts.scheme
            )));
        }

        let parse_bool = |name: &str, value: &str| -> Result<bool, CamelError> {
            match value.to_ascii_lowercase().as_str() {
                "true" | "1" | "yes" => Ok(true),
                "false" | "0" | "no" => Ok(false),
                _ => Err(CamelError::InvalidUri(format!(
                    "invalid value for {}: invalid boolean value: '{}'",
                    name, value
                ))),
            }
        };

        let timeout_ms = parts
            .params
            .get("timeout_ms")
            .map(|v| {
                v.parse::<u64>().map_err(|e| {
                    CamelError::InvalidUri(format!("invalid value for timeout_ms: {}", e))
                })
            })
            .transpose()?;

        if parts.params.contains_key("block") {
            return Err(CamelError::InvalidUri("block is not supported".into()));
        }

        let fail_if_no_consumers = parts
            .params
            .get("fail_if_no_consumers")
            .or_else(|| parts.params.get("failIfNoConsumers"))
            .map(|v| parse_bool("fail_if_no_consumers", v))
            .transpose()?;

        if parts.params.contains_key("exchange_pattern")
            || parts.params.contains_key("exchangePattern")
        {
            return Err(CamelError::InvalidUri(
                "exchange_pattern is not supported".into(),
            ));
        }

        Ok(Self {
            name: parts.path,
            timeout_ms,
            fail_if_no_consumers,
        })
    }
}

// ---------------------------------------------------------------------------
// DirectComponent
// ---------------------------------------------------------------------------

/// The Direct component provides in-memory synchronous communication between
/// routes.
///
/// URI format: `direct:name`
///
/// A producer sending to `direct:foo` will block until the consumer on
/// `direct:foo` has finished processing the exchange.
pub struct DirectComponent {
    registry: DirectRegistry,
}

impl DirectComponent {
    pub fn new() -> Self {
        Self {
            registry: Arc::new(Mutex::new(HashMap::new())),
        }
    }
}

impl Default for DirectComponent {
    fn default() -> Self {
        Self::new()
    }
}

impl Component for DirectComponent {
    fn scheme(&self) -> &str {
        "direct"
    }

    fn metadata(&self) -> ComponentMetadata {
        DirectConfig::metadata()
    }

    fn create_endpoint(
        &self,
        uri: &str,
        _ctx: &dyn camel_component_api::ComponentContext,
    ) -> Result<Box<dyn Endpoint>, CamelError> {
        let config = DirectConfig::from_uri(uri)?;
        validate_name(&config.name)?;
        let name = config.name.clone();
        debug!(endpoint_name = %name, "direct endpoint created");
        Ok(Box::new(DirectEndpoint {
            uri: uri.to_string(),
            config,
            registry: Arc::clone(&self.registry),
        }))
    }
}

// ---------------------------------------------------------------------------
// DirectEndpoint
// ---------------------------------------------------------------------------

struct DirectEndpoint {
    uri: String,
    config: DirectConfig,
    registry: DirectRegistry,
}

impl Endpoint for DirectEndpoint {
    fn uri(&self) -> &str {
        &self.uri
    }

    fn create_consumer(
        &self,
        rt: Arc<dyn camel_component_api::RuntimeObservability>,
    ) -> Result<Box<dyn Consumer>, CamelError> {
        Ok(Box::new(DirectConsumer::new(
            self.config.name.clone(),
            Arc::clone(&self.registry),
            rt,
        )))
    }

    fn create_producer(
        &self,
        _rt: Arc<dyn camel_component_api::RuntimeObservability>,
        _ctx: &ProducerContext,
    ) -> Result<BoxProcessor, CamelError> {
        Ok(BoxProcessor::new(DirectProducer {
            name: self.config.name.clone(),
            registry: Arc::clone(&self.registry),
            config: self.config.clone(),
            semaphore: Arc::new(Semaphore::new(1)),
            fail_if_no_consumers: self.config.fail_if_no_consumers,
        }))
    }
}

// ---------------------------------------------------------------------------
// DirectConsumer
// ---------------------------------------------------------------------------

/// The Direct consumer registers itself in the shared registry and forwards
/// incoming exchanges to the route pipeline via `ConsumerContext`.
struct DirectConsumer {
    name: String,
    registry: DirectRegistry,
    cancel: Option<CancellationToken>,
    runtime: Arc<dyn camel_component_api::RuntimeObservability>,
}

impl DirectConsumer {
    fn new(
        name: String,
        registry: DirectRegistry,
        runtime: Arc<dyn camel_component_api::RuntimeObservability>,
    ) -> Self {
        Self {
            name,
            registry,
            cancel: None,
            runtime,
        }
    }
}

#[async_trait]
impl Consumer for DirectConsumer {
    fn startup_mode(&self) -> ConsumerStartupMode {
        ConsumerStartupMode::Explicit
    }

    async fn start(&mut self, context: ConsumerContext) -> Result<(), CamelError> {
        // Create a channel for producers to send exchanges to this consumer.
        let (tx, mut rx) =
            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(32);

        // Register ourselves so producers can find us.
        {
            let mut reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
            if let Some(existing) = reg.get(&self.name)
                && !existing.is_closed()
            {
                return Err(CamelError::EndpointCreationFailed(format!(
                    "direct endpoint '{}' already has a registered consumer",
                    self.name
                )));
            }
            reg.insert(self.name.clone(), tx);
        }

        context.mark_ready();

        let name = self.name.clone();
        let registry = Arc::clone(&self.registry);
        let cancel = context.cancel_token();
        let cancel_clone = cancel.clone();
        let route_id = context.route_id().to_owned();
        let runtime = Arc::clone(&self.runtime);

        info!(endpoint_name = %self.name, "direct consumer started");

        self.cancel = Some(cancel);

        loop {
            tokio::select! {
                _ = cancel_clone.cancelled() => {
                    debug!(endpoint_name = %name, "direct consumer received cancellation");
                    break;
                }
                msg = rx.recv() => {
                    match msg {
                        Some((exchange, reply_tx)) => {
                            debug!(
                                endpoint_name = %name,
                                exchange_id = %exchange.correlation_id,
                                "direct consumer received exchange"
                            );
                            let result = context.send_and_wait(exchange).await;
                            if let Err(ref err) = result {
                                // (category b′: send_and_wait returned Err on a normal-data send,
                                // meaning the route handler did NOT absorb the failure —
                                // see ADR-0012 "b-bridged discriminator". This emitter is the
                                // only ERROR signal for the unhandled failure; must stay loud.)
                                runtime
                                    .metrics()
                                    .increment_errors(&route_id, "b-prime:direct:send-and-wait");
                                // log-policy: outside-contract
                                error!(
                                    endpoint_name = %name,
                                    error = %err,
                                    "direct consumer pipeline error"
                                );
                            }
                            let _ = reply_tx.send(result);
                        }
                        None => break,
                    }
                }
            }
        }

        // Cleanup: remove from registry on exit
        {
            let mut reg = registry.lock().unwrap_or_else(|e| e.into_inner());
            reg.remove(&name);
        }

        debug!(endpoint_name = %name, "direct consumer stopped");
        Ok(())
    }

    async fn stop(&mut self) -> Result<(), CamelError> {
        // Cancel the consumer loop if we have a cancellation token.
        if let Some(cancel) = self.cancel.take() {
            cancel.cancel();
        }

        let mut reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
        reg.remove(&self.name);

        debug!(endpoint_name = %self.name, "direct consumer stopped");
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// DirectProducer
// ---------------------------------------------------------------------------

/// The Direct producer sends an exchange to the named direct endpoint and
/// waits for a reply (synchronous in-memory call).
struct DirectProducer {
    name: String,
    registry: DirectRegistry,
    config: DirectConfig,
    semaphore: Arc<Semaphore>,
    fail_if_no_consumers: Option<bool>,
}

impl Clone for DirectProducer {
    fn clone(&self) -> Self {
        Self {
            name: self.name.clone(),
            registry: self.registry.clone(),
            config: self.config.clone(),
            semaphore: self.semaphore.clone(),
            fail_if_no_consumers: self.fail_if_no_consumers,
        }
    }
}

impl Service<Exchange> for DirectProducer {
    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>> {
        // Check that the endpoint is registered. Permits are NOT acquired
        // here: a permit reserved in poll_ready is held across the
        // poll_ready/call boundary, wedging the semaphore when a wrapping
        // Service re-readies a clone (tower contract: call() may only
        // reserve resources for its own future).
        let reg = self.registry.lock().unwrap_or_else(|e| e.into_inner());
        match reg.get(&self.name) {
            None => {
                if self.fail_if_no_consumers != Some(false) {
                    return Poll::Ready(Err(CamelError::EndpointCreationFailed(format!(
                        "direct endpoint '{}' not registered",
                        self.name
                    ))));
                }
                Poll::Ready(Ok(()))
            }
            Some(sender) if sender.is_closed() => {
                Poll::Ready(Err(CamelError::EndpointCreationFailed(format!(
                    "direct endpoint '{}' channel closed",
                    self.name
                ))))
            }
            Some(_) => Poll::Ready(Ok(())),
        }
    }

    fn call(&mut self, exchange: Exchange) -> Self::Future {
        let name = self.name.clone();
        let registry = Arc::clone(&self.registry);
        let semaphore = Arc::clone(&self.semaphore);
        let timeout = Duration::from_millis(self.config.timeout_ms.unwrap_or(30_000));
        let exchange_id = exchange.correlation_id.clone();

        debug!(
            endpoint_name = %name,
            exchange_id = %exchange_id,
            "direct producer call entry"
        );

        Box::pin(async move {
            // Acquire the sole concurrency permit inside call()'s future and
            // OUTSIDE the timeout: the permit is held only while this one
            // exchange is in flight, and permit contention must not eat the
            // dispatch timeout budget.
            let _permit = semaphore
                .acquire_owned()
                .await
                .map_err(|_| CamelError::ChannelClosed)?;

            tokio::time::timeout(timeout, async {
                let sender = {
                    let reg = registry.lock().unwrap_or_else(|e| e.into_inner());
                    reg.get(&name)
                        .ok_or_else(|| {
                            let err = CamelError::EndpointCreationFailed(format!(
                                "no consumer registered for direct:{name}"
                            ));
                            // log-policy: handler-owned
                            // (category a: producer send failure inside the route pipeline)
                            warn!(endpoint_name = %name, error = %err, "direct send failed");
                            err
                        })?
                        .clone()
                };

                let (reply_tx, reply_rx) = oneshot::channel();
                sender.send((exchange, reply_tx)).await.map_err(|err| {
                    // log-policy: handler-owned
                    // (category a: producer send failure inside the route pipeline)
                    warn!(endpoint_name = %name, error = %err, "direct send failed");
                    CamelError::ChannelClosed
                })?;

                let result = reply_rx.await.map_err(|err| {
                    // log-policy: handler-owned
                    // (category a: producer send failure inside the route pipeline)
                    warn!(endpoint_name = %name, error = %err, "direct send failed");
                    CamelError::ChannelClosed
                })?;

                debug!(endpoint_name = %name, "direct message sent");
                result
            })
            .await
            .map_err(|_| CamelError::ProcessorError(format!("direct:{name} call timed out")))?
        })
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::sync::Mutex;

    use camel_api::MetricsCollector;
    use camel_component_api::HealthCheckRegistry;
    use std::time::Duration;
    fn rt() -> std::sync::Arc<dyn camel_component_api::RuntimeObservability> {
        // NoOpComponentContext implements RuntimeObservability via blanket
        // impl and returns a no-op metrics collector — avoids panicking now
        // that direct consumer calls runtime.metrics() on send_and_wait errors.
        std::sync::Arc::new(NoOpComponentContext)
    }

    use super::*;
    use camel_component_api::ExchangeEnvelope;
    use camel_component_api::Message;
    use camel_component_api::NoOpComponentContext;
    use camel_component_api::RuntimeObservability;
    use camel_component_api::StartupSignal;
    use std::task::RawWakerVTable;
    use tower::ServiceExt;

    // -----------------------------------------------------------------------
    // Recording metrics collector for testing increment_errors calls
    // -----------------------------------------------------------------------

    struct RecordingMetrics {
        errors: Arc<Mutex<Vec<(String, String)>>>,
    }

    impl MetricsCollector for RecordingMetrics {
        fn record_exchange_duration(&self, _: &str, _: Duration) {}
        fn increment_errors(&self, route_id: &str, error_type: &str) {
            self.errors
                .lock()
                .unwrap()
                .push((route_id.to_string(), error_type.to_string()));
        }
        fn increment_exchanges(&self, _: &str) {}
        fn set_queue_depth(&self, _: &str, _: usize) {}
        fn record_circuit_breaker_change(&self, _: &str, _: &str, _: &str) {}
    }

    struct RecordingRuntime {
        metrics_collector: Arc<RecordingMetrics>,
    }

    impl RecordingRuntime {
        fn new(errors: Arc<Mutex<Vec<(String, String)>>>) -> Self {
            Self {
                metrics_collector: Arc::new(RecordingMetrics { errors }),
            }
        }
    }

    impl RuntimeObservability for RecordingRuntime {
        fn metrics(&self) -> Arc<dyn MetricsCollector> {
            self.metrics_collector.clone() as Arc<dyn MetricsCollector>
        }
        fn health(&self) -> Arc<dyn HealthCheckRegistry> {
            panic!("RecordingRuntime::health not used in this test")
        }
    }

    fn noop_waker() -> std::task::Waker {
        const VTABLE: RawWakerVTable = RawWakerVTable::new(|_| RAW, |_| {}, |_| {}, |_| {});
        const RAW: std::task::RawWaker = std::task::RawWaker::new(std::ptr::null(), &VTABLE);
        unsafe { std::task::Waker::from_raw(RAW) }
    }

    fn test_producer_ctx() -> ProducerContext {
        ProducerContext::new()
    }

    #[test]
    fn test_direct_component_scheme() {
        let component = DirectComponent::new();
        assert_eq!(component.scheme(), "direct");
    }

    #[test]
    fn test_direct_component_default() {
        let component = DirectComponent::default();
        assert_eq!(component.scheme(), "direct");
    }

    #[test]
    fn test_direct_config_from_uri() {
        let config = DirectConfig::from_uri("direct:orders").unwrap();
        assert_eq!(config.name, "orders");
    }

    #[test]
    fn rejects_block_param() {
        let result = DirectConfig::from_uri("direct:foo?block=true");
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("block is not supported"),
            "expected block rejection"
        );
    }

    #[test]
    fn rejects_exchange_pattern_snake_case() {
        let result = DirectConfig::from_uri("direct:foo?exchange_pattern=InOnly");
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("exchange_pattern is not supported"),
            "expected exchange_pattern rejection"
        );
    }

    #[test]
    fn rejects_exchange_pattern_camel_case() {
        let result = DirectConfig::from_uri("direct:foo?exchangePattern=InOnly");
        assert!(result.is_err());
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("exchange_pattern is not supported"),
            "expected exchangePattern rejection"
        );
    }

    #[test]
    fn test_direct_endpoint_uri() {
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:uri-check", &NoOpComponentContext)
            .unwrap();
        assert_eq!(endpoint.uri(), "direct:uri-check");
    }

    #[test]
    fn test_direct_creates_endpoint() {
        let component = DirectComponent::new();
        let endpoint = component.create_endpoint("direct:foo", &NoOpComponentContext);
        assert!(endpoint.is_ok());
    }

    #[test]
    fn test_direct_wrong_scheme() {
        let component = DirectComponent::new();
        let result = component.create_endpoint("timer:tick", &NoOpComponentContext);
        assert!(result.is_err());
    }

    #[test]
    fn test_direct_endpoint_creates_consumer() {
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:foo", &NoOpComponentContext)
            .unwrap();
        assert!(endpoint.create_consumer(rt()).is_ok());
    }

    #[test]
    fn test_direct_endpoint_creates_producer() {
        let ctx = test_producer_ctx();
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:foo", &NoOpComponentContext)
            .unwrap();
        assert!(endpoint.create_producer(rt(), &ctx).is_ok());
    }

    #[test]
    fn test_direct_empty_name_rejected() {
        let component = DirectComponent::new();
        match component.create_endpoint("direct:", &NoOpComponentContext) {
            Err(e) => assert!(
                e.to_string().contains("must not be empty"),
                "unexpected error: {e}"
            ),
            Ok(_) => panic!("expected error for empty name"),
        }
    }

    #[tokio::test]
    async fn test_direct_producer_no_consumer_registered() {
        let ctx = test_producer_ctx();
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:missing", &NoOpComponentContext)
            .unwrap();
        let producer = endpoint.create_producer(rt(), &ctx).unwrap();

        let exchange = Exchange::new(Message::new("test"));
        let result = producer.oneshot(exchange).await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_direct_duplicate_consumer_returns_error() {
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:dup", &NoOpComponentContext)
            .unwrap();

        let mut consumer_a = endpoint.create_consumer(rt()).unwrap();
        let mut consumer_b = endpoint.create_consumer(rt()).unwrap();

        let (route_tx_a, _route_rx_a) = mpsc::channel::<ExchangeEnvelope>(16);
        let token_a = tokio_util::sync::CancellationToken::new();
        let ctx_a = ConsumerContext::new(
            route_tx_a,
            token_a.clone(),
            "direct-test-route-a".to_string(),
        );
        let handle_a = tokio::spawn(async move {
            consumer_a.start(ctx_a).await.unwrap();
        });
        tokio::time::sleep(std::time::Duration::from_millis(10)).await;

        let (route_tx_b, _route_rx_b) = mpsc::channel::<ExchangeEnvelope>(16);
        let ctx_b = ConsumerContext::new(
            route_tx_b,
            tokio_util::sync::CancellationToken::new(),
            "direct-test-route-b".to_string(),
        );
        let result = consumer_b.start(ctx_b).await;

        assert!(matches!(
            result,
            Err(CamelError::EndpointCreationFailed(msg))
                if msg.contains("already has a registered consumer")
        ));

        token_a.cancel();
        handle_a.await.unwrap();
    }

    #[tokio::test]
    async fn test_direct_producer_consumer_roundtrip() {
        let component = DirectComponent::new();

        // Create consumer endpoint and start it
        let consumer_endpoint = component
            .create_endpoint("direct:test", &NoOpComponentContext)
            .unwrap();
        let mut consumer = consumer_endpoint.create_consumer(rt()).unwrap();

        // The route channel now carries ExchangeEnvelope (request-reply support).
        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let ctx = ConsumerContext::new(
            route_tx,
            tokio_util::sync::CancellationToken::new(),
            "direct-test-route".to_string(),
        );

        // Start the consumer in a background task
        tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        // Give the consumer a moment to register
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Spawn a pipeline simulator that reads envelopes and replies Ok.
        tokio::spawn(async move {
            while let Some(envelope) = route_rx.recv().await {
                let ExchangeEnvelope { exchange, reply_tx } = envelope;
                if let Some(tx) = reply_tx {
                    let _ = tx.send(Ok(exchange));
                }
            }
        });

        // Now send an exchange via the producer
        let ctx = test_producer_ctx();
        let producer_endpoint = component
            .create_endpoint("direct:test", &NoOpComponentContext)
            .unwrap();
        let producer = producer_endpoint.create_producer(rt(), &ctx).unwrap();

        let exchange = Exchange::new(Message::new("hello direct"));
        let result = producer.oneshot(exchange).await;

        assert!(result.is_ok());
        let reply = result.unwrap();
        assert_eq!(reply.input.body.as_text(), Some("hello direct"));
    }

    #[tokio::test]
    async fn test_direct_propagates_error_when_no_handler() {
        let component = DirectComponent::new();

        let consumer_endpoint = component
            .create_endpoint("direct:err-test", &NoOpComponentContext)
            .unwrap();
        let mut consumer = consumer_endpoint.create_consumer(rt()).unwrap();

        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let ctx = ConsumerContext::new(
            route_tx,
            tokio_util::sync::CancellationToken::new(),
            "direct-test-route".to_string(),
        );

        tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Pipeline simulator that replies with Err (simulates no error handler).
        tokio::spawn(async move {
            while let Some(envelope) = route_rx.recv().await {
                if let Some(tx) = envelope.reply_tx {
                    let _ = tx.send(Err(CamelError::ProcessorError("subroute failed".into())));
                }
            }
        });

        let ctx = test_producer_ctx();
        let producer_endpoint = component
            .create_endpoint("direct:err-test", &NoOpComponentContext)
            .unwrap();
        let producer = producer_endpoint.create_producer(rt(), &ctx).unwrap();

        let exchange = Exchange::new(Message::new("test"));
        let result = producer.oneshot(exchange).await;
        assert!(result.is_err());
        assert!(matches!(result.unwrap_err(), CamelError::ProcessorError(_)));
    }

    #[tokio::test]
    async fn test_direct_consumer_stop_unregisters() {
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:cleanup", &NoOpComponentContext)
            .unwrap();

        // We need a consumer to register
        let mut consumer = endpoint.create_consumer(rt()).unwrap();

        let (route_tx, _route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let ctx = ConsumerContext::new(
            route_tx,
            tokio_util::sync::CancellationToken::new(),
            "direct-test-route".to_string(),
        );

        // Start consumer in background
        let handle = tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Verify the name is registered
        {
            let reg = component.registry.lock().unwrap_or_else(|e| e.into_inner());
            assert!(reg.contains_key("cleanup"));
        }

        // Create a new consumer just to call stop (stop removes from registry)
        let mut stop_consumer = DirectConsumer {
            name: "cleanup".to_string(),
            registry: Arc::clone(&component.registry),
            cancel: None,
            runtime: rt(),
        };
        stop_consumer.stop().await.unwrap();

        // Verify removed from registry
        {
            let reg = component.registry.lock().unwrap_or_else(|e| e.into_inner());
            assert!(!reg.contains_key("cleanup"));
        }

        handle.abort();
    }

    #[tokio::test]
    async fn test_direct_consumer_respects_cancellation() {
        use tokio_util::sync::CancellationToken;

        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let token = CancellationToken::new();
        let (tx, _rx) = mpsc::channel(16);
        let ctx = ConsumerContext::new(tx, token.clone(), "direct-test-route".to_string());

        let mut consumer = DirectConsumer {
            name: "cancel-test".to_string(),
            registry: registry.clone(),
            cancel: None,
            runtime: rt(),
        };

        let handle = tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert!(
            registry
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .contains_key("cancel-test")
        );

        token.cancel();
        // start() now spawns the loop internally and returns immediately,
        // so the outer handle completes right away. Give the inner task time
        // to react to the cancellation and clean up the registry.
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;

        // After cancellation, the consumer should have cleaned up the registry
        assert!(
            !registry
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .contains_key("cancel-test")
        );

        let _ = handle.await;
    }

    #[tokio::test]
    async fn test_direct_consumer_stop_missing_entry_is_ok() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let mut consumer = DirectConsumer {
            name: "never-registered".to_string(),
            registry,
            cancel: None,
            runtime: rt(),
        };
        let result = consumer.stop().await;
        assert!(result.is_ok());
    }

    fn direct_producer(
        name: &str,
        registry: DirectRegistry,
        fail_if_no_consumers: Option<bool>,
        timeout_ms: Option<u64>,
    ) -> DirectProducer {
        DirectProducer {
            name: name.to_string(),
            registry,
            config: DirectConfig {
                name: name.to_string(),
                timeout_ms,
                fail_if_no_consumers,
            },
            semaphore: Arc::new(Semaphore::new(1)),
            fail_if_no_consumers,
        }
    }

    #[test]
    fn poll_ready_absent_consumer_fails_fast() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let mut producer = direct_producer("missing", registry, None, None);
        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);
        let result = producer.poll_ready(&mut cx);
        assert!(matches!(
            result,
            Poll::Ready(Err(CamelError::EndpointCreationFailed(_)))
        ));
    }

    #[test]
    fn poll_ready_live_consumer_ok_without_permit() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let (tx, _rx) =
            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(1);
        registry.lock().unwrap().insert("active".to_string(), tx);
        let mut producer = direct_producer("active", registry, None, None);
        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);
        let result = producer.poll_ready(&mut cx);
        assert!(matches!(result, Poll::Ready(Ok(()))));
        assert_eq!(
            producer.semaphore.available_permits(),
            1,
            "poll_ready must not consume the call permit"
        );
    }

    #[test]
    fn test_poll_ready_allows_missing_consumer_when_fail_if_no_consumers_false() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let mut producer = direct_producer("missing-ok", registry, Some(false), None);

        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);
        let result = producer.poll_ready(&mut cx);
        assert!(matches!(result, Poll::Ready(Ok(()))));
    }

    #[test]
    fn poll_ready_closed_channel_fails_fast() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let (tx, rx) =
            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(1);
        drop(rx);
        registry.lock().unwrap().insert("closed".to_string(), tx);
        let mut producer = direct_producer("closed", registry, None, None);
        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);
        match producer.poll_ready(&mut cx) {
            Poll::Ready(Err(CamelError::EndpointCreationFailed(msg))) => {
                assert!(
                    msg.contains("channel closed"),
                    "unexpected error message: {msg}"
                );
            }
            other => panic!("expected channel-closed error, got: {other:?}"),
        }
    }

    #[tokio::test]
    async fn call_blocks_on_semaphore_until_release() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let (tx, mut rx) =
            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(4);
        registry.lock().unwrap().insert("park".to_string(), tx);

        let mut producer = direct_producer("park", registry, None, None);
        let semaphore = Arc::clone(&producer.semaphore);

        // Consumer stub: parks the FIRST exchange's reply on the
        // test-controlled release signal; later exchanges reply immediately.
        let (release_tx, release_rx) = oneshot::channel::<()>();
        let (parked_tx, parked_rx) = oneshot::channel::<()>();
        tokio::spawn(async move {
            let mut parked_tx = Some(parked_tx);
            let mut release_rx = Some(release_rx);
            let mut first = true;
            while let Some((exchange, reply_tx)) = rx.recv().await {
                if first {
                    first = false;
                    if let Some(signal) = parked_tx.take() {
                        let _ = signal.send(());
                    }
                    if let Some(release) = release_rx.take() {
                        let _ = release.await;
                    }
                }
                let _ = reply_tx.send(Ok(exchange));
            }
        });

        let fut_a = producer.call(Exchange::new(Message::new("a")));
        let fut_b = producer.call(Exchange::new(Message::new("b")));
        let a = tokio::spawn(fut_a);
        let b = tokio::spawn(fut_b);

        tokio::time::timeout(Duration::from_secs(5), parked_rx)
            .await
            .expect("consumer stub must receive call A's exchange within 5s")
            .expect("parked signal must be sent");
        // Let B run into the semaphore acquisition and park there.
        tokio::task::yield_now().await;
        tokio::task::yield_now().await;

        assert_eq!(
            semaphore.available_permits(),
            0,
            "call A must hold the sole permit while awaiting its reply"
        );
        assert!(
            !b.is_finished(),
            "call B must be blocked on the semaphore while call A is in flight"
        );

        release_tx
            .send(())
            .expect("release signal must be received");

        let reply_a = a
            .await
            .expect("call A task must not panic")
            .expect("call A must complete Ok once its reply arrives");
        assert_eq!(reply_a.input.body.as_text(), Some("a"));
        let reply_b = b
            .await
            .expect("call B task must not panic")
            .expect("call B must complete Ok after the permit is released");
        assert_eq!(reply_b.input.body.as_text(), Some("b"));
    }

    #[tokio::test]
    async fn call_pending_when_all_permits_held() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));
        let (tx, _rx) =
            mpsc::channel::<(Exchange, oneshot::Sender<Result<Exchange, CamelError>>)>(1);
        registry.lock().unwrap().insert("held".to_string(), tx);

        let mut producer = direct_producer("held", registry, None, None);
        let permit = Arc::clone(&producer.semaphore)
            .try_acquire_owned()
            .expect("sole permit must be free before call");

        let mut fut = producer.call(Exchange::new(Message::new("x")));
        let waker = noop_waker();
        let mut cx = Context::from_waker(&waker);
        assert!(
            fut.as_mut().poll(&mut cx).is_pending(),
            "call must pend while the sole permit is held elsewhere"
        );

        drop(permit);
        // Next poll completes acquisition and proceeds into the send/reply
        // round-trip (which pends: nobody replies). Holding the permit proves
        // the future got past acquisition.
        assert!(fut.as_mut().poll(&mut cx).is_pending());
        assert_eq!(
            producer.semaphore.available_permits(),
            0,
            "call future must now hold the permit (past acquisition)"
        );
    }

    #[tokio::test]
    async fn test_direct_stop_cancels_loop() {
        use tokio_util::sync::CancellationToken;

        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:stop-test", &NoOpComponentContext)
            .unwrap();
        let mut consumer = endpoint.create_consumer(rt()).unwrap();

        let token = CancellationToken::new();
        let (route_tx, _route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let ctx = ConsumerContext::new(route_tx, token.clone(), "direct-test-route".to_string());

        // start() runs the consumer loop inline on the managed consumer task.
        // We test stop() cancels the loop.
        let handle = tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        assert!(
            component
                .registry
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .contains_key("stop-test")
        );

        // Create a new consumer just for stop
        let mut stop_consumer = DirectConsumer {
            name: "stop-test".to_string(),
            registry: Arc::clone(&component.registry),
            cancel: Some(token.clone()),
            runtime: rt(),
        };
        stop_consumer.stop().await.unwrap();

        // The consumer loop should finish within 2s after stop
        let result = tokio::time::timeout(std::time::Duration::from_secs(2), handle).await;
        assert!(result.is_ok(), "Consumer loop did not stop within 2s");

        // Registry should be cleaned up
        assert!(
            !component
                .registry
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .contains_key("stop-test")
        );
    }

    #[tokio::test]
    async fn test_direct_producer_timeout() {
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:timeout-test", &NoOpComponentContext)
            .unwrap();
        let mut consumer = endpoint.create_consumer(rt()).unwrap();

        // Consumer that never replies (simulates a stuck pipeline)
        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let token = tokio_util::sync::CancellationToken::new();
        let ctx = ConsumerContext::new(route_tx, token.clone(), "direct-test-route".to_string());
        tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        // Drain envelopes but hold onto reply_tx so the producer never gets a reply
        tokio::spawn(async move {
            let mut held_reply: Vec<oneshot::Sender<Result<Exchange, CamelError>>> = Vec::new();
            while let Some(envelope) = route_rx.recv().await {
                held_reply.push(envelope.reply_tx.unwrap());
            }
            drop(held_reply);
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Create producer with a short timeout
        let _ = test_producer_ctx();
        let _producer_endpoint = component
            .create_endpoint("direct:timeout-test", &NoOpComponentContext)
            .unwrap();
        let producer = direct_producer(
            "timeout-test",
            Arc::clone(&component.registry),
            None,
            Some(100), // 100ms timeout
        );

        let exchange = Exchange::new(Message::new("test"));
        let mut svc = producer;
        let _ = svc.poll_ready(&mut Context::from_waker(&noop_waker()));
        let result = svc.call(exchange).await;
        assert!(result.is_err(), "Expected timeout error");
        assert!(
            result.unwrap_err().to_string().contains("timed out"),
            "Expected timeout message"
        );

        token.cancel();
    }

    #[tokio::test]
    async fn test_send_and_wait_error_increments_errors_metric() {
        let errors: Arc<Mutex<Vec<(String, String)>>> = Arc::new(Mutex::new(Vec::new()));
        let runtime = Arc::new(RecordingRuntime::new(Arc::clone(&errors)));

        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:metrics-error", &NoOpComponentContext)
            .unwrap();
        let mut consumer = endpoint.create_consumer(runtime).unwrap();

        // Route that returns Err — simulates unhandled pipeline failure
        let cancel = tokio_util::sync::CancellationToken::new();
        let (route_tx, mut route_rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let ctx = ConsumerContext::new(route_tx, cancel.clone(), "test-route-id".to_string());

        tokio::spawn(async move {
            consumer.start(ctx).await.unwrap();
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Route pipeline: reply with Err for every incoming exchange
        tokio::spawn(async move {
            while let Some(envelope) = route_rx.recv().await {
                if let Some(tx) = envelope.reply_tx {
                    let _ = tx.send(Err(CamelError::ProcessorError(
                        "pipeline failure".to_string(),
                    )));
                }
            }
        });

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Send an exchange via the producer so the consumer's send_and_wait
        // returns Err, triggering the metrics call.
        let ctx = test_producer_ctx();
        let producer_endpoint = component
            .create_endpoint("direct:metrics-error", &NoOpComponentContext)
            .unwrap();
        let producer = producer_endpoint.create_producer(rt(), &ctx).unwrap();

        let exchange = Exchange::new(Message::new("test"));
        // The producer oneshot wraps the call — it returns the Err from
        // send_and_wait back to us.
        let _result = producer.oneshot(exchange).await;

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        cancel.cancel();

        // Verify MetricsCollector::increment_errors was called
        let recorded = errors.lock().unwrap();
        assert_eq!(
            recorded.len(),
            1,
            "expected 1 increment_errors call, got {}: {:?}",
            recorded.len(),
            *recorded
        );
        assert_eq!(recorded[0].0, "test-route-id");
        assert_eq!(recorded[0].1, "b-prime:direct:send-and-wait");
    }

    #[test]
    fn test_empty_endpoint_name_rejected() {
        let result = DirectConfig::from_uri("direct:");
        // from_uri may parse empty name — validate catches it
        if let Ok(config) = result {
            assert!(
                validate_name(&config.name).is_err(),
                "expected validation error for empty name"
            );
        }
        // Also verify via Component (the main entry point)
        let component = DirectComponent::new();
        let result = component.create_endpoint("direct:", &NoOpComponentContext);
        assert!(result.is_err(), "empty endpoint name must be rejected");
    }

    #[test]
    fn test_whitespace_endpoint_name_rejected() {
        let result = DirectConfig::from_uri("direct:my endpoint");
        if let Ok(config) = result {
            assert!(
                validate_name(&config.name).is_err(),
                "expected validation error for whitespace in name"
            );
        }
        let component = DirectComponent::new();
        let result = component.create_endpoint("direct:my endpoint", &NoOpComponentContext);
        assert!(result.is_err(), "whitespace endpoint name must be rejected");
    }

    #[test]
    fn test_valid_endpoint_name_accepted() {
        let component = DirectComponent::new();
        let result = component.create_endpoint("direct:my-endpoint", &NoOpComponentContext);
        assert!(result.is_ok(), "valid endpoint name should be accepted");
    }

    #[test]
    fn test_direct_consumer_startup_mode_is_explicit() {
        let component = DirectComponent::new();
        let endpoint = component
            .create_endpoint("direct:ready-check", &NoOpComponentContext)
            .unwrap();
        let consumer = endpoint.create_consumer(rt()).unwrap();
        assert_eq!(
            consumer.startup_mode(),
            ConsumerStartupMode::Explicit,
            "DirectConsumer must opt into Explicit startup"
        );
    }

    #[tokio::test]
    async fn test_direct_consumer_marks_ready_after_registration() {
        let registry: DirectRegistry = Arc::new(Mutex::new(HashMap::new()));

        let mut consumer = DirectConsumer {
            name: "ready-probe-direct".into(),
            registry: registry.clone(),
            cancel: None,
            runtime: rt(),
        };

        let (tx, _rx) = mpsc::channel::<ExchangeEnvelope>(16);
        let token = CancellationToken::new();
        let ctx = ConsumerContext::new(tx, token.clone(), "ready-probe-route".to_string());

        let (signal, startup_rx) = StartupSignal::pair();
        let ctx = ctx.with_startup(signal);

        tokio::spawn(async move {
            let _ = consumer.start(ctx).await;
        });

        let result = tokio::time::timeout(Duration::from_secs(2), startup_rx.await_ready())
            .await
            .expect("DirectConsumer must call ctx.mark_ready() after registration");
        assert!(
            result.is_ok(),
            "mark_ready must resolve Ok after registration"
        );

        {
            let reg = registry.lock().unwrap_or_else(|e| e.into_inner());
            assert!(
                reg.contains_key("ready-probe-direct"),
                "registry must contain consumer name after mark_ready resolves"
            );
        }

        token.cancel();
    }
}