camel_processor/
dynamic_set_header.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use tower::Service;
6
7use camel_api::{CamelError, Exchange, Value};
8
9/// A processor that sets a header using an expression closure.
10/// The closure receives the full exchange (read-only) and returns a Value.
11#[derive(Clone)]
12pub struct DynamicSetHeader<P, F> {
13    inner: P,
14    key: String,
15    expr: F,
16}
17
18impl<P, F> DynamicSetHeader<P, F>
19where
20    F: Fn(&Exchange) -> Value,
21{
22    pub fn new(inner: P, key: impl Into<String>, expr: F) -> Self {
23        Self {
24            inner,
25            key: key.into(),
26            expr,
27        }
28    }
29}
30
31/// A Tower Layer that wraps an inner service with a [`DynamicSetHeader`].
32#[derive(Clone)]
33pub struct DynamicSetHeaderLayer<F> {
34    key: String,
35    expr: F,
36}
37
38impl<F> DynamicSetHeaderLayer<F> {
39    pub fn new(key: impl Into<String>, expr: F) -> Self {
40        Self {
41            key: key.into(),
42            expr,
43        }
44    }
45}
46
47impl<S, F> tower::Layer<S> for DynamicSetHeaderLayer<F>
48where
49    F: Clone,
50{
51    type Service = DynamicSetHeader<S, F>;
52
53    fn layer(&self, inner: S) -> Self::Service {
54        DynamicSetHeader {
55            inner,
56            key: self.key.clone(),
57            expr: self.expr.clone(),
58        }
59    }
60}
61
62impl<P, F> Service<Exchange> for DynamicSetHeader<P, F>
63where
64    P: Service<Exchange, Response = Exchange, Error = CamelError> + Clone + Send + 'static,
65    P::Future: Send,
66    F: Fn(&Exchange) -> Value + Clone + Send + Sync + 'static,
67{
68    type Response = Exchange;
69    type Error = CamelError;
70    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
71
72    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
73        self.inner.poll_ready(cx)
74    }
75
76    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
77        let value = (self.expr)(&exchange);
78        exchange.input.headers.insert(self.key.clone(), value);
79        let fut = self.inner.call(exchange);
80        Box::pin(fut)
81    }
82}
83
84/// A processor that sets a header using an expression closure,
85/// but ONLY if the header is not already present (if-absent semantics).
86/// The presence check happens BEFORE expression evaluation — if the
87/// header exists, the expression is never called.
88#[derive(Clone)]
89pub struct DynamicSetHeaderIfAbsent<P, F> {
90    inner: P,
91    key: String,
92    expr: F,
93}
94
95impl<P, F> DynamicSetHeaderIfAbsent<P, F>
96where
97    F: Fn(&Exchange) -> Value,
98{
99    pub fn new(inner: P, key: impl Into<String>, expr: F) -> Self {
100        Self {
101            inner,
102            key: key.into(),
103            expr,
104        }
105    }
106}
107
108impl<P, F> Service<Exchange> for DynamicSetHeaderIfAbsent<P, F>
109where
110    P: Service<Exchange, Response = Exchange, Error = CamelError> + Clone + Send + 'static,
111    P::Future: Send,
112    F: Fn(&Exchange) -> Value + Clone + Send + Sync + 'static,
113{
114    type Response = Exchange;
115    type Error = CamelError;
116    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
117
118    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
119        self.inner.poll_ready(cx)
120    }
121
122    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
123        // Check BEFORE evaluating the expression.
124        if !exchange.input.headers.contains_key(&self.key) {
125            let value = (self.expr)(&exchange);
126            exchange.input.headers.insert(self.key.clone(), value);
127        }
128        let fut = self.inner.call(exchange);
129        Box::pin(fut)
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use camel_api::{Exchange, IdentityProcessor, Message, Value};
136    use tower::ServiceExt;
137
138    use super::*;
139
140    #[tokio::test]
141    async fn test_dynamic_set_header_from_body() {
142        let exchange = Exchange::new(Message::new("world"));
143
144        let svc = DynamicSetHeader::new(IdentityProcessor, "greeting", |ex: &Exchange| {
145            Value::String(format!("hello {}", ex.input.body.as_text().unwrap_or("")))
146        });
147
148        let result = svc.oneshot(exchange).await.unwrap();
149        assert_eq!(
150            result.input.header("greeting"),
151            Some(&Value::String("hello world".into()))
152        );
153    }
154
155    #[tokio::test]
156    async fn test_dynamic_set_header_overwrites_existing() {
157        let mut msg = Message::new("new");
158        msg.set_header("key", Value::String("old".into()));
159        let exchange = Exchange::new(msg);
160
161        let svc = DynamicSetHeader::new(IdentityProcessor, "key", |ex: &Exchange| {
162            Value::String(ex.input.body.as_text().unwrap_or("").into())
163        });
164
165        let result = svc.oneshot(exchange).await.unwrap();
166        assert_eq!(
167            result.input.header("key"),
168            Some(&Value::String("new".into()))
169        );
170    }
171
172    #[tokio::test]
173    async fn test_dynamic_set_header_preserves_body() {
174        let exchange = Exchange::new(Message::new("body content"));
175
176        let svc = DynamicSetHeader::new(IdentityProcessor, "len", |ex: &Exchange| {
177            let len = ex.input.body.as_text().map(|t| t.len() as i64).unwrap_or(0);
178            Value::Number(len.into())
179        });
180
181        let result = svc.oneshot(exchange).await.unwrap();
182        assert_eq!(result.input.body.as_text(), Some("body content"));
183        assert_eq!(result.input.header("len"), Some(&Value::Number(12.into())));
184    }
185
186    #[tokio::test]
187    async fn test_dynamic_set_header_layer_composes() {
188        use tower::ServiceBuilder;
189
190        let svc = ServiceBuilder::new()
191            .layer(DynamicSetHeaderLayer::new("computed", |_ex: &Exchange| {
192                Value::Bool(true)
193            }))
194            .service(IdentityProcessor);
195
196        let exchange = Exchange::new(Message::default());
197        let result = svc.oneshot(exchange).await.unwrap();
198        assert_eq!(result.input.header("computed"), Some(&Value::Bool(true)));
199    }
200
201    // ── DynamicSetHeaderIfAbsent ──
202
203    #[tokio::test]
204    async fn test_dynamic_set_header_if_absent_adds_when_missing() {
205        let exchange = Exchange::new(Message::new("world"));
206
207        let svc = DynamicSetHeaderIfAbsent::new(IdentityProcessor, "greeting", |ex: &Exchange| {
208            Value::String(format!("hello {}", ex.input.body.as_text().unwrap_or("")))
209        });
210
211        let result = svc.oneshot(exchange).await.unwrap();
212        assert_eq!(
213            result.input.header("greeting"),
214            Some(&Value::String("hello world".into()))
215        );
216    }
217
218    #[tokio::test]
219    async fn test_dynamic_set_header_if_absent_preserves_existing() {
220        use std::sync::Arc;
221        use std::sync::atomic::{AtomicUsize, Ordering};
222
223        let mut msg = Message::new("computed");
224        msg.set_header("key", Value::String("original".into()));
225        let exchange = Exchange::new(msg);
226
227        // Side-effect counter: expression increments this.
228        // If the header is present, the expression must NEVER be called.
229        let call_count = Arc::new(AtomicUsize::new(0));
230        let cc = call_count.clone();
231
232        let svc = DynamicSetHeaderIfAbsent::new(IdentityProcessor, "key", move |ex: &Exchange| {
233            cc.fetch_add(1, Ordering::SeqCst);
234            Value::String(ex.input.body.as_text().unwrap_or("").into())
235        });
236
237        let result = svc.oneshot(exchange).await.unwrap();
238        assert_eq!(
239            result.input.header("key"),
240            Some(&Value::String("original".into()))
241        );
242        assert_eq!(
243            call_count.load(Ordering::SeqCst),
244            0,
245            "expression must NOT be evaluated when header is present"
246        );
247    }
248
249    #[tokio::test]
250    async fn test_dynamic_set_header_if_absent_preserves_body() {
251        let exchange = Exchange::new(Message::new("body content"));
252
253        let svc = DynamicSetHeaderIfAbsent::new(IdentityProcessor, "len", |ex: &Exchange| {
254            let len = ex.input.body.as_text().map(|t| t.len() as i64).unwrap_or(0);
255            Value::Number(len.into())
256        });
257
258        let result = svc.oneshot(exchange).await.unwrap();
259        assert_eq!(result.input.body.as_text(), Some("body content"));
260        assert_eq!(result.input.header("len"), Some(&Value::Number(12.into())));
261    }
262}
camel_processor/dynamic_set_header.rs

camel_processor/
dynamic_set_header.rs
