camel_processor/
content_negotiation.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::sync::Arc;
4use std::task::{Context, Poll};
5
6use tower::Service;
7
8use camel_api::{CamelError, Exchange, Value};
9
10/// Injected verdict closure for content negotiation.
11///
12/// Receives the resolved `Content-Type` and `Accept` header values
13/// (`None` when absent or not a JSON string) and returns `Ok(())` to pass,
14/// or a [`CamelError`] (e.g. `UnsupportedMediaType` / `NotAcceptable`) to
15/// reject the exchange.
16pub type ContentNegotiationCheck =
17    Arc<dyn Fn(Option<&str>, Option<&str>) -> Result<(), CamelError> + Send + Sync>;
18
19/// Header-only media negotiation gate (REST DSL v2 strict mode).
20///
21/// The verdict is injected at compile time as a [`ContentNegotiationCheck`]
22/// closure; the processor never touches the message body (no polling, no
23/// materialization) and passes the exchange through unchanged when the
24/// check accepts it.
25#[derive(Clone)]
26pub struct ContentNegotiationProcessor {
27    check: ContentNegotiationCheck,
28}
29
30impl ContentNegotiationProcessor {
31    /// Create a new gate with the given verdict closure.
32    pub fn new(check: ContentNegotiationCheck) -> Self {
33        Self { check }
34    }
35}
36
37impl Service<Exchange> for ContentNegotiationProcessor {
38    type Response = Exchange;
39    type Error = CamelError;
40    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
41
42    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
43        Poll::Ready(Ok(()))
44    }
45
46    fn call(&mut self, exchange: Exchange) -> Self::Future {
47        let check = Arc::clone(&self.check);
48        Box::pin(async move {
49            let content_type = exchange
50                .input
51                .header_ic("Content-Type")
52                .and_then(Value::as_str);
53            let accept = exchange.input.header_ic("Accept").and_then(Value::as_str);
54            check(content_type, accept)?;
55            Ok(exchange)
56        })
57    }
58}
59
60#[cfg(test)]
61mod tests {
62    use super::*;
63    use bytes::Bytes;
64    use camel_api::{Body, Message, StreamBody};
65    use futures::Stream;
66    use std::sync::atomic::{AtomicUsize, Ordering};
67    use tower::ServiceExt;
68
69    type SeenArgs = Vec<(Option<String>, Option<String>)>;
70
71    /// One-chunk stream that counts every `poll_next` on a shared counter.
72    struct PollCountingStream {
73        inner: futures::stream::Iter<std::vec::IntoIter<Result<Bytes, CamelError>>>,
74        polls: Arc<AtomicUsize>,
75    }
76
77    impl Stream for PollCountingStream {
78        type Item = Result<Bytes, CamelError>;
79
80        fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
81            self.polls.fetch_add(1, Ordering::SeqCst);
82            Pin::new(&mut self.inner).poll_next(cx)
83        }
84    }
85
86    fn counting_stream_body(polls: Arc<AtomicUsize>) -> Body {
87        let chunks: Vec<Result<Bytes, CamelError>> = vec![Ok(Bytes::from_static(b"payload"))];
88        let stream = PollCountingStream {
89            inner: futures::stream::iter(chunks),
90            polls,
91        };
92        Body::Stream(StreamBody {
93            stream: Arc::new(tokio::sync::Mutex::new(Some(Box::pin(stream)))),
94            metadata: Default::default(),
95        })
96    }
97
98    fn recording_check(seen: Arc<std::sync::Mutex<SeenArgs>>) -> ContentNegotiationCheck {
99        Arc::new(move |content_type: Option<&str>, accept: Option<&str>| {
100            seen.lock()
101                .unwrap()
102                .push((content_type.map(str::to_string), accept.map(str::to_string)));
103            Ok(())
104        })
105    }
106
107    #[tokio::test]
108    async fn gate_passes_exchange_through_untouched() {
109        let polls = Arc::new(AtomicUsize::new(0));
110        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
111
112        let mut msg = Message::default();
113        msg.set_header("Content-Type", "application/json");
114        msg.set_header("Accept", "application/json");
115        msg.body = counting_stream_body(Arc::clone(&polls));
116        let headers_before = msg.headers.clone();
117        let exchange = Exchange::new(msg);
118
119        let processor = ContentNegotiationProcessor::new(recording_check(Arc::clone(&seen)));
120
121        let result = processor.oneshot(exchange).await.unwrap();
122
123        assert_eq!(result.input.headers, headers_before);
124        assert!(matches!(result.input.body, Body::Stream { .. }));
125        assert_eq!(polls.load(Ordering::SeqCst), 0);
126        let seen_args = seen.lock().unwrap();
127        assert_eq!(seen_args.len(), 1);
128        assert_eq!(
129            seen_args[0],
130            (
131                Some("application/json".to_string()),
132                Some("application/json".to_string())
133            )
134        );
135    }
136
137    #[tokio::test]
138    async fn gate_propagates_check_error() {
139        let mut msg = Message::default();
140        msg.set_header("Content-Type", "text/plain");
141        let exchange = Exchange::new(msg);
142
143        let check: ContentNegotiationCheck = Arc::new(|_content_type, _accept| {
144            Err(CamelError::UnsupportedMediaType {
145                consumed: "text/plain".into(),
146                declared: "application/json".into(),
147            })
148        });
149        let processor = ContentNegotiationProcessor::new(check);
150
151        let result = processor.oneshot(exchange).await;
152
153        match result {
154            Err(CamelError::UnsupportedMediaType { consumed, declared }) => {
155                assert_eq!(consumed, "text/plain");
156                assert_eq!(declared, "application/json");
157            }
158            other => panic!("expected UnsupportedMediaType, got {other:?}"),
159        }
160    }
161
162    #[tokio::test]
163    async fn gate_header_lookup_resolves_casings() {
164        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
165
166        let mut msg = Message::default();
167        msg.set_header("content-type", "application/json");
168        msg.set_header("accept", "*/*");
169        let exchange = Exchange::new(msg);
170
171        let processor = ContentNegotiationProcessor::new(recording_check(Arc::clone(&seen)));
172
173        processor.oneshot(exchange).await.unwrap();
174
175        let seen_args = seen.lock().unwrap();
176        assert_eq!(seen_args.len(), 1);
177        assert_eq!(
178            seen_args[0],
179            (
180                Some("application/json".to_string()),
181                Some("*/*".to_string())
182            )
183        );
184    }
185
186    #[tokio::test]
187    async fn gate_absent_headers_yield_none() {
188        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
189
190        let exchange = Exchange::new(Message::default());
191
192        let processor = ContentNegotiationProcessor::new(recording_check(Arc::clone(&seen)));
193
194        let result = processor.oneshot(exchange).await.unwrap();
195
196        let seen_args = seen.lock().unwrap();
197        assert_eq!(seen_args[0], (None, None));
198        assert!(result.input.headers.is_empty());
199    }
200
201    #[tokio::test]
202    async fn gate_non_string_header_values_yield_none() {
203        let seen = Arc::new(std::sync::Mutex::new(Vec::new()));
204
205        let mut msg = Message::default();
206        msg.set_header("Content-Type", 1);
207        msg.set_header("Accept", true);
208        let exchange = Exchange::new(msg);
209
210        let processor = ContentNegotiationProcessor::new(recording_check(Arc::clone(&seen)));
211
212        let result = processor.oneshot(exchange).await.unwrap();
213
214        let seen_args = seen.lock().unwrap();
215        assert_eq!(seen_args[0], (None, None));
216        assert!(result.input.headers.contains_key("Content-Type"));
217    }
218}
camel_processor/content_negotiation.rs

camel_processor/
content_negotiation.rs
