camel_api/

exchange.rs

use std::any::Any;
use std::collections::HashMap;
use std::sync::Arc;

use opentelemetry::Context;
use uuid::Uuid;

use crate::error::CamelError;
use crate::from_body::FromBody;
use crate::message::Message;
use crate::value::Value;

/// Extension key used by use_original_message: stashes the pre-route Message
/// so the error handler can restore body+headers before DLC dispatch.
pub const ORIGINAL_MESSAGE_EXTENSION: &str = "CamelOriginalMessage";

/// Extension key for the Content-Type provenance snapshot taken with
/// `ORIGINAL_MESSAGE_EXTENSION` at route entry when `use_original_message` is
/// enabled. Capture replaces the snapshot unconditionally; restore reads it to
/// rebuild the inherited marker.
pub const ORIGINAL_CONTENT_TYPE_INHERITED_EXTENSION: &str = "CamelOriginalContentTypeInherited";

/// Marker-presence snapshot stored under `ORIGINAL_CONTENT_TYPE_INHERITED_EXTENSION`.
/// Stored as `Arc<OriginalContentTypeInherited>` via `Exchange::set_extension`.
pub struct OriginalContentTypeInherited(pub bool);

/// Exchange property marking that the current `Content-Type` was inherited from
/// the incoming request rather than written by a route step.
///
/// Set by the HTTP consumer at ingress and cleared by any route step that
/// writes or removes `Content-Type`, so downstream components can suppress
/// value-equality checks in favour of provenance.
pub const PROPERTY_HTTP_CONTENT_TYPE_INHERITED: &str = "CamelHttpContentTypeInherited";

/// Pipeline-executor-recognized stop signal. Set by processors (ThrottleStrategy::Drop,
/// SamplingService) that cannot return PipelineOutcome::Stopped directly because they
/// are Tower `Service<Exchange>` (Process mode), not OutcomePipeline (Segment).
/// The executor checks this after each step completion. See ADR-0024 amendment.
pub const CAMEL_STOP: &str = "CamelStop";

/// Check whether the exchange carries the CamelStop signal.
///
/// Returns `true` if the `CamelStop` property is set to a boolean `true` value.
/// Used by the pipeline executor (`run_steps`) and segment adapters to detect
/// Process-mode processors that cannot return `PipelineOutcome::Stopped` directly.
pub fn is_camel_stop(exchange: &Exchange) -> bool {
    exchange
        .property(CAMEL_STOP)
        .and_then(|v| v.as_bool())
        .unwrap_or(false)
}

/// Property key for the exception message (error Display string).
pub const PROPERTY_EXCEPTION_MESSAGE: &str = "CamelExceptionMessage";
/// Property key for the exception kind (error variant name via classify()).
pub const PROPERTY_EXCEPTION_KIND: &str = "CamelExceptionKind";
/// Property key for the caught exception (Java Camel parity alias).
pub const PROPERTY_EXCEPTION_CAUGHT: &str = "CamelExceptionCaught";
/// Property key set when an exception has been handled (Java Camel parity).
pub const PROPERTY_EXCEPTION_HANDLED: &str = "CamelExceptionHandled";

/// The exchange pattern (fire-and-forget or request-reply).
/// exhaustive-by-contract: the InOnly|InOut MEP dichotomy is a fixed, spec-level closed set.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ExchangePattern {
    /// Fire-and-forget: message sent, no reply expected.
    #[default]
    InOnly,
    /// Request-reply: message sent, reply expected.
    InOut,
}

/// An Exchange represents a message being routed through the framework.
///
/// It contains the input message, an optional output message,
/// properties for passing data between processors, and error state.
#[derive(Debug)]
pub struct Exchange {
    /// The input (incoming) message.
    pub input: Message,
    /// The output (response) message, populated for InOut patterns.
    pub output: Option<Message>,
    /// Exchange-scoped properties for passing data between processors.
    pub properties: HashMap<String, Value>,
    /// Non-serializable extension values (e.g., channel senders).
    /// Stored as `Arc<dyn Any + Send + Sync>` so cloning is cheap (ref-count bump).
    pub extensions: HashMap<String, Arc<dyn Any + Send + Sync>>,
    /// Error state, if processing failed.
    pub error: Option<CamelError>,
    /// The exchange pattern.
    pub pattern: ExchangePattern,
    /// Unique correlation ID for distributed tracing.
    pub correlation_id: String,
    /// OpenTelemetry context for distributed tracing propagation.
    /// Carries the active span context between processing steps.
    /// Defaults to an empty context (noop span) if OTel is not active.
    pub otel_context: Context,
    /// RAII claim on the context-global in-flight gauge — the
    /// accepted-not-completed counter (drainclaim, claimfamily). Pipeline
    /// drain sites split a sibling
    /// of the envelope's claim onto the exchange so residency inside
    /// pipeline-embedded stash sites (resequencer buffers, aggregator
    /// buckets — raw-`Exchange` territory with no envelope) stays
    /// counted. `None` on every constructor, on clones, and on
    /// exchanges accepted through uncounted paths.
    pub in_flight_claim: Option<crate::in_flight::InFlightClaim>,
}

impl Exchange {
    /// Create a new exchange with the given input message.
    pub fn new(input: Message) -> Self {
        Self {
            input,
            output: None,
            properties: HashMap::new(),
            extensions: HashMap::new(),
            error: None,
            pattern: ExchangePattern::default(),
            correlation_id: Uuid::new_v4().to_string(),
            otel_context: Context::new(),
            in_flight_claim: None,
        }
    }

    /// Create a new exchange with the InOut pattern.
    pub fn new_in_out(input: Message) -> Self {
        Self {
            input,
            output: None,
            properties: HashMap::new(),
            extensions: HashMap::new(),
            error: None,
            pattern: ExchangePattern::InOut,
            correlation_id: Uuid::new_v4().to_string(),
            otel_context: Context::new(),
            in_flight_claim: None,
        }
    }

    /// Get the correlation ID for this exchange.
    pub fn correlation_id(&self) -> &str {
        &self.correlation_id
    }

    /// Get a property value.
    pub fn property(&self, key: &str) -> Option<&Value> {
        self.properties.get(key)
    }

    /// Set a property value.
    pub fn set_property(&mut self, key: impl Into<String>, value: impl Into<Value>) {
        self.properties.insert(key.into(), value.into());
    }

    /// Remove an exchange property. Idempotent: removing an absent key is a no-op.
    /// Used by provenance writers to clear `PROPERTY_HTTP_CONTENT_TYPE_INHERITED`.
    pub fn remove_property(&mut self, key: &str) {
        self.properties.remove(key);
    }

    /// Check if the exchange has an error.
    pub fn has_error(&self) -> bool {
        self.error.is_some()
    }

    /// Set an error on this exchange.
    ///
    /// Automatically populates exchange properties with error context so that
    /// all languages can access error information via their property mechanisms:
    /// - `CamelExceptionMessage` — the error's Display string
    /// - `CamelExceptionKind` — the error variant name (via `CamelError::classify()`)
    /// - `CamelExceptionCaught` — alias for Java Camel parity
    ///
    /// For `HttpOperationFailed` it also sets the `CamelHttpResponseCode` and
    /// `CamelHttpResponseText` headers, because a failing producer loses its
    /// exchange with the `Err` and error-handling expressions would otherwise
    /// have no way to read the status.
    pub fn set_error(&mut self, error: CamelError) {
        if let CamelError::HttpOperationFailed {
            status_code,
            status_text,
            ..
        } = &error
        {
            self.input.set_header(
                "CamelHttpResponseCode",
                Value::Number((*status_code).into()),
            );
            self.input
                .set_header("CamelHttpResponseText", Value::String(status_text.clone()));
        }
        let msg = error.to_string();
        let kind = error.classify().to_string();
        self.properties.insert(
            PROPERTY_EXCEPTION_MESSAGE.to_string(),
            Value::String(msg.clone()),
        );
        self.properties
            .insert(PROPERTY_EXCEPTION_KIND.to_string(), Value::String(kind));
        self.properties
            .insert(PROPERTY_EXCEPTION_CAUGHT.to_string(), Value::String(msg));
        self.error = Some(error);
    }

    /// Clear the error and remove all exception properties.
    ///
    /// Called after `handled:true` in on_exception steps to prevent stale
    /// error state from leaking into subsequent processing.
    pub fn clear_error(&mut self) {
        self.error = None;
        self.properties.remove(PROPERTY_EXCEPTION_MESSAGE);
        self.properties.remove(PROPERTY_EXCEPTION_KIND);
        self.properties.remove(PROPERTY_EXCEPTION_CAUGHT);
    }

    /// Mark the exception as handled and clear the error.
    ///
    /// Sets `CamelExceptionHandled = true` then calls `clear_error()`.
    /// This matches Java Camel's `Exchange.EXCEPTION_HANDLED` semantics.
    pub fn handle_error(&mut self) {
        self.properties
            .insert(PROPERTY_EXCEPTION_HANDLED.to_string(), Value::Bool(true));
        self.clear_error();
    }

    /// Store a non-serializable extension value (e.g. a channel sender).
    pub fn set_extension(&mut self, key: impl Into<String>, value: Arc<dyn Any + Send + Sync>) {
        self.extensions.insert(key.into(), value);
    }

    /// Retrieve a typed extension value. Returns `None` if the key is absent
    /// or the stored value is not of type `T`.
    pub fn get_extension<T: Any>(&self, key: &str) -> Option<&T> {
        self.extensions.get(key)?.downcast_ref::<T>()
    }

    /// Deserialize the body into type `T`.
    ///
    /// Uses built-in conversions for `String`, `Vec<u8>`, [`bytes::Bytes`], and
    /// `serde_json::Value`. For custom types, implement [`FromBody`] or use
    /// the `impl_from_body_via_serde!` macro.
    ///
    /// # Example
    /// ```rust,ignore
    /// let text: String = exchange.body_as::<String>()?;
    /// let raw: Vec<u8> = exchange.body_as::<Vec<u8>>()?;
    /// ```
    pub fn body_as<T: FromBody>(&self) -> Result<T, CamelError> {
        T::from_body(&self.input.body)
    }

    /// Clone for a fanout branch copy. Unlike [`Exchange::clone`], a
    /// carried in-flight claim is split onto the copy (claimfamily):
    /// the sibling counts the copy independently and releases when the
    /// copy completes; the original claim stays live.
    pub fn branch_clone(&self) -> Self {
        let mut copy = self.clone();
        copy.in_flight_claim = self.in_flight_claim.as_ref().map(|claim| claim.split());
        copy
    }
}

impl Clone for Exchange {
    fn clone(&self) -> Self {
        Self {
            input: self.input.clone(),
            output: self.output.clone(),
            properties: self.properties.clone(),
            extensions: self.extensions.clone(), // Arc ref-count bump, cheap
            error: self.error.clone(),
            pattern: self.pattern,
            correlation_id: self.correlation_id.clone(),
            otel_context: self.otel_context.clone(),
            // claimfamily: clones carry no claim — duplicating one would
            // double-release on drop. Fanout sites that need each copy
            // counted split a sibling explicitly.
            in_flight_claim: None,
        }
    }
}

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

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Body;
    use crate::in_flight::{InFlightClaim, InFlightGauge};
    use serde_json::json;
    use std::sync::Arc;

    #[test]
    fn remove_property_removes_and_is_idempotent() {
        let mut ex = Exchange::new(Message::default());
        ex.set_property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED, Value::Bool(true));
        assert_eq!(
            ex.property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED),
            Some(&Value::Bool(true))
        );

        ex.remove_property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED);
        assert_eq!(ex.property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED), None);

        ex.remove_property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED);
        assert_eq!(ex.property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED), None);
    }

    #[test]
    fn remove_property_absent_is_noop() {
        let mut ex = Exchange::new(Message::default());
        assert_eq!(ex.property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED), None);

        ex.remove_property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED);

        assert_eq!(ex.property(PROPERTY_HTTP_CONTENT_TYPE_INHERITED), None);
    }

    #[test]
    fn original_content_type_snapshot_roundtrips() {
        let mut ex = Exchange::new(Message::default());
        ex.set_extension(
            ORIGINAL_CONTENT_TYPE_INHERITED_EXTENSION,
            Arc::new(OriginalContentTypeInherited(true)),
        );

        let snapshot = ex
            .get_extension::<OriginalContentTypeInherited>(
                ORIGINAL_CONTENT_TYPE_INHERITED_EXTENSION,
            )
            .expect("snapshot stored under its extension key");
        assert!(snapshot.0);

        let fresh = Exchange::new(Message::default());
        assert!(
            fresh
                .get_extension::<OriginalContentTypeInherited>(
                    ORIGINAL_CONTENT_TYPE_INHERITED_EXTENSION
                )
                .is_none()
        );
    }

    #[test]
    fn test_exchange_new() {
        let msg = Message::new("test");
        let ex = Exchange::new(msg);
        assert_eq!(ex.input.body.as_text(), Some("test"));
        assert!(ex.output.is_none());
        assert!(!ex.has_error());
        assert_eq!(ex.pattern, ExchangePattern::InOnly);
    }

    #[test]
    fn test_exchange_in_out() {
        let ex = Exchange::new_in_out(Message::default());
        assert_eq!(ex.pattern, ExchangePattern::InOut);
    }

    #[test]
    fn test_exchange_properties() {
        let mut ex = Exchange::default();
        ex.set_property("key", Value::Bool(true));
        assert_eq!(ex.property("key"), Some(&Value::Bool(true)));
        assert_eq!(ex.property("missing"), None);
    }

    #[test]
    fn test_exchange_error() {
        let mut ex = Exchange::default();
        assert!(!ex.has_error());
        ex.set_error(CamelError::ProcessorError("test".into()));
        assert!(ex.has_error());
    }

    #[test]
    fn test_set_error_populates_properties() {
        let mut ex = Exchange::default();
        ex.set_error(CamelError::ProcessorError("boom".into()));

        assert!(ex.has_error());
        assert_eq!(
            ex.properties.get(PROPERTY_EXCEPTION_MESSAGE),
            Some(&Value::String("Processor error: boom".to_string()))
        );
        assert_eq!(
            ex.properties.get(PROPERTY_EXCEPTION_KIND),
            Some(&Value::String("processor".to_string()))
        );
        assert_eq!(
            ex.properties.get(PROPERTY_EXCEPTION_CAUGHT),
            Some(&Value::String("Processor error: boom".to_string()))
        );
    }

    #[test]
    fn test_set_error_http_failure_exposes_status_headers() {
        let mut ex = Exchange::default();
        ex.set_error(CamelError::HttpOperationFailed {
            method: "GET".into(),
            url: "http://example.test/x".into(),
            status_code: 404,
            status_text: "Not Found".into(),
            response_body: None,
        });

        assert_eq!(
            ex.input.header("CamelHttpResponseCode"),
            Some(&Value::Number(404.into()))
        );
        assert_eq!(
            ex.input.header("CamelHttpResponseText"),
            Some(&Value::String("Not Found".to_string()))
        );
    }

    #[test]
    fn test_set_error_non_http_leaves_http_headers_untouched() {
        let mut ex = Exchange::default();
        ex.set_error(CamelError::ProcessorError("boom".into()));

        assert!(ex.input.header("CamelHttpResponseCode").is_none());
        assert!(ex.input.header("CamelHttpResponseText").is_none());
    }

    #[test]
    fn test_clear_error_removes_properties() {
        let mut ex = Exchange::default();
        ex.set_error(CamelError::RouteError("fail".into()));
        assert!(ex.has_error());
        assert!(ex.properties.contains_key(PROPERTY_EXCEPTION_MESSAGE));

        ex.clear_error();

        assert!(!ex.has_error());
        assert!(!ex.properties.contains_key(PROPERTY_EXCEPTION_MESSAGE));
        assert!(!ex.properties.contains_key(PROPERTY_EXCEPTION_KIND));
        assert!(!ex.properties.contains_key(PROPERTY_EXCEPTION_CAUGHT));
    }

    #[test]
    fn test_exchange_lifecycle() {
        let mut ex = Exchange::new(Message::new("input data"));
        assert_eq!(ex.input.body.as_text(), Some("input data"));

        // Set some properties
        ex.set_property("processed", Value::Bool(true));

        // Set output
        ex.output = Some(Message::new("output data"));
        assert!(ex.output.is_some());

        // Verify no error
        assert!(!ex.has_error());
    }

    #[test]
    fn test_exchange_otel_context_default() {
        let ex = Exchange::default();
        // Field must exist and be accessible — compilation is the test
        // Also verify it's a fresh context (noop span)
        use opentelemetry::trace::TraceContextExt;
        assert!(!ex.otel_context.span().span_context().is_valid());
    }

    #[test]
    fn test_exchange_otel_context_propagates_in_clone() {
        let ex = Exchange::default();
        let cloned = ex.clone();
        // Both should have the same (empty) context
        use opentelemetry::trace::TraceContextExt;
        assert!(!cloned.otel_context.span().span_context().is_valid());
    }

    #[test]
    fn test_set_and_get_extension() {
        use std::sync::Arc;
        let mut ex = Exchange::default();
        ex.set_extension("my.key", Arc::new(42u32));
        let val: Option<&u32> = ex.get_extension("my.key");
        assert_eq!(val, Some(&42u32));
    }

    #[test]
    fn test_get_extension_wrong_type_returns_none() {
        use std::sync::Arc;
        let mut ex = Exchange::default();
        ex.set_extension("my.key", Arc::new(42u32));
        let val: Option<&String> = ex.get_extension("my.key");
        assert!(val.is_none());
    }

    #[test]
    fn test_get_extension_missing_key_returns_none() {
        let ex = Exchange::default();
        let val: Option<&u32> = ex.get_extension("nope");
        assert!(val.is_none());
    }

    #[test]
    fn test_clone_shares_extension_arc() {
        use std::sync::Arc;
        let mut ex = Exchange::default();
        ex.set_extension("shared", Arc::new(99u64));
        let cloned = ex.clone();
        // Both see the same value
        assert_eq!(ex.get_extension::<u64>("shared"), Some(&99u64));
        assert_eq!(cloned.get_extension::<u64>("shared"), Some(&99u64));
    }

    #[test]
    fn test_body_as_string_from_text() {
        let ex = Exchange::new(Message::new(Body::Text("hello".to_string())));

        let result = ex.body_as::<String>();

        assert_eq!(result.unwrap(), "hello");
    }

    #[test]
    fn test_body_as_string_from_json_string() {
        let ex = Exchange::new(Message::new(Body::Json(json!("hello"))));

        let result = ex.body_as::<String>();

        assert_eq!(result.unwrap(), "hello");
    }

    #[test]
    fn test_body_as_json_value_from_json_number() {
        let ex = Exchange::new(Message::new(Body::Json(json!(42))));

        let result = ex.body_as::<serde_json::Value>();

        assert_eq!(result.unwrap(), json!(42));
    }

    #[test]
    fn test_body_as_vec_u8_from_bytes() {
        let ex = Exchange::new(Message::new(Body::from(vec![1u8, 2, 3, 4])));

        let result = ex.body_as::<Vec<u8>>();

        assert_eq!(result.unwrap(), vec![1u8, 2, 3, 4]);
    }

    #[test]
    fn test_body_as_string_from_empty_returns_err() {
        let ex = Exchange::new(Message::new(Body::Empty));

        let result = ex.body_as::<String>();

        assert!(result.is_err());
    }

    #[test]
    fn branch_clone_splits_claim_original_unaffected() {
        let gauge = Arc::new(InFlightGauge::new());
        let mut src = Exchange::new(Message::default());
        src.in_flight_claim = Some(InFlightClaim::attach(&gauge));

        let copy = src.branch_clone();

        assert_eq!(gauge.total(), 2);
        assert!(copy.in_flight_claim.is_some());
        assert!(src.in_flight_claim.is_some());
    }

    #[test]
    fn branch_clone_sibling_releases_independently() {
        let gauge = Arc::new(InFlightGauge::new());
        let mut src = Exchange::new(Message::default());
        src.in_flight_claim = Some(InFlightClaim::attach(&gauge));

        let copy = src.branch_clone();
        drop(copy);
        assert_eq!(gauge.total(), 1);

        drop(src);
        assert_eq!(gauge.total(), 0);
    }

    #[test]
    fn plain_clone_carries_no_claim() {
        let gauge = Arc::new(InFlightGauge::new());
        let mut src = Exchange::new(Message::default());
        src.in_flight_claim = Some(InFlightClaim::attach(&gauge));
        assert_eq!(gauge.total(), 1);

        let copy = src.clone();

        assert_eq!(gauge.total(), 1);
        assert!(copy.in_flight_claim.is_none());
        assert!(src.in_flight_claim.is_some());
    }
}