camel_function/protocol/
mod.rs

1use base64::Engine;
2use serde::{Deserialize, Serialize};
3use std::collections::HashMap;
4
5// ---------------------------------------------------------------------------
6// RegisterRequest
7// ---------------------------------------------------------------------------
8
9#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
10pub struct RegisterRequest {
11    pub function_id: String,
12    pub runtime: String,
13    pub source: String,
14    pub timeout_ms: u64,
15}
16
17// ---------------------------------------------------------------------------
18// BodyWire
19// ---------------------------------------------------------------------------
20
21/// Wire representation of a message body.
22///
23/// v1 supports `Empty`, `Text`, `Json`. `Bytes` (base64) and `Xml` are
24/// forward-looking extensions for future protocol versions.
25#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
26#[serde(rename_all = "lowercase", tag = "kind", content = "value")]
27pub enum BodyWire {
28    Empty,
29    Text(String),
30    Json(serde_json::Value),
31    Bytes(String),
32    Xml(String),
33}
34
35impl BodyWire {
36    pub fn from_body(body: &camel_api::Body) -> Self {
37        match body {
38            camel_api::Body::Empty => BodyWire::Empty,
39            camel_api::Body::Text(s) => BodyWire::Text(s.clone()),
40            camel_api::Body::Json(v) => BodyWire::Json(v.clone()),
41            camel_api::Body::Bytes(b) => {
42                BodyWire::Bytes(base64::engine::general_purpose::STANDARD.encode(b))
43            }
44            camel_api::Body::Xml(s) => BodyWire::Xml(s.clone()),
45            camel_api::Body::Stream(_) => {
46                tracing::debug!("stream body cannot cross process boundary, mapping to Empty");
47                BodyWire::Empty
48            }
49            // Future body variants cannot cross the process boundary.
50            _ => BodyWire::Empty,
51        }
52    }
53
54    pub fn to_body(&self) -> camel_api::Body {
55        match self {
56            BodyWire::Empty => camel_api::Body::Empty,
57            BodyWire::Text(s) => camel_api::Body::Text(s.clone()),
58            BodyWire::Json(v) => camel_api::Body::Json(v.clone()),
59            BodyWire::Bytes(b64) => match base64::engine::general_purpose::STANDARD.decode(b64) {
60                Ok(bytes) => camel_api::Body::Bytes(bytes::Bytes::from(bytes)),
61                Err(e) => {
62                    tracing::warn!(error = %e, "invalid base64 in wire body, falling back to Empty");
63                    camel_api::Body::Empty
64                }
65            },
66            BodyWire::Xml(s) => camel_api::Body::Xml(s.clone()),
67        }
68    }
69
70    pub fn to_patch_body(self) -> Result<camel_api::function::PatchBody, camel_api::CamelError> {
71        use camel_api::function::PatchBody;
72        match self {
73            BodyWire::Empty => Ok(PatchBody::Empty),
74            BodyWire::Text(s) => Ok(PatchBody::Text(s)),
75            BodyWire::Json(v) => Ok(PatchBody::Json(v)),
76            BodyWire::Bytes(_) => Err(camel_api::CamelError::ProcessorError(
77                "unsupported body type for function: Bytes".into(),
78            )),
79            BodyWire::Xml(_) => Err(camel_api::CamelError::ProcessorError(
80                "unsupported body type for function: Xml".into(),
81            )),
82        }
83    }
84}
85
86// ---------------------------------------------------------------------------
87// ExchangeWire
88// ---------------------------------------------------------------------------
89
90#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
91pub struct ExchangeWire {
92    pub function_id: String,
93    pub correlation_id: String,
94    pub body: BodyWire,
95    pub headers: HashMap<String, serde_json::Value>,
96    pub properties: HashMap<String, serde_json::Value>,
97    pub timeout_ms: u64,
98}
99
100// ---------------------------------------------------------------------------
101// InvokeResponse
102// ---------------------------------------------------------------------------
103
104#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
105pub struct InvokeResponse {
106    pub ok: bool,
107    #[serde(skip_serializing_if = "Option::is_none")]
108    pub patch: Option<PatchWire>,
109    #[serde(skip_serializing_if = "Option::is_none")]
110    pub error: Option<ErrorWire>,
111}
112
113// ---------------------------------------------------------------------------
114// PatchWire
115// ---------------------------------------------------------------------------
116
117#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
118pub struct PatchWire {
119    pub body: Option<BodyWire>,
120    pub headers_set: Vec<(String, serde_json::Value)>,
121    pub headers_removed: Vec<String>,
122    pub properties_set: Vec<(String, serde_json::Value)>,
123}
124
125impl PatchWire {
126    pub fn to_exchange_patch(
127        self,
128    ) -> Result<camel_api::function::ExchangePatch, camel_api::CamelError> {
129        let body = self.body.map(BodyWire::to_patch_body).transpose()?;
130        Ok(camel_api::function::ExchangePatch {
131            body,
132            headers_set: self.headers_set,
133            headers_removed: self.headers_removed,
134            properties_set: self.properties_set,
135        })
136    }
137}
138
139// ---------------------------------------------------------------------------
140// ErrorWire
141// ---------------------------------------------------------------------------
142
143#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
144pub struct ErrorWire {
145    pub kind: String,
146    pub message: String,
147    pub stack: Option<String>,
148}
149
150// ---------------------------------------------------------------------------
151// HealthResponse
152// ---------------------------------------------------------------------------
153
154#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
155pub struct HealthResponse {
156    pub status: String,
157    pub registered: Vec<String>,
158}
159
160// ---------------------------------------------------------------------------
161// ErrorResponse
162// ---------------------------------------------------------------------------
163
164#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
165pub struct ErrorResponse {
166    pub error: String,
167    pub kind: String,
168}
169
170pub mod client;
171
172pub use client::ProtocolClient;
173
174// ---------------------------------------------------------------------------
175// Tests
176// ---------------------------------------------------------------------------
177
178#[cfg(test)]
179mod tests {
180    use super::*;
181
182    #[test]
183    fn test_register_request_roundtrip() {
184        let req = RegisterRequest {
185            function_id: "fn-123".into(),
186            runtime: "deno".into(),
187            source: "export default function(ex) { return ex; }".into(),
188            timeout_ms: 5000,
189        };
190        let json = serde_json::to_string(&req).unwrap();
191        let decoded: RegisterRequest = serde_json::from_str(&json).unwrap();
192        assert_eq!(req, decoded);
193    }
194
195    fn make_exchange_wire(body: BodyWire) -> ExchangeWire {
196        let mut headers = HashMap::new();
197        headers.insert("content-type".into(), serde_json::json!("text/plain"));
198        let mut properties = HashMap::new();
199        properties.insert("retry-count".into(), serde_json::json!(3));
200        ExchangeWire {
201            function_id: "fn-abc".into(),
202            correlation_id: "corr-001".into(),
203            body,
204            headers,
205            properties,
206            timeout_ms: 3000,
207        }
208    }
209
210    #[test]
211    fn test_exchange_wire_roundtrip_text() {
212        let wire = make_exchange_wire(BodyWire::Text("hello world".into()));
213        let json = serde_json::to_string(&wire).unwrap();
214        let decoded: ExchangeWire = serde_json::from_str(&json).unwrap();
215        assert_eq!(wire, decoded);
216    }
217
218    #[test]
219    fn test_exchange_wire_roundtrip_json() {
220        let wire = make_exchange_wire(BodyWire::Json(serde_json::json!({"key": "value"})));
221        let json = serde_json::to_string(&wire).unwrap();
222        let decoded: ExchangeWire = serde_json::from_str(&json).unwrap();
223        assert_eq!(wire, decoded);
224    }
225
226    #[test]
227    fn test_exchange_wire_roundtrip_bytes() {
228        let original = b"binary data here";
229        let encoded = base64::engine::general_purpose::STANDARD.encode(original);
230        let wire = make_exchange_wire(BodyWire::Bytes(encoded));
231        let json = serde_json::to_string(&wire).unwrap();
232        let decoded: ExchangeWire = serde_json::from_str(&json).unwrap();
233        assert_eq!(wire, decoded);
234        // Verify base64 roundtrip
235        if let BodyWire::Bytes(b64) = &decoded.body {
236            let decoded_bytes = base64::engine::general_purpose::STANDARD
237                .decode(b64)
238                .unwrap();
239            assert_eq!(decoded_bytes, original);
240        } else {
241            panic!("expected Bytes variant");
242        }
243    }
244
245    #[test]
246    fn test_exchange_wire_roundtrip_xml() {
247        let wire = make_exchange_wire(BodyWire::Xml("<root><item>1</item></root>".into()));
248        let json = serde_json::to_string(&wire).unwrap();
249        let decoded: ExchangeWire = serde_json::from_str(&json).unwrap();
250        assert_eq!(wire, decoded);
251    }
252
253    #[test]
254    fn test_exchange_wire_roundtrip_empty() {
255        let wire = make_exchange_wire(BodyWire::Empty);
256        let json = serde_json::to_string(&wire).unwrap();
257        let decoded: ExchangeWire = serde_json::from_str(&json).unwrap();
258        assert_eq!(wire, decoded);
259    }
260
261    #[test]
262    fn test_invoke_response_ok() {
263        let resp = InvokeResponse {
264            ok: true,
265            patch: Some(PatchWire {
266                body: Some(BodyWire::Text("processed".into())),
267                headers_set: vec![("x-custom".into(), serde_json::json!("added"))],
268                headers_removed: vec!["x-old".into()],
269                properties_set: vec![("status".into(), serde_json::json!("done"))],
270            }),
271            error: None,
272        };
273        let json = serde_json::to_string(&resp).unwrap();
274        let decoded: InvokeResponse = serde_json::from_str(&json).unwrap();
275        assert_eq!(resp, decoded);
276        assert!(decoded.ok);
277        assert!(decoded.patch.as_ref().unwrap().body.is_some());
278    }
279
280    #[test]
281    fn test_invoke_response_error() {
282        let resp = InvokeResponse {
283            ok: false,
284            patch: None,
285            error: Some(ErrorWire {
286                kind: "user_error".into(),
287                message: "ReferenceError: x is not defined".into(),
288                stack: Some("at main (file:///fn.ts:3:1)".into()),
289            }),
290        };
291        let json = serde_json::to_string(&resp).unwrap();
292        let decoded: InvokeResponse = serde_json::from_str(&json).unwrap();
293        assert_eq!(resp, decoded);
294        assert!(!decoded.ok);
295        let err = decoded.error.unwrap();
296        assert_eq!(err.kind, "user_error");
297        assert!(err.stack.is_some());
298    }
299
300    #[test]
301    fn test_health_response() {
302        let resp = HealthResponse {
303            status: "ok".into(),
304            registered: vec!["fn-a".into(), "fn-b".into()],
305        };
306        let json = serde_json::to_string(&resp).unwrap();
307        let decoded: HealthResponse = serde_json::from_str(&json).unwrap();
308        assert_eq!(resp, decoded);
309        assert_eq!(decoded.registered.len(), 2);
310    }
311
312    #[test]
313    fn test_error_response() {
314        let resp = ErrorResponse {
315            error: "function not found".into(),
316            kind: "not_registered".into(),
317        };
318        let json = serde_json::to_string(&resp).unwrap();
319        let decoded: ErrorResponse = serde_json::from_str(&json).unwrap();
320        assert_eq!(resp, decoded);
321    }
322
323    #[test]
324    fn test_patch_wire() {
325        let patch = PatchWire {
326            body: Some(BodyWire::Json(serde_json::json!({"updated": true}))),
327            headers_set: vec![("x-new".into(), serde_json::json!("val"))],
328            headers_removed: vec!["x-old".into()],
329            properties_set: vec![("key".into(), serde_json::json!(42))],
330        };
331        let json = serde_json::to_string(&patch).unwrap();
332        let decoded: PatchWire = serde_json::from_str(&json).unwrap();
333        assert_eq!(patch, decoded);
334    }
335
336    #[test]
337    fn test_body_wire_serde_lowercase() {
338        let wire = BodyWire::Text("hello".into());
339        let json = serde_json::to_string(&wire).unwrap();
340        assert!(
341            json.contains("\"text\""),
342            "expected lowercase variant name, got: {json}"
343        );
344        assert!(
345            !json.contains("\"Text\""),
346            "should not have UpperCamelCase variant"
347        );
348        let decoded: BodyWire = serde_json::from_str(&json).unwrap();
349        assert_eq!(wire, decoded);
350    }
351
352    #[test]
353    fn test_body_wire_bytes_base64_roundtrip() {
354        let original_bytes = vec![0xDE, 0xAD, 0xBE, 0xEF, 0xCA, 0xFE];
355        let encoded = base64::engine::general_purpose::STANDARD.encode(&original_bytes);
356        let wire = BodyWire::Bytes(encoded.clone());
357
358        let json = serde_json::to_string(&wire).unwrap();
359        let decoded: BodyWire = serde_json::from_str(&json).unwrap();
360
361        if let BodyWire::Bytes(b64) = &decoded {
362            let roundtrip = base64::engine::general_purpose::STANDARD
363                .decode(b64)
364                .unwrap();
365            assert_eq!(roundtrip, original_bytes);
366        } else {
367            panic!("expected Bytes variant after roundtrip");
368        }
369
370        // Also verify to_body conversion
371        let body = wire.to_body();
372        if let camel_api::Body::Bytes(b) = body {
373            assert_eq!(b.to_vec(), original_bytes);
374        } else {
375            panic!("expected Body::Bytes from to_body()");
376        }
377    }
378
379    #[test]
380    fn test_body_wire_from_body_roundtrip() {
381        let bodies = vec![
382            ("Empty", camel_api::Body::Empty),
383            ("Text", camel_api::Body::Text("hello world".into())),
384            (
385                "Json",
386                camel_api::Body::Json(serde_json::json!({"key": "value"})),
387            ),
388            (
389                "Xml",
390                camel_api::Body::Xml("<root><item>1</item></root>".into()),
391            ),
392        ];
393
394        for (name, body) in bodies {
395            let wire = BodyWire::from_body(&body);
396            let roundtripped = wire.to_body();
397            assert_eq!(body, roundtripped, "roundtrip failed for {name}");
398        }
399
400        // Bytes need special handling since from_body base64-encodes
401        let original_bytes = vec![0xDE, 0xAD, 0xBE, 0xEF];
402        let body = camel_api::Body::Bytes(bytes::Bytes::from(original_bytes.clone()));
403        let wire = BodyWire::from_body(&body);
404        let roundtripped = wire.to_body();
405        if let camel_api::Body::Bytes(b) = roundtripped {
406            assert_eq!(b.to_vec(), original_bytes);
407        } else {
408            panic!("expected Body::Bytes after Bytes roundtrip");
409        }
410    }
411
412    #[test]
413    fn test_body_wire_from_body_stream_maps_to_empty() {
414        use camel_api::{StreamBody, StreamMetadata};
415        use futures::stream;
416
417        let chunks = vec![Ok(bytes::Bytes::from("stream data"))];
418        let stream_body = camel_api::Body::Stream(StreamBody {
419            stream: std::sync::Arc::new(tokio::sync::Mutex::new(Some(Box::pin(stream::iter(
420                chunks,
421            ))))),
422            metadata: StreamMetadata::default(),
423        });
424
425        let wire = BodyWire::from_body(&stream_body);
426        assert!(matches!(wire, BodyWire::Empty));
427    }
428
429    #[test]
430    fn test_body_wire_to_body_from_body_text() {
431        let wire = BodyWire::Text("hello world".into());
432        let body = wire.to_body();
433        let wire2 = BodyWire::from_body(&body);
434
435        assert!(matches!(wire2, BodyWire::Text(ref s) if s == "hello world"));
436    }
437}
camel_function/protocol/mod.rs

camel_function/protocol/
mod.rs
