camel_processor/
log.rs

1use camel_api::{CamelError, Exchange, IdentityProcessor};
2use std::future::Future;
3use std::pin::Pin;
4use std::task::{Context, Poll};
5use tower::Service;
6use tracing::{debug, info, trace, warn};
7
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub enum LogLevel {
10    Trace,
11    Debug,
12    Info,
13    Warn,
14    Error,
15}
16
17#[derive(Clone)]
18pub struct LogProcessor {
19    inner: IdentityProcessor,
20    level: LogLevel,
21    message: String,
22}
23
24// TODO(PROC-004): Add metrics instrumentation — processed count, error count, latency histograms
25// are not yet instrumented on processors. Consider wiring MetricsCollector into LogProcessor and
26// incrementing a counter on each call, recording elapsed time, and tracking errors.
27
28impl LogProcessor {
29    pub fn new(level: LogLevel, message: String) -> Self {
30        Self {
31            inner: IdentityProcessor,
32            level,
33            message,
34        }
35    }
36}
37
38impl Service<Exchange> for LogProcessor {
39    type Response = Exchange;
40    type Error = CamelError;
41    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
42
43    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
44        self.inner.poll_ready(cx)
45    }
46
47    fn call(&mut self, exchange: Exchange) -> Self::Future {
48        let msg = self.message.clone();
49        let exchange_id = exchange.correlation_id.clone();
50        let body_preview = sanitize_preview(
51            &exchange
52                .input
53                .body
54                .as_text()
55                .unwrap_or("")
56                .chars()
57                .take(64)
58                .collect::<String>(),
59        );
60        debug!(exchange_id = %exchange_id, body_preview = %body_preview, "LogProcessor processing exchange");
61        match self.level {
62            LogLevel::Trace => trace!(exchange_id = %exchange_id, "{}", msg),
63            LogLevel::Debug => debug!(exchange_id = %exchange_id, "{}", msg),
64            LogLevel::Info => info!(exchange_id = %exchange_id, "{}", msg),
65            LogLevel::Warn => warn!(exchange_id = %exchange_id, "{}", msg),
66            // log-policy: handler-owned
67            LogLevel::Error => warn!(exchange_id = %exchange_id, "{}", msg),
68        }
69        self.inner.call(exchange)
70    }
71}
72
73/// A log processor that evaluates a message expression against the Exchange at call-time.
74/// Analogous to [`DynamicSetHeader`](crate::dynamic_set_header::DynamicSetHeader).
75#[derive(Clone)]
76pub struct DynamicLog<F> {
77    inner: IdentityProcessor,
78    level: LogLevel,
79    expr: F,
80}
81
82impl<F> DynamicLog<F>
83where
84    F: Fn(&Exchange) -> String + Clone + Send + Sync + 'static,
85{
86    pub fn new(level: LogLevel, expr: F) -> Self {
87        Self {
88            inner: IdentityProcessor,
89            level,
90            expr,
91        }
92    }
93}
94
95impl<F> Service<Exchange> for DynamicLog<F>
96where
97    F: Fn(&Exchange) -> String + Clone + Send + Sync + 'static,
98{
99    type Response = Exchange;
100    type Error = CamelError;
101    type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;
102
103    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
104        self.inner.poll_ready(cx)
105    }
106
107    fn call(&mut self, exchange: Exchange) -> Self::Future {
108        let exchange_id = exchange.correlation_id.clone();
109        let msg = (self.expr)(&exchange);
110        match self.level {
111            LogLevel::Trace => trace!(exchange_id = %exchange_id, "{}", msg),
112            LogLevel::Debug => debug!(exchange_id = %exchange_id, "{}", msg),
113            LogLevel::Info => info!(exchange_id = %exchange_id, "{}", msg),
114            LogLevel::Warn => warn!(exchange_id = %exchange_id, "{}", msg),
115            // log-policy: handler-owned
116            LogLevel::Error => warn!(exchange_id = %exchange_id, "{}", msg),
117        }
118        self.inner.call(exchange)
119    }
120}
121
122/// Strip control characters that could enable log injection.
123/// Replaces control chars with U+FFFD (replacement char) to preserve
124/// alignment while making injection visible.
125fn sanitize_preview(s: &str) -> String {
126    s.chars()
127        .map(|c| if c.is_control() { '\u{FFFD}' } else { c })
128        .collect()
129}
130
131#[cfg(test)]
132mod tests {
133    use super::*;
134    use camel_api::body::Body;
135    use camel_api::{Message, Value};
136    use tower::ServiceExt;
137
138    #[tokio::test]
139    async fn test_log_processor_passes_exchange_through() {
140        let mut processor = LogProcessor::new(LogLevel::Info, "test message".into());
141        let exchange = Exchange::default();
142        let result = processor.call(exchange).await;
143        assert!(result.is_ok());
144    }
145
146    #[tokio::test]
147    async fn test_log_processor_preserves_exchange_body() {
148        let mut processor = LogProcessor::new(LogLevel::Debug, "debug message".into());
149        let mut exchange = Exchange::default();
150        exchange.input.body = Body::Text("test body".into());
151        let result = processor.call(exchange).await.unwrap();
152        assert_eq!(result.input.body.as_text(), Some("test body"));
153    }
154
155    #[tokio::test]
156    async fn test_dynamic_log_evaluates_body() {
157        let svc = DynamicLog::new(LogLevel::Info, |ex: &Exchange| {
158            format!("body={}", ex.input.body.as_text().unwrap_or(""))
159        });
160        let exchange = Exchange::new(Message::new("hello"));
161        let result = svc.oneshot(exchange).await.unwrap();
162        // Exchange passes through unchanged
163        assert_eq!(result.input.body.as_text(), Some("hello"));
164    }
165
166    #[tokio::test]
167    async fn test_dynamic_log_evaluates_header() {
168        let svc = DynamicLog::new(LogLevel::Info, |ex: &Exchange| {
169            let counter = ex
170                .input
171                .header("CamelTimerCounter")
172                .and_then(|v| v.as_i64())
173                .unwrap_or(0);
174            format!("{} World", counter)
175        });
176        let mut msg = Message::new("");
177        msg.set_header("CamelTimerCounter", Value::Number(42.into()));
178        let exchange = Exchange::new(msg);
179        let result = svc.oneshot(exchange).await.unwrap();
180        // Exchange passes through unchanged
181        assert_eq!(
182            result.input.header("CamelTimerCounter"),
183            Some(&Value::Number(42.into()))
184        );
185    }
186
187    #[test]
188    fn body_preview_strips_control_chars() {
189        let body_with_injection = "no\ttab\nnl\rcr\0null\u{1B}esc";
190        let preview: String = body_with_injection.chars().take(64).collect();
191        let sanitized = sanitize_preview(&preview);
192        assert!(!sanitized.contains('\n'), "newlines must be replaced");
193        assert!(
194            !sanitized.contains('\r'),
195            "carriage returns must be replaced"
196        );
197        assert!(!sanitized.contains('\t'), "tabs must be replaced");
198        assert!(!sanitized.contains('\0'), "null must be replaced");
199        assert!(
200            !sanitized.contains('\u{1B}'),
201            "ESC (ANSI injection) must be replaced"
202        );
203        // U+FFFD replacement char must appear (replacement, not deletion)
204        assert!(
205            sanitized.contains('\u{FFFD}'),
206            "control chars must be replaced with U+FFFD, not deleted"
207        );
208    }
209}
camel_processor/log.rs

camel_processor/
log.rs
