camel_processor/
dynamic_set_property.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use tower::Service;
6
7use camel_api::{CamelError, Exchange, ValueSource};
8
9/// Sets an exchange property from a fallible [`ValueSource`].
10///
11/// The value source is evaluated first; a failed evaluation fails the step
12/// WITHOUT setting the property and WITHOUT invoking the inner service.
13#[derive(Clone)]
14pub struct DynamicSetProperty<P> {
15    inner: P,
16    key: String,
17    source: ValueSource,
18}
19
20impl<P> DynamicSetProperty<P> {
21    pub fn new(inner: P, key: impl Into<String>, source: impl Into<ValueSource>) -> Self {
22        Self {
23            inner,
24            key: key.into(),
25            source: source.into(),
26        }
27    }
28}
29
30#[derive(Clone)]
31pub struct DynamicSetPropertyLayer {
32    key: String,
33    source: ValueSource,
34}
35
36impl DynamicSetPropertyLayer {
37    pub fn new(key: impl Into<String>, source: impl Into<ValueSource>) -> Self {
38        Self {
39            key: key.into(),
40            source: source.into(),
41        }
42    }
43}
44
45impl<S> tower::Layer<S> for DynamicSetPropertyLayer {
46    type Service = DynamicSetProperty<S>;
47
48    fn layer(&self, inner: S) -> Self::Service {
49        DynamicSetProperty {
50            inner,
51            key: self.key.clone(),
52            source: self.source.clone(),
53        }
54    }
55}
56
57impl<P> Service<Exchange> for DynamicSetProperty<P>
58where
59    P: Service<Exchange, Response = Exchange, Error = CamelError> + Clone + Send + 'static,
60    P::Future: Send,
61{
62    type Response = Exchange;
63    type Error = CamelError;
64    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
65
66    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
67        self.inner.poll_ready(cx)
68    }
69
70    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
71        let source = self.source.clone();
72        let key = self.key.clone();
73        // Clone-and-replace: move the polled inner into the future so the
74        // async body owns its service; a fresh clone stays behind.
75        let clone = self.inner.clone();
76        let mut inner = std::mem::replace(&mut self.inner, clone);
77        Box::pin(async move {
78            let value = source.evaluate(&exchange).await?;
79            exchange.set_property(key, value);
80            inner.call(exchange).await
81        })
82    }
83}
84
85#[cfg(test)]
86mod tests {
87    use std::sync::Arc;
88    use std::sync::atomic::{AtomicUsize, Ordering};
89
90    use camel_api::{BoxValueFuture, Exchange, IdentityProcessor, Message, Value, ValueSource};
91    use tower::ServiceExt;
92
93    use super::*;
94
95    fn sync_source<F>(f: F) -> ValueSource
96    where
97        F: Fn(&Exchange) -> Value + Send + Sync + 'static,
98    {
99        ValueSource::Sync(Arc::new(f))
100    }
101
102    fn failing_source() -> ValueSource {
103        ValueSource::Async(Arc::new(|_: &Exchange| {
104            Box::pin(async { Err(CamelError::ProcessorError("expr boom".into())) })
105                as BoxValueFuture
106        }))
107    }
108
109    /// Inner service that records how many times it was invoked.
110    #[derive(Clone)]
111    struct CountingInner {
112        called: Arc<AtomicUsize>,
113    }
114
115    impl Service<Exchange> for CountingInner {
116        type Response = Exchange;
117        type Error = CamelError;
118        type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
119
120        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
121            Poll::Ready(Ok(()))
122        }
123
124        fn call(&mut self, exchange: Exchange) -> Self::Future {
125            self.called.fetch_add(1, Ordering::SeqCst);
126            Box::pin(async { Ok(exchange) })
127        }
128    }
129
130    #[tokio::test]
131    async fn dynamic_set_property_error_sets_nothing() {
132        let called = Arc::new(AtomicUsize::new(0));
133        let svc = DynamicSetProperty::new(
134            CountingInner {
135                called: Arc::clone(&called),
136            },
137            "greeting",
138            failing_source(),
139        );
140
141        let result = svc.oneshot(Exchange::new(Message::new("world"))).await;
142
143        assert!(result.is_err(), "failed evaluation must fail the step");
144        assert_eq!(
145            called.load(Ordering::SeqCst),
146            0,
147            "inner must NOT run when evaluation fails"
148        );
149    }
150
151    #[tokio::test(flavor = "current_thread")]
152    async fn setter_works_on_current_thread_runtime() {
153        let source = ValueSource::Async(Arc::new(|ex: &Exchange| {
154            let text = ex.input.body.as_text().unwrap_or("").to_string();
155            Box::pin(async move { Ok(Value::String(format!("hello {text}"))) }) as BoxValueFuture
156        }));
157        let svc = DynamicSetProperty::new(IdentityProcessor, "greeting", source);
158
159        let result = svc
160            .oneshot(Exchange::new(Message::new("world")))
161            .await
162            .unwrap();
163        assert_eq!(
164            result.property("greeting"),
165            Some(&Value::String("hello world".into()))
166        );
167    }
168
169    #[tokio::test]
170    async fn setter_poll_ready_delegates_to_inner() {
171        /// Inner service that reports Pending on its first poll and Ready on
172        /// every poll after, counting how many times it was polled.
173        #[derive(Clone)]
174        struct PendingOnceInner {
175            polls: Arc<AtomicUsize>,
176        }
177
178        impl Service<Exchange> for PendingOnceInner {
179            type Response = Exchange;
180            type Error = CamelError;
181            type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
182
183            fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
184                if self.polls.fetch_add(1, Ordering::SeqCst) == 0 {
185                    Poll::Pending
186                } else {
187                    Poll::Ready(Ok(()))
188                }
189            }
190
191            fn call(&mut self, exchange: Exchange) -> Self::Future {
192                Box::pin(async { Ok(exchange) })
193            }
194        }
195
196        let polls = Arc::new(AtomicUsize::new(0));
197        let mut svc = DynamicSetProperty::new(
198            PendingOnceInner {
199                polls: Arc::clone(&polls),
200            },
201            "k",
202            sync_source(|_: &Exchange| Value::String("v".into())),
203        );
204
205        let waker = futures::task::noop_waker();
206        let mut cx = Context::from_waker(&waker);
207
208        let first = Pin::new(&mut svc).poll_ready(&mut cx);
209        assert!(
210            first.is_pending(),
211            "first poll_ready must surface the inner Pending, got {first:?}"
212        );
213
214        let second = Pin::new(&mut svc).poll_ready(&mut cx);
215        assert!(
216            matches!(second, Poll::Ready(Ok(()))),
217            "second poll_ready must surface the inner Ready, got {second:?}"
218        );
219
220        let result = svc
221            .call(Exchange::new(Message::new("world")))
222            .await
223            .expect("call must complete through the inner service");
224        assert_eq!(result.property("k"), Some(&Value::String("v".into())));
225        assert_eq!(
226            polls.load(Ordering::SeqCst),
227            2,
228            "inner must be polled exactly twice before call"
229        );
230    }
231
232    #[tokio::test]
233    async fn test_dynamic_set_property_from_body() {
234        let exchange = Exchange::new(Message::new("world"));
235
236        let svc = DynamicSetProperty::new(
237            IdentityProcessor,
238            "greeting",
239            sync_source(|ex: &Exchange| {
240                Value::String(format!("hello {}", ex.input.body.as_text().unwrap_or("")))
241            }),
242        );
243
244        let result = svc.oneshot(exchange).await.unwrap();
245        assert_eq!(
246            result.property("greeting"),
247            Some(&Value::String("hello world".into()))
248        );
249    }
250
251    #[tokio::test]
252    async fn test_dynamic_set_property_overwrites_existing() {
253        let mut exchange = Exchange::new(Message::new("new"));
254        exchange.set_property("key", Value::String("old".into()));
255
256        let svc = DynamicSetProperty::new(
257            IdentityProcessor,
258            "key",
259            sync_source(|ex: &Exchange| {
260                Value::String(ex.input.body.as_text().unwrap_or("").into())
261            }),
262        );
263
264        let result = svc.oneshot(exchange).await.unwrap();
265        assert_eq!(result.property("key"), Some(&Value::String("new".into())));
266    }
267
268    #[tokio::test]
269    async fn test_dynamic_set_property_preserves_body() {
270        let exchange = Exchange::new(Message::new("body content"));
271
272        let svc = DynamicSetProperty::new(
273            IdentityProcessor,
274            "len",
275            sync_source(|ex: &Exchange| {
276                let len = ex.input.body.as_text().map(|t| t.len() as i64).unwrap_or(0);
277                Value::Number(len.into())
278            }),
279        );
280
281        let result = svc.oneshot(exchange).await.unwrap();
282        assert_eq!(result.input.body.as_text(), Some("body content"));
283        assert_eq!(result.property("len"), Some(&Value::Number(12.into())));
284    }
285
286    #[tokio::test]
287    async fn test_dynamic_set_property_layer_composes() {
288        use tower::ServiceBuilder;
289
290        let svc = ServiceBuilder::new()
291            .layer(DynamicSetPropertyLayer::new(
292                "computed",
293                sync_source(|_ex: &Exchange| Value::Bool(true)),
294            ))
295            .service(IdentityProcessor);
296
297        let exchange = Exchange::new(Message::default());
298        let result = svc.oneshot(exchange).await.unwrap();
299        assert_eq!(result.property("computed"), Some(&Value::Bool(true)));
300    }
301}
camel_processor/dynamic_set_property.rs

camel_processor/
dynamic_set_property.rs
