camel_component_api/

consumer.rs

use std::sync::Arc;

use async_trait::async_trait;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;

use camel_api::security_policy::SecurityPolicy;
use camel_api::{CamelError, Exchange};
use camel_auth::{CredentialSource, TokenAuthenticator};

/// A message sent from a consumer to the route pipeline.
///
/// Fire-and-forget exchanges use `reply_tx = None`.
/// Request-reply exchanges (e.g. `direct:`) provide a `reply_tx` so the
/// pipeline result can be sent back to the consumer.
pub struct ExchangeEnvelope {
    pub exchange: Exchange,
    pub reply_tx: Option<oneshot::Sender<Result<Exchange, CamelError>>>,
}

/// Context provided to a Consumer, allowing it to send exchanges into the route.
#[derive(Clone)]
pub struct ConsumerContext {
    sender: mpsc::Sender<ExchangeEnvelope>,
    cancel_token: CancellationToken,
    route_id: String,
}

impl ConsumerContext {
    /// Create a new consumer context wrapping the given channel sender.
    ///
    /// The `route_id` identifies the route this consumer is bound to, enabling
    /// ADR-0012 per-route metrics and health observations.
    pub fn new(
        sender: mpsc::Sender<ExchangeEnvelope>,
        cancel_token: CancellationToken,
        route_id: String,
    ) -> Self {
        Self {
            sender,
            cancel_token,
            route_id,
        }
    }

    /// Returns a future that resolves when shutdown is requested.
    /// Use in `tokio::select!` inside consumer loops.
    pub async fn cancelled(&self) {
        self.cancel_token.cancelled().await
    }

    /// Returns true if shutdown has been requested.
    pub fn is_cancelled(&self) -> bool {
        self.cancel_token.is_cancelled()
    }

    /// Returns the route_id this consumer is bound to.
    ///
    /// Available for ADR-0012 metrics/health calls that require a route_id
    /// (categories (b′), (e), (g)). Set at construction time by the route
    /// controller when spawning the consumer task.
    pub fn route_id(&self) -> &str {
        &self.route_id
    }

    /// Returns a clone of the `CancellationToken`.
    ///
    /// Useful for consumers that spawn per-request tasks and need to propagate
    /// shutdown to each task. See `HttpConsumer` for an example.
    pub fn cancel_token(&self) -> CancellationToken {
        self.cancel_token.clone()
    }

    /// Returns a clone of the channel sender for manual exchange submission.
    ///
    /// Useful for consumers that spawn per-request tasks (e.g., `HttpConsumer`)
    /// where each task independently sends exchanges into the pipeline.
    /// For simple consumers, prefer `send()` or `send_and_wait()` instead.
    pub fn sender(&self) -> mpsc::Sender<ExchangeEnvelope> {
        self.sender.clone()
    }

    /// Send an exchange into the route pipeline (fire-and-forget).
    pub async fn send(&self, exchange: Exchange) -> Result<(), CamelError> {
        self.sender
            .send(ExchangeEnvelope {
                exchange,
                reply_tx: None,
            })
            .await
            .map_err(|_| CamelError::ChannelClosed)
    }

    /// Send an exchange and wait for the pipeline result (request-reply).
    ///
    /// Returns `Ok(exchange)` on success or `Err(e)` if the pipeline failed
    /// without an error handler absorbing the error.
    pub async fn send_and_wait(&self, exchange: Exchange) -> Result<Exchange, CamelError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.sender
            .send(ExchangeEnvelope {
                exchange,
                reply_tx: Some(reply_tx),
            })
            .await
            .map_err(|_| CamelError::ChannelClosed)?;
        reply_rx.await.map_err(|_| CamelError::ChannelClosed)?
    }
}

/// Security context passed to a consumer before `start()`.
///
/// Carries the `SecurityPolicy` and `TokenAuthenticator` from the route
/// controller so consumers (e.g. WebSocket) can register auth state
/// before accepting connections.
pub struct SecurityContext {
    pub policy: Arc<dyn SecurityPolicy>,
    pub authenticator: Arc<dyn TokenAuthenticator>,
    pub credential_sources: Vec<CredentialSource>,
}

impl SecurityContext {
    pub fn new(
        policy: impl SecurityPolicy + 'static,
        authenticator: Arc<dyn TokenAuthenticator>,
    ) -> Self {
        Self {
            policy: Arc::new(policy),
            authenticator,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        }
    }

    pub fn from_arc(
        policy: Arc<dyn SecurityPolicy>,
        authenticator: Arc<dyn TokenAuthenticator>,
    ) -> Self {
        Self {
            policy,
            authenticator,
            credential_sources: vec![CredentialSource::AuthorizationHeader],
        }
    }

    pub fn with_credential_sources(mut self, sources: Vec<CredentialSource>) -> Self {
        self.credential_sources = sources;
        self
    }
}

impl Clone for SecurityContext {
    fn clone(&self) -> Self {
        Self {
            policy: Arc::clone(&self.policy),
            authenticator: Arc::clone(&self.authenticator),
            credential_sources: self.credential_sources.clone(),
        }
    }
}

impl std::fmt::Debug for SecurityContext {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SecurityContext")
            .field("policy", &"<SecurityPolicy>")
            .field("authenticator", &"<TokenAuthenticator>")
            .field("credential_sources", &self.credential_sources)
            .finish()
    }
}

/// How a consumer's exchanges should be processed by the pipeline.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ConcurrencyModel {
    /// Exchanges are processed one at a time, in order. Default for polling
    /// consumers (timer, file) and synchronous consumers (direct).
    Sequential,
    /// Exchanges are processed concurrently via `tokio::spawn`. Optional
    /// semaphore limit (`max`). `None` means unbounded (channel buffer is
    /// the only backpressure).
    Concurrent { max: Option<usize> },
}

/// A Consumer receives data from an external system and submits Exchanges
/// to the Route's Pipeline via the [`ConsumerContext`].
///
/// # Shutdown Contract
///
/// The Runtime guarantees the following lifecycle:
///
/// 1. `start()` is called once. The Runtime spawns a task that owns the Consumer.
/// 2. On route stop, the Runtime cancels the [`ConsumerContext`] cancel token.
/// 3. The spawned task calls `stop()` on ALL exit paths after `start()` succeeds
///    (clean exit, crash, cancellation, natural completion).
/// 4. `background_task_handle()` is a supervision hook for crash propagation (ADR-0007),
///    NOT the shutdown API.
///
/// Component authors MUST ensure:
///
/// - `stop()` cancels all component-owned inner tasks and cleans up registrations/resources.
/// - If inner tasks use a private `CancellationToken`, `stop()` MUST cancel it.
/// - Best practice: inner tasks should use the [`ConsumerContext`] cancel token (or a child)
///   so they respond to runtime shutdown without waiting for `stop()`.
/// - If using a private token, `stop()` must cancel it to ensure prompt cleanup.
/// - `background_task_handle()` returns the `JoinHandle` of the primary background task,
///   if any. The Runtime monitors this handle for unexpected exits (crash propagation).
#[async_trait]
pub trait Consumer: Send + Sync {
    /// Start consuming messages, sending them through the provided context.
    async fn start(&mut self, context: ConsumerContext) -> Result<(), CamelError>;

    /// Stop consuming messages and clean up all resources.
    ///
    /// Called by the Runtime on every exit path after `start()` succeeds.
    /// See the [Shutdown Contract](#shutdown-contract) above.
    async fn stop(&mut self) -> Result<(), CamelError>;

    /// Temporarily suspend consuming messages without fully stopping.
    ///
    /// Default: no-op, returns `Ok(())`.
    async fn suspend(&self) -> Result<(), CamelError> {
        Ok(())
    }

    /// Resume consuming after a previous suspension.
    ///
    /// Default: no-op, returns `Ok(())`.
    async fn resume(&self) -> Result<(), CamelError> {
        Ok(())
    }

    /// Declares this consumer's natural concurrency model.
    ///
    /// The runtime uses this to decide whether to process exchanges
    /// sequentially or spawn per-exchange. Consumers that accept inbound
    /// connections (HTTP, WebSocket, Kafka) should override this to return
    /// `ConcurrencyModel::Concurrent`.
    ///
    /// Default: `Sequential`.
    fn concurrency_model(&self) -> ConcurrencyModel {
        ConcurrencyModel::Sequential
    }

    /// Return a handle to the consumer's long-running background task so the
    /// runtime can monitor it for unexpected exits after `start()` returns `Ok`.
    ///
    /// Default: `None` — consumer's work completes entirely within `start()`.
    /// Override: return `Some(handle)` if `start()` spawns a detached task.
    ///
    /// **Contract:** the task must observe `ConsumerContext::cancelled()` so
    /// runtime shutdown is distinguishable from crash exits.
    ///
    /// This method is called at most once; implementations should use `.take()`
    /// to transfer ownership of the handle.
    fn background_task_handle(&mut self) -> Option<JoinHandle<Result<(), CamelError>>> {
        None
    }

    /// Set the security context for this consumer.
    ///
    /// Called by the route controller before `start()` so the consumer
    /// can register auth state (e.g. WebSocket auth in `WsAppState`).
    ///
    /// Default: no-op, returns `Ok(())`.
    fn set_security_context(&mut self, _ctx: SecurityContext) {}
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn consumer_context_exposes_route_id() {
        let (tx, _rx) = mpsc::channel(1);
        let ctx = ConsumerContext::new(tx, CancellationToken::new(), "test-route".to_string());
        assert_eq!(ctx.route_id(), "test-route");
    }

    #[tokio::test]
    async fn test_consumer_context_cancelled() {
        let (tx, _rx) = mpsc::channel(16);
        let token = CancellationToken::new();
        let ctx = ConsumerContext::new(tx, token.clone(), "test-route".to_string());

        assert!(!ctx.is_cancelled());
        token.cancel();
        ctx.cancelled().await;
        assert!(ctx.is_cancelled());
    }

    #[test]
    fn test_concurrency_model_default_is_sequential() {
        use super::ConcurrencyModel;

        struct DummyConsumer;

        #[async_trait::async_trait]
        impl super::Consumer for DummyConsumer {
            async fn start(&mut self, _ctx: super::ConsumerContext) -> Result<(), CamelError> {
                Ok(())
            }
            async fn stop(&mut self) -> Result<(), CamelError> {
                Ok(())
            }
        }

        let consumer = DummyConsumer;
        assert_eq!(consumer.concurrency_model(), ConcurrencyModel::Sequential);
    }

    #[test]
    fn test_concurrency_model_concurrent_override() {
        use super::ConcurrencyModel;

        struct ConcurrentConsumer;

        #[async_trait::async_trait]
        impl super::Consumer for ConcurrentConsumer {
            async fn start(&mut self, _ctx: super::ConsumerContext) -> Result<(), CamelError> {
                Ok(())
            }
            async fn stop(&mut self) -> Result<(), CamelError> {
                Ok(())
            }
            fn concurrency_model(&self) -> ConcurrencyModel {
                ConcurrencyModel::Concurrent { max: Some(16) }
            }
        }

        let consumer = ConcurrentConsumer;
        assert_eq!(
            consumer.concurrency_model(),
            ConcurrencyModel::Concurrent { max: Some(16) }
        );
    }

    #[tokio::test]
    async fn test_consumer_default_suspend_resume() {
        struct DummyConsumer;

        #[async_trait::async_trait]
        impl super::Consumer for DummyConsumer {
            async fn start(&mut self, _ctx: super::ConsumerContext) -> Result<(), CamelError> {
                Ok(())
            }
            async fn stop(&mut self) -> Result<(), CamelError> {
                Ok(())
            }
        }

        let consumer = DummyConsumer;
        assert!(consumer.suspend().await.is_ok());
        assert!(consumer.resume().await.is_ok());
    }

    // --- SecurityContext tests ---

    struct StubPolicy;

    #[async_trait::async_trait]
    impl SecurityPolicy for StubPolicy {
        async fn evaluate(
            &self,
            _exchange: &mut Exchange,
        ) -> Result<camel_api::security_policy::AuthorizationDecision, CamelError> {
            Ok(camel_api::security_policy::AuthorizationDecision::Granted {
                principal: camel_api::security_policy::Principal {
                    subject: "stub".into(),
                    issuer: "stub".into(),
                    audience: vec![],
                    scopes: vec![],
                    roles: vec![],
                    claims: serde_json::json!({}),
                },
            })
        }
    }

    struct StubAuthenticator;

    #[async_trait::async_trait]
    impl camel_auth::TokenAuthenticator for StubAuthenticator {
        async fn authenticate_bearer(
            &self,
            _token: &str,
        ) -> Result<camel_api::security_policy::Principal, CamelError> {
            Ok(camel_api::security_policy::Principal {
                subject: "stub".into(),
                issuer: "stub".into(),
                audience: vec![],
                scopes: vec![],
                roles: vec![],
                claims: serde_json::json!({}),
            })
        }
    }

    #[test]
    fn test_security_context_new() {
        let ctx = SecurityContext::new(StubPolicy, Arc::new(StubAuthenticator));
        assert!(Arc::strong_count(&ctx.policy) == 1);
        assert!(Arc::strong_count(&ctx.authenticator) == 1);
        assert_eq!(
            ctx.credential_sources,
            vec![camel_auth::CredentialSource::AuthorizationHeader]
        );
    }

    #[test]
    fn test_security_context_from_arc() {
        let policy: Arc<dyn SecurityPolicy> = Arc::new(StubPolicy);
        let authenticator: Arc<dyn camel_auth::TokenAuthenticator> = Arc::new(StubAuthenticator);
        let ctx = SecurityContext::from_arc(Arc::clone(&policy), Arc::clone(&authenticator));
        assert!(Arc::ptr_eq(&ctx.policy, &policy));
        assert!(Arc::ptr_eq(&ctx.authenticator, &authenticator));
        assert_eq!(
            ctx.credential_sources,
            vec![camel_auth::CredentialSource::AuthorizationHeader]
        );
    }

    #[test]
    fn test_security_context_clone_independent() {
        let ctx = SecurityContext::new(StubPolicy, Arc::new(StubAuthenticator));
        let cloned = ctx.clone();
        assert!(Arc::ptr_eq(&ctx.policy, &cloned.policy));
        assert!(Arc::ptr_eq(&ctx.authenticator, &cloned.authenticator));
        assert_eq!(ctx.credential_sources, cloned.credential_sources);
    }

    #[test]
    fn test_security_context_debug_redacts_traits() {
        let ctx = SecurityContext::new(StubPolicy, Arc::new(StubAuthenticator));
        let debug_str = format!("{ctx:?}");
        assert!(debug_str.contains("<SecurityPolicy>"));
        assert!(debug_str.contains("<TokenAuthenticator>"));
        assert!(debug_str.contains("credential_sources"));
    }

    #[test]
    fn test_security_context_with_credential_sources() {
        let ctx = SecurityContext::new(StubPolicy, Arc::new(StubAuthenticator))
            .with_credential_sources(vec![
                camel_auth::CredentialSource::Cookie {
                    name: "session".into(),
                },
                camel_auth::CredentialSource::AuthorizationHeader,
            ]);
        assert_eq!(ctx.credential_sources.len(), 2);
        assert!(matches!(
            &ctx.credential_sources[0],
            camel_auth::CredentialSource::Cookie { .. }
        ));
    }
}