camel_processor/data_format/
mod.rs

1mod csv;
2// `pub(crate)` so the TAR.GZ stream splitter can reuse the bounded
3// single-member GZIP decoder without widening the public data-format API.
4pub(crate) mod gzip;
5mod json;
6mod tar;
7mod tar_gz;
8#[cfg(test)]
9mod test_util;
10mod xml;
11mod zip;
12
13pub use csv::{CAMEL_CSV_HEADER_RECORD, CsvConfig, CsvDataFormat, QuoteMode, RecordSeparator};
14pub use gzip::{GzipConfig, GzipDataFormat};
15pub use json::{JsonConfig, JsonDataFormat};
16pub use tar::{TarConfig, TarDataFormat};
17pub use tar_gz::{TarGzConfig, TarGzDataFormat};
18pub use xml::{XmlConfig, XmlDataFormat};
19pub use zip::{ZipConfig, ZipDataFormat};
20
21use camel_api::DataFormat;
22use camel_api::body::Body;
23use camel_api::error::CamelError;
24use std::sync::Arc;
25
26/// Config-aware factory. `config` is the raw `config:` block from the DSL
27/// (or `Null` for defaults). Each arm deserializes into its own typed config
28/// with `deny_unknown_fields`, so a stray key fails closed with a precise message.
29pub fn builtin_data_format_with_config(
30    name: &str,
31    config: &serde_json::Value,
32) -> Result<Option<Arc<dyn DataFormat>>, CamelError> {
33    let df: Arc<dyn DataFormat> = match name {
34        "json" => Arc::new(JsonDataFormat::new(parse_cfg::<JsonConfig>(name, config)?)),
35        "csv" => Arc::new(CsvDataFormat::new(parse_cfg::<CsvConfig>(name, config)?)),
36        "xml" => Arc::new(XmlDataFormat::new(parse_cfg::<XmlConfig>(name, config)?)),
37        "zip" => Arc::new(ZipDataFormat::new(parse_cfg::<ZipConfig>(name, config)?)),
38        "tar" => Arc::new(TarDataFormat::new(parse_cfg::<TarConfig>(name, config)?)),
39        "gzip" => Arc::new(GzipDataFormat::new(parse_cfg::<GzipConfig>(name, config)?)),
40        "tar.gz" => Arc::new(TarGzDataFormat::new(parse_cfg::<TarGzConfig>(
41            name, config,
42        )?)),
43        _ => return Ok(None),
44    };
45    Ok(Some(df))
46}
47
48/// Back-compat shim: existing callers keep working (config = Null → defaults).
49pub fn builtin_data_format(name: &str) -> Option<Arc<dyn DataFormat>> {
50    builtin_data_format_with_config(name, &serde_json::Value::Null)
51        .ok()
52        .flatten()
53}
54
55fn parse_cfg<T>(name: &str, v: &serde_json::Value) -> Result<T, CamelError>
56where
57    T: serde::de::DeserializeOwned + Default,
58{
59    if v.is_null() {
60        return Ok(T::default());
61    }
62    serde_json::from_value::<T>(v.clone()).map_err(|e| {
63        CamelError::RouteError(format!("invalid config for data format '{name}': {e}"))
64    })
65}
66
67/// Materializes the `marshal` input body into owned bytes, enforcing
68/// `max_input_size` (DoS cap, R3-L1) after materialization. `df` names the
69/// calling data format for error messages (e.g. `TarDataFormat`). `Text` and
70/// `Xml` use their UTF-8 bytes, `Json` is serialized, `Bytes` is copied;
71/// `Empty` and `Stream` fail closed, and streams are never consumed.
72fn materialize_marshal_input(
73    df: &str,
74    body: &Body,
75    max_input_size: u64,
76) -> Result<Vec<u8>, CamelError> {
77    let content: Vec<u8> = match body {
78        Body::Text(s) => s.as_bytes().to_vec(),
79        Body::Json(v) => serde_json::to_vec(v).map_err(|e| {
80            CamelError::TypeConversionFailed(format!("{df}::marshal cannot serialize JSON: {e}"))
81        })?,
82        Body::Bytes(b) => b.to_vec(),
83        Body::Xml(s) => s.as_bytes().to_vec(),
84        Body::Empty => {
85            return Err(CamelError::TypeConversionFailed(format!(
86                "{df}::marshal requires non-empty body"
87            )));
88        }
89        Body::Stream(_) => {
90            return Err(CamelError::TypeConversionFailed(
91                "cannot marshal Body::Stream — add 'stream_cache' or 'convert_body_to' before this step"
92                    .to_string(),
93            ));
94        }
95        _ => {
96            return Err(CamelError::TypeConversionFailed(format!(
97                "{df}::marshal does not support this body type"
98            )));
99        }
100    };
101
102    if content.len() as u64 > max_input_size {
103        return Err(CamelError::TypeConversionFailed(format!(
104            "{df}::marshal input {} bytes exceeds max_input_size {}",
105            content.len(),
106            max_input_size
107        )));
108    }
109    Ok(content)
110}
111
112/// Extracts the raw wire bytes for `unmarshal` from a materialized body.
113/// `df` names the calling data format and `wire` labels the expected payload
114/// shape (e.g. `TAR data`) for error messages. `Bytes` and `Text` yield their
115/// bytes; `Json` and `Xml` fail closed because they cannot be archive data;
116/// `Empty` fails; `Stream` fails without being consumed.
117fn raw_unmarshal_body(df: &str, wire: &str, body: &Body) -> Result<Vec<u8>, CamelError> {
118    match body {
119        Body::Bytes(b) => Ok(b.to_vec()),
120        Body::Text(s) => Ok(s.as_bytes().to_vec()),
121        Body::Empty => Err(CamelError::TypeConversionFailed(format!(
122            "{df}::unmarshal requires non-empty body"
123        ))),
124        Body::Stream(_) => Err(CamelError::TypeConversionFailed(
125            "cannot unmarshal Body::Stream — use UnmarshalService which auto-materializes"
126                .to_string(),
127        )),
128        Body::Json(_) | Body::Xml(_) => Err(CamelError::TypeConversionFailed(format!(
129            "{df}::unmarshal only supports Body::Bytes and Body::Text ({wire})"
130        ))),
131        _ => Err(CamelError::TypeConversionFailed(format!(
132            "{df}::unmarshal does not support this body type"
133        ))),
134    }
135}
136
137#[cfg(test)]
138mod tests {
139    use super::*;
140
141    #[test]
142    fn test_builtin_json() {
143        let df = builtin_data_format("json").unwrap();
144        assert_eq!(df.name(), "json");
145    }
146
147    #[test]
148    fn test_builtin_xml() {
149        let df = builtin_data_format("xml").unwrap();
150        assert_eq!(df.name(), "xml");
151    }
152
153    #[test]
154    fn test_builtin_csv() {
155        let csv_df = builtin_data_format("csv").unwrap();
156        assert_eq!(csv_df.name(), "csv");
157    }
158
159    #[test]
160    fn test_builtin_unknown_returns_none() {
161        assert!(builtin_data_format("protobuf").is_none());
162        assert!(builtin_data_format("").is_none());
163    }
164
165    #[test]
166    fn test_builtin_json_with_config() {
167        let config = serde_json::json!({"max_bytes": 67108864});
168        let df = builtin_data_format_with_config("json", &config)
169            .unwrap()
170            .unwrap();
171        assert_eq!(df.name(), "json");
172    }
173
174    #[test]
175    fn test_builtin_json_with_null_config_returns_default() {
176        let df = builtin_data_format_with_config("json", &serde_json::Value::Null)
177            .unwrap()
178            .unwrap();
179        assert_eq!(df.name(), "json");
180    }
181
182    #[test]
183    fn test_builtin_json_with_unknown_key_fails() {
184        let config = serde_json::json!({"max_byte": 100});
185        let result = builtin_data_format_with_config("json", &config);
186        match result {
187            Err(CamelError::RouteError(msg)) => {
188                assert!(msg.contains("invalid config"), "msg: {msg}");
189            }
190            Err(other) => panic!("expected RouteError, got: {other:?}"),
191            Ok(_) => panic!("typo should fail closed"),
192        }
193    }
194
195    #[test]
196    fn test_builtin_shim_still_works() {
197        let df = builtin_data_format("json").unwrap();
198        assert_eq!(df.name(), "json");
199    }
200
201    #[test]
202    fn builtin_archive_formats_resolve() {
203        for (name, expected) in [("tar", "tar"), ("gzip", "gzip"), ("tar.gz", "tar.gz")] {
204            let df = builtin_data_format_with_config(name, &serde_json::Value::Null)
205                .unwrap()
206                .unwrap_or_else(|| panic!("'{name}' should resolve to a built-in format"));
207            assert_eq!(df.name(), expected, "format name mismatch for '{name}'");
208        }
209    }
210
211    #[test]
212    fn builtin_archive_config_rejects_unknown_fields() {
213        for name in ["tar", "gzip", "tar.gz"] {
214            let config = serde_json::json!({"unknown_field": true});
215            match builtin_data_format_with_config(name, &config) {
216                Err(CamelError::RouteError(msg)) => {
217                    assert!(
218                        msg.contains("invalid config") && msg.contains(name),
219                        "unexpected message for '{name}': {msg}"
220                    );
221                }
222                Err(other) => panic!("expected RouteError for '{name}', got: {other:?}"),
223                Ok(_) => panic!("unknown key for '{name}' should fail closed"),
224            }
225        }
226    }
227
228    #[test]
229    fn archive_marshal_serializes_json_bodies() {
230        // The shared marshal materialization serializes `Body::Json` instead
231        // of rejecting it; every archive format must round trip it as bytes.
232        let json = serde_json::json!({"payload": "text"});
233        let expected = serde_json::to_vec(&json).unwrap();
234        for name in ["tar", "gzip", "tar.gz", "zip"] {
235            let df = builtin_data_format(name).unwrap();
236            let restored = df
237                .unmarshal(df.marshal(Body::Json(json.clone())).unwrap())
238                .unwrap();
239            match restored {
240                Body::Bytes(b) => assert_eq!(&b[..], &expected[..], "format '{name}'"),
241                other => panic!("format '{name}' must return bytes, got: {other:?}"),
242            }
243        }
244    }
245
246    #[test]
247    fn archive_unmarshal_rejects_structured_bodies_with_format_named() {
248        // The shared raw-body extraction keeps the calling format's name in
249        // the rejection message for `Json`/`Xml` unmarshal input.
250        for name in ["tar", "gzip", "tar.gz", "zip"] {
251            let df = builtin_data_format(name).unwrap();
252            let msg = format!(
253                "{}",
254                df.unmarshal(Body::Json(serde_json::json!({}))).unwrap_err()
255            );
256            assert!(
257                msg.contains("::unmarshal only supports Body::Bytes and Body::Text"),
258                "format '{name}' should name the supported body kinds: {msg}"
259            );
260        }
261    }
262}
camel_processor/data_format/mod.rs

camel_processor/data_format/
mod.rs
