camel_processor/
json_schema_validate.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::task::{Context, Poll};
4
5use tower::Service;
6
7use camel_api::body::Body;
8use camel_api::{CamelError, Exchange};
9
10/// Tower Service that validates the exchange body against a compiled JSON
11/// Schema. Returns `CamelError::ValidationError` if the body does not match.
12/// Validation runs BEFORE the inner service is called.
13///
14/// Non-JSON bodies (Body::Text, Body::Bytes, Body::Xml, Body::Empty,
15/// Body::Stream) are passed through without validation — the validator only
16/// runs against Body::Json. Callers that want to validate raw text/bytes
17/// must first run an Unmarshal step that produces Body::Json.
18pub struct JsonSchemaValidateService<P> {
19    inner: P,
20    validator: jsonschema::Validator,
21}
22
23impl<P> std::fmt::Debug for JsonSchemaValidateService<P> {
24    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
25        f.debug_struct("JsonSchemaValidateService")
26            .finish_non_exhaustive()
27    }
28}
29
30impl<P> Clone for JsonSchemaValidateService<P>
31where
32    P: Clone,
33{
34    fn clone(&self) -> Self {
35        Self {
36            inner: self.inner.clone(),
37            validator: self.validator.clone(),
38        }
39    }
40}
41
42impl<P> JsonSchemaValidateService<P> {
43    /// Compile a JSON Schema and wrap the given inner service. Returns
44    /// `CamelError::RouteError` if the schema is not a valid JSON Schema.
45    pub fn new(inner: P, schema: &serde_json::Value) -> Result<Self, CamelError> {
46        let validator = jsonschema::validator_for(schema)
47            .map_err(|e| CamelError::RouteError(format!("invalid JSON schema: {e}")))?;
48        Ok(Self { inner, validator })
49    }
50}
51
52impl<P> Service<Exchange> for JsonSchemaValidateService<P>
53where
54    P: Service<Exchange, Response = Exchange, Error = CamelError> + Clone + Send + 'static,
55    P::Future: Send,
56{
57    type Response = Exchange;
58    type Error = CamelError;
59    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
60
61    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
62        self.inner.poll_ready(cx)
63    }
64
65    fn call(&mut self, exchange: Exchange) -> Self::Future {
66        // Only Body::Json is validated. Anything else passes through.
67        let result = match &exchange.input.body {
68            Body::Json(v) => {
69                let errors: Vec<String> = self
70                    .validator
71                    .iter_errors(v)
72                    .map(|e| format!("{e} at {}", e.instance_path()))
73                    .collect();
74                if errors.is_empty() {
75                    Ok(())
76                } else {
77                    Err(errors.join("; "))
78                }
79            }
80            _ => Ok(()),
81        };
82        match result {
83            Ok(()) => {
84                let fut = self.inner.call(exchange);
85                Box::pin(fut)
86            }
87            Err(msg) => Box::pin(async move {
88                Err(CamelError::ValidationError(format!(
89                    "JSON schema validation failed: {msg}"
90                )))
91            }),
92        }
93    }
94}
camel_processor/json_schema_validate.rs

camel_processor/
json_schema_validate.rs
