camel_processor/
remove_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};
8
9/// A processor that removes a header from the exchange's input message.
10#[derive(Clone)]
11pub struct RemoveHeader<P> {
12    inner: P,
13    key: String,
14}
15
16impl<P> RemoveHeader<P> {
17    /// Create a new RemoveHeader processor that removes the given header.
18    pub fn new(inner: P, key: impl Into<String>) -> Self {
19        Self {
20            inner,
21            key: key.into(),
22        }
23    }
24}
25
26/// A Tower Layer that wraps an inner service with a [`RemoveHeader`].
27#[derive(Clone)]
28pub struct RemoveHeaderLayer {
29    key: String,
30}
31
32impl RemoveHeaderLayer {
33    pub fn new(key: impl Into<String>) -> Self {
34        Self { key: key.into() }
35    }
36}
37
38impl<S> tower::Layer<S> for RemoveHeaderLayer {
39    type Service = RemoveHeader<S>;
40
41    fn layer(&self, inner: S) -> Self::Service {
42        RemoveHeader::new(inner, self.key.clone())
43    }
44}
45
46impl<P> Service<Exchange> for RemoveHeader<P>
47where
48    P: Service<Exchange, Response = Exchange, Error = CamelError> + Clone + Send + 'static,
49    P::Future: Send,
50{
51    type Response = Exchange;
52    type Error = CamelError;
53    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
54
55    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
56        self.inner.poll_ready(cx)
57    }
58
59    fn call(&mut self, mut exchange: Exchange) -> Self::Future {
60        exchange.input.headers.remove(&self.key);
61        let fut = self.inner.call(exchange);
62        Box::pin(fut)
63    }
64}
65
66#[cfg(test)]
67mod tests {
68    use super::*;
69    use camel_api::{IdentityProcessor, Message, Value};
70    use tower::ServiceExt;
71
72    #[tokio::test]
73    async fn test_remove_header_deletes_existing() {
74        let mut exchange = Exchange::new(Message::default());
75        exchange
76            .input
77            .set_header("CamelHttpPath", Value::String("x".into()));
78
79        let processor = RemoveHeader::new(IdentityProcessor, "CamelHttpPath");
80
81        let result = processor.oneshot(exchange).await.unwrap();
82        assert!(!result.input.headers.contains_key("CamelHttpPath"));
83    }
84
85    #[tokio::test]
86    async fn test_remove_header_noop_on_missing() {
87        let mut exchange = Exchange::new(Message::default());
88        exchange.input.set_header("A", Value::from(1));
89        exchange.input.set_header("B", Value::from(2));
90
91        let processor = RemoveHeader::new(IdentityProcessor, "C");
92
93        let result = processor.oneshot(exchange).await.unwrap();
94        assert_eq!(result.input.header("A"), Some(&Value::from(1)));
95        assert_eq!(result.input.header("B"), Some(&Value::from(2)));
96        assert!(!result.input.headers.contains_key("C"));
97    }
98
99    #[tokio::test]
100    async fn test_remove_header_preserves_other_headers() {
101        let mut exchange = Exchange::new(Message::default());
102        exchange.input.set_header("X", Value::from(1));
103        exchange.input.set_header("Y", Value::from(2));
104        exchange.input.set_header("Z", Value::from(3));
105
106        let processor = RemoveHeader::new(IdentityProcessor, "Y");
107
108        let result = processor.oneshot(exchange).await.unwrap();
109        assert_eq!(result.input.header("X"), Some(&Value::from(1)));
110        assert_eq!(result.input.header("Z"), Some(&Value::from(3)));
111        assert!(!result.input.headers.contains_key("Y"));
112    }
113
114    #[tokio::test]
115    async fn test_remove_header_preserves_body() {
116        let exchange = Exchange::new(Message::new("hello"));
117
118        let processor = RemoveHeader::new(IdentityProcessor, "anything");
119
120        let result = processor.oneshot(exchange).await.unwrap();
121        assert_eq!(result.input.body.as_text(), Some("hello"));
122    }
123}
camel_processor/remove_header.rs

camel_processor/
remove_header.rs
