camel_processor/
validate.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use camel_api::{CamelError, Exchange, FilterPredicate};
6use tower::Service;
7
8/// Tower Service implementing the Validate EIP.
9///
10/// If the predicate returns `true`, the exchange continues (returned as `Ok`).
11/// If `false`, a `CamelError::ValidationError` is returned.
12#[derive(Clone)]
13pub struct ValidateService {
14    predicate: FilterPredicate,
15    expression_source: String,
16}
17
18impl ValidateService {
19    /// Create from a closure predicate and an expression source string (for error messages).
20    pub fn new(
21        predicate: impl Fn(&Exchange) -> bool + Send + Sync + 'static,
22        expression_source: impl Into<String>,
23    ) -> Self {
24        Self {
25            predicate: FilterPredicate::new(predicate),
26            expression_source: expression_source.into(),
27        }
28    }
29
30    /// Create from a pre-boxed `FilterPredicate` (used by `resolve_steps`).
31    pub fn from_predicate(
32        predicate: FilterPredicate,
33        expression_source: impl Into<String>,
34    ) -> Self {
35        Self {
36            predicate,
37            expression_source: expression_source.into(),
38        }
39    }
40}
41
42impl Service<Exchange> for ValidateService {
43    type Response = Exchange;
44    type Error = CamelError;
45    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
46
47    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
48        Poll::Ready(Ok(()))
49    }
50
51    fn call(&mut self, exchange: Exchange) -> Self::Future {
52        if (self.predicate)(&exchange) {
53            Box::pin(async move { Ok(exchange) })
54        } else {
55            let source = self.expression_source.clone();
56            Box::pin(async move {
57                Err(CamelError::ValidationError(format!(
58                    "validate('{source}'): predicate returned false",
59                )))
60            })
61        }
62    }
63}
64
65#[cfg(test)]
66mod tests {
67    use super::*;
68    use camel_api::Message;
69
70    // ── ValidateService tests ──
71
72    // 1. Passing predicate returns Ok(exchange)
73    #[tokio::test]
74    async fn test_validate_passing_predicate_returns_ok() {
75        let mut svc = ValidateService::new(|_ex: &Exchange| true, "true predicate");
76        let ex = Exchange::new(Message::new("hello"));
77        let result = svc.call(ex).await;
78        assert!(result.is_ok());
79        assert_eq!(result.unwrap().input.body.as_text(), Some("hello"));
80    }
81
82    // 2. Failing predicate returns Err(ValidationError)
83    #[tokio::test]
84    async fn test_validate_failing_predicate_returns_err() {
85        let mut svc = ValidateService::new(|_ex: &Exchange| false, "false predicate");
86        let ex = Exchange::new(Message::new("hello"));
87        let result = svc.call(ex).await;
88        assert!(result.is_err());
89        match result.unwrap_err() {
90            CamelError::ValidationError(msg) => {
91                assert!(
92                    msg.contains("false predicate"),
93                    "error message should contain expression source, got: {msg}"
94                );
95            }
96            other => panic!("expected ValidationError, got: {other:?}"),
97        }
98    }
99
100    // 3. Predicate evaluates the exchange (body-based validation)
101    #[tokio::test]
102    async fn test_validate_predicate_evaluates_body() {
103        let mut svc = ValidateService::new(
104            |ex: &Exchange| ex.input.body.as_text().is_some_and(|s| s.len() > 3),
105            "body length > 3",
106        );
107        let short = Exchange::new(Message::new("ab"));
108        let long = Exchange::new(Message::new("abcdef"));
109
110        assert!(svc.call(short).await.is_err());
111        assert!(svc.call(long).await.is_ok());
112    }
113
114    // 4. ValidateService is Clone
115    #[tokio::test]
116    async fn test_validate_clone_is_independent() {
117        let svc = ValidateService::new(|_ex: &Exchange| true, "true predicate");
118        let mut cloned = svc.clone();
119        let ex = Exchange::new(Message::new("hi"));
120        let result = cloned.call(ex).await;
121        assert!(result.is_ok());
122    }
123
124    // 5. poll_ready is always Ready(Ok(()))
125    #[tokio::test]
126    async fn test_validate_poll_ready() {
127        let mut svc = ValidateService::new(|_ex: &Exchange| true, "true predicate");
128        let poll = svc.poll_ready(&mut Context::from_waker(futures::task::noop_waker_ref()));
129        assert!(poll.is_ready());
130        // unwrap the Poll<Result<...>>
131        match poll {
132            std::task::Poll::Ready(Ok(())) => {}
133            other => panic!("expected Ready(Ok(())), got: {other:?}"),
134        }
135    }
136}
camel_processor/validate.rs

camel_processor/
validate.rs
