camel_api/
exchange.rs

1use std::any::Any;
2use std::collections::HashMap;
3use std::sync::Arc;
4
5use opentelemetry::Context;
6use uuid::Uuid;
7
8use crate::error::CamelError;
9use crate::from_body::FromBody;
10use crate::message::Message;
11use crate::value::Value;
12
13/// Extension key used by use_original_message: stashes the pre-route Message
14/// so the error handler can restore body+headers before DLC dispatch.
15pub const ORIGINAL_MESSAGE_EXTENSION: &str = "CamelOriginalMessage";
16
17/// Pipeline-executor-recognized stop signal. Set by processors (ThrottleStrategy::Drop,
18/// SamplingService) that cannot return PipelineOutcome::Stopped directly because they
19/// are Tower `Service<Exchange>` (Process mode), not OutcomePipeline (Segment).
20/// The executor checks this after each step completion. See ADR-0024 amendment.
21pub const CAMEL_STOP: &str = "CamelStop";
22
23/// Check whether the exchange carries the CamelStop signal.
24///
25/// Returns `true` if the `CamelStop` property is set to a boolean `true` value.
26/// Used by the pipeline executor (`run_steps`) and segment adapters to detect
27/// Process-mode processors that cannot return `PipelineOutcome::Stopped` directly.
28pub fn is_camel_stop(exchange: &Exchange) -> bool {
29    exchange
30        .property(CAMEL_STOP)
31        .and_then(|v| v.as_bool())
32        .unwrap_or(false)
33}
34
35/// Property key for the exception message (error Display string).
36pub const PROPERTY_EXCEPTION_MESSAGE: &str = "CamelExceptionMessage";
37/// Property key for the exception kind (error variant name via classify()).
38pub const PROPERTY_EXCEPTION_KIND: &str = "CamelExceptionKind";
39/// Property key for the caught exception (Java Camel parity alias).
40pub const PROPERTY_EXCEPTION_CAUGHT: &str = "CamelExceptionCaught";
41/// Property key set when an exception has been handled (Java Camel parity).
42pub const PROPERTY_EXCEPTION_HANDLED: &str = "CamelExceptionHandled";
43
44/// The exchange pattern (fire-and-forget or request-reply).
45/// exhaustive-by-contract: the InOnly|InOut MEP dichotomy is a fixed, spec-level closed set.
46#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
47pub enum ExchangePattern {
48    /// Fire-and-forget: message sent, no reply expected.
49    #[default]
50    InOnly,
51    /// Request-reply: message sent, reply expected.
52    InOut,
53}
54
55/// An Exchange represents a message being routed through the framework.
56///
57/// It contains the input message, an optional output message,
58/// properties for passing data between processors, and error state.
59#[derive(Debug)]
60pub struct Exchange {
61    /// The input (incoming) message.
62    pub input: Message,
63    /// The output (response) message, populated for InOut patterns.
64    pub output: Option<Message>,
65    /// Exchange-scoped properties for passing data between processors.
66    pub properties: HashMap<String, Value>,
67    /// Non-serializable extension values (e.g., channel senders).
68    /// Stored as `Arc<dyn Any + Send + Sync>` so cloning is cheap (ref-count bump).
69    pub extensions: HashMap<String, Arc<dyn Any + Send + Sync>>,
70    /// Error state, if processing failed.
71    pub error: Option<CamelError>,
72    /// The exchange pattern.
73    pub pattern: ExchangePattern,
74    /// Unique correlation ID for distributed tracing.
75    pub correlation_id: String,
76    /// OpenTelemetry context for distributed tracing propagation.
77    /// Carries the active span context between processing steps.
78    /// Defaults to an empty context (noop span) if OTel is not active.
79    pub otel_context: Context,
80}
81
82impl Exchange {
83    /// Create a new exchange with the given input message.
84    pub fn new(input: Message) -> Self {
85        Self {
86            input,
87            output: None,
88            properties: HashMap::new(),
89            extensions: HashMap::new(),
90            error: None,
91            pattern: ExchangePattern::default(),
92            correlation_id: Uuid::new_v4().to_string(),
93            otel_context: Context::new(),
94        }
95    }
96
97    /// Create a new exchange with the InOut pattern.
98    pub fn new_in_out(input: Message) -> Self {
99        Self {
100            input,
101            output: None,
102            properties: HashMap::new(),
103            extensions: HashMap::new(),
104            error: None,
105            pattern: ExchangePattern::InOut,
106            correlation_id: Uuid::new_v4().to_string(),
107            otel_context: Context::new(),
108        }
109    }
110
111    /// Get the correlation ID for this exchange.
112    pub fn correlation_id(&self) -> &str {
113        &self.correlation_id
114    }
115
116    /// Get a property value.
117    pub fn property(&self, key: &str) -> Option<&Value> {
118        self.properties.get(key)
119    }
120
121    /// Set a property value.
122    pub fn set_property(&mut self, key: impl Into<String>, value: impl Into<Value>) {
123        self.properties.insert(key.into(), value.into());
124    }
125
126    /// Check if the exchange has an error.
127    pub fn has_error(&self) -> bool {
128        self.error.is_some()
129    }
130
131    /// Set an error on this exchange.
132    ///
133    /// Automatically populates exchange properties with error context so that
134    /// all languages can access error information via their property mechanisms:
135    /// - `CamelExceptionMessage` — the error's Display string
136    /// - `CamelExceptionKind` — the error variant name (via `CamelError::classify()`)
137    /// - `CamelExceptionCaught` — alias for Java Camel parity
138    pub fn set_error(&mut self, error: CamelError) {
139        let msg = error.to_string();
140        let kind = error.classify().to_string();
141        self.properties.insert(
142            PROPERTY_EXCEPTION_MESSAGE.to_string(),
143            Value::String(msg.clone()),
144        );
145        self.properties
146            .insert(PROPERTY_EXCEPTION_KIND.to_string(), Value::String(kind));
147        self.properties
148            .insert(PROPERTY_EXCEPTION_CAUGHT.to_string(), Value::String(msg));
149        self.error = Some(error);
150    }
151
152    /// Clear the error and remove all exception properties.
153    ///
154    /// Called after `handled:true` in on_exception steps to prevent stale
155    /// error state from leaking into subsequent processing.
156    pub fn clear_error(&mut self) {
157        self.error = None;
158        self.properties.remove(PROPERTY_EXCEPTION_MESSAGE);
159        self.properties.remove(PROPERTY_EXCEPTION_KIND);
160        self.properties.remove(PROPERTY_EXCEPTION_CAUGHT);
161    }
162
163    /// Mark the exception as handled and clear the error.
164    ///
165    /// Sets `CamelExceptionHandled = true` then calls `clear_error()`.
166    /// This matches Java Camel's `Exchange.EXCEPTION_HANDLED` semantics.
167    pub fn handle_error(&mut self) {
168        self.properties
169            .insert(PROPERTY_EXCEPTION_HANDLED.to_string(), Value::Bool(true));
170        self.clear_error();
171    }
172
173    /// Store a non-serializable extension value (e.g. a channel sender).
174    pub fn set_extension(&mut self, key: impl Into<String>, value: Arc<dyn Any + Send + Sync>) {
175        self.extensions.insert(key.into(), value);
176    }
177
178    /// Retrieve a typed extension value. Returns `None` if the key is absent
179    /// or the stored value is not of type `T`.
180    pub fn get_extension<T: Any>(&self, key: &str) -> Option<&T> {
181        self.extensions.get(key)?.downcast_ref::<T>()
182    }
183
184    /// Deserialize the body into type `T`.
185    ///
186    /// Uses built-in conversions for `String`, `Vec<u8>`, [`bytes::Bytes`], and
187    /// `serde_json::Value`. For custom types, implement [`FromBody`] or use
188    /// the `impl_from_body_via_serde!` macro.
189    ///
190    /// # Example
191    /// ```rust,ignore
192    /// let text: String = exchange.body_as::<String>()?;
193    /// let raw: Vec<u8> = exchange.body_as::<Vec<u8>>()?;
194    /// ```
195    pub fn body_as<T: FromBody>(&self) -> Result<T, CamelError> {
196        T::from_body(&self.input.body)
197    }
198}
199
200impl Clone for Exchange {
201    fn clone(&self) -> Self {
202        Self {
203            input: self.input.clone(),
204            output: self.output.clone(),
205            properties: self.properties.clone(),
206            extensions: self.extensions.clone(), // Arc ref-count bump, cheap
207            error: self.error.clone(),
208            pattern: self.pattern,
209            correlation_id: self.correlation_id.clone(),
210            otel_context: self.otel_context.clone(),
211        }
212    }
213}
214
215impl Default for Exchange {
216    fn default() -> Self {
217        Self::new(Message::default())
218    }
219}
220
221#[cfg(test)]
222mod tests {
223    use super::*;
224    use crate::Body;
225    use serde_json::json;
226
227    #[test]
228    fn test_exchange_new() {
229        let msg = Message::new("test");
230        let ex = Exchange::new(msg);
231        assert_eq!(ex.input.body.as_text(), Some("test"));
232        assert!(ex.output.is_none());
233        assert!(!ex.has_error());
234        assert_eq!(ex.pattern, ExchangePattern::InOnly);
235    }
236
237    #[test]
238    fn test_exchange_in_out() {
239        let ex = Exchange::new_in_out(Message::default());
240        assert_eq!(ex.pattern, ExchangePattern::InOut);
241    }
242
243    #[test]
244    fn test_exchange_properties() {
245        let mut ex = Exchange::default();
246        ex.set_property("key", Value::Bool(true));
247        assert_eq!(ex.property("key"), Some(&Value::Bool(true)));
248        assert_eq!(ex.property("missing"), None);
249    }
250
251    #[test]
252    fn test_exchange_error() {
253        let mut ex = Exchange::default();
254        assert!(!ex.has_error());
255        ex.set_error(CamelError::ProcessorError("test".into()));
256        assert!(ex.has_error());
257    }
258
259    #[test]
260    fn test_set_error_populates_properties() {
261        let mut ex = Exchange::default();
262        ex.set_error(CamelError::ProcessorError("boom".into()));
263
264        assert!(ex.has_error());
265        assert_eq!(
266            ex.properties.get(PROPERTY_EXCEPTION_MESSAGE),
267            Some(&Value::String("Processor error: boom".to_string()))
268        );
269        assert_eq!(
270            ex.properties.get(PROPERTY_EXCEPTION_KIND),
271            Some(&Value::String("processor".to_string()))
272        );
273        assert_eq!(
274            ex.properties.get(PROPERTY_EXCEPTION_CAUGHT),
275            Some(&Value::String("Processor error: boom".to_string()))
276        );
277    }
278
279    #[test]
280    fn test_clear_error_removes_properties() {
281        let mut ex = Exchange::default();
282        ex.set_error(CamelError::RouteError("fail".into()));
283        assert!(ex.has_error());
284        assert!(ex.properties.contains_key(PROPERTY_EXCEPTION_MESSAGE));
285
286        ex.clear_error();
287
288        assert!(!ex.has_error());
289        assert!(!ex.properties.contains_key(PROPERTY_EXCEPTION_MESSAGE));
290        assert!(!ex.properties.contains_key(PROPERTY_EXCEPTION_KIND));
291        assert!(!ex.properties.contains_key(PROPERTY_EXCEPTION_CAUGHT));
292    }
293
294    #[test]
295    fn test_exchange_lifecycle() {
296        let mut ex = Exchange::new(Message::new("input data"));
297        assert_eq!(ex.input.body.as_text(), Some("input data"));
298
299        // Set some properties
300        ex.set_property("processed", Value::Bool(true));
301
302        // Set output
303        ex.output = Some(Message::new("output data"));
304        assert!(ex.output.is_some());
305
306        // Verify no error
307        assert!(!ex.has_error());
308    }
309
310    #[test]
311    fn test_exchange_otel_context_default() {
312        let ex = Exchange::default();
313        // Field must exist and be accessible — compilation is the test
314        // Also verify it's a fresh context (noop span)
315        use opentelemetry::trace::TraceContextExt;
316        assert!(!ex.otel_context.span().span_context().is_valid());
317    }
318
319    #[test]
320    fn test_exchange_otel_context_propagates_in_clone() {
321        let ex = Exchange::default();
322        let cloned = ex.clone();
323        // Both should have the same (empty) context
324        use opentelemetry::trace::TraceContextExt;
325        assert!(!cloned.otel_context.span().span_context().is_valid());
326    }
327
328    #[test]
329    fn test_set_and_get_extension() {
330        use std::sync::Arc;
331        let mut ex = Exchange::default();
332        ex.set_extension("my.key", Arc::new(42u32));
333        let val: Option<&u32> = ex.get_extension("my.key");
334        assert_eq!(val, Some(&42u32));
335    }
336
337    #[test]
338    fn test_get_extension_wrong_type_returns_none() {
339        use std::sync::Arc;
340        let mut ex = Exchange::default();
341        ex.set_extension("my.key", Arc::new(42u32));
342        let val: Option<&String> = ex.get_extension("my.key");
343        assert!(val.is_none());
344    }
345
346    #[test]
347    fn test_get_extension_missing_key_returns_none() {
348        let ex = Exchange::default();
349        let val: Option<&u32> = ex.get_extension("nope");
350        assert!(val.is_none());
351    }
352
353    #[test]
354    fn test_clone_shares_extension_arc() {
355        use std::sync::Arc;
356        let mut ex = Exchange::default();
357        ex.set_extension("shared", Arc::new(99u64));
358        let cloned = ex.clone();
359        // Both see the same value
360        assert_eq!(ex.get_extension::<u64>("shared"), Some(&99u64));
361        assert_eq!(cloned.get_extension::<u64>("shared"), Some(&99u64));
362    }
363
364    #[test]
365    fn test_body_as_string_from_text() {
366        let ex = Exchange::new(Message::new(Body::Text("hello".to_string())));
367
368        let result = ex.body_as::<String>();
369
370        assert_eq!(result.unwrap(), "hello");
371    }
372
373    #[test]
374    fn test_body_as_string_from_json_string() {
375        let ex = Exchange::new(Message::new(Body::Json(json!("hello"))));
376
377        let result = ex.body_as::<String>();
378
379        assert_eq!(result.unwrap(), "hello");
380    }
381
382    #[test]
383    fn test_body_as_json_value_from_json_number() {
384        let ex = Exchange::new(Message::new(Body::Json(json!(42))));
385
386        let result = ex.body_as::<serde_json::Value>();
387
388        assert_eq!(result.unwrap(), json!(42));
389    }
390
391    #[test]
392    fn test_body_as_vec_u8_from_bytes() {
393        let ex = Exchange::new(Message::new(Body::from(vec![1u8, 2, 3, 4])));
394
395        let result = ex.body_as::<Vec<u8>>();
396
397        assert_eq!(result.unwrap(), vec![1u8, 2, 3, 4]);
398    }
399
400    #[test]
401    fn test_body_as_string_from_empty_returns_err() {
402        let ex = Exchange::new(Message::new(Body::Empty));
403
404        let result = ex.body_as::<String>();
405
406        assert!(result.is_err());
407    }
408}
camel_api/exchange.rs

camel_api/
exchange.rs
