mq_bridge/
canonical_message.rs

1//  mq-bridge
2//  © Copyright 2025, by Marco Mengelkoch
3//  Licensed under MIT License, see License file for more details
4//  git clone https://github.com/marcomq/mq-bridge
5
6use bytes::Bytes;
7use serde::de::DeserializeOwned;
8use serde::{Deserialize, Serialize};
9use std::collections::HashMap;
10use uuid::Uuid;
11
12use crate::type_handler::KIND_KEY;
13
14#[derive(Debug, Serialize, Deserialize, Clone)]
15pub struct CanonicalMessage {
16    #[serde(serialize_with = "print_uuidv7", deserialize_with = "deserialize_u128")]
17    pub message_id: u128,
18    pub payload: Bytes,
19    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
20    pub metadata: HashMap<String, String>,
21}
22
23pub fn print_uuidv7<S>(value: &u128, serializer: S) -> Result<S::Ok, S::Error>
24where
25    S: serde::Serializer,
26{
27    serializer.serialize_str(fast_uuid_v7::format_uuid(*value).as_ref())
28}
29
30/// Custom deserializer for u128 that handles UUID strings, hex, and numeric formats.
31pub fn deserialize_u128<'de, D>(deserializer: D) -> Result<u128, D::Error>
32where
33    D: serde::Deserializer<'de>,
34{
35    let val = serde_json::Value::deserialize(deserializer)?;
36    u128_from_json(&val).map_err(serde::de::Error::custom)
37}
38
39pub(crate) fn u128_from_json(val: &serde_json::Value) -> Result<u128, String> {
40    if let Some(s) = val.as_str() {
41        if let Ok(uuid) = Uuid::parse_str(s) {
42            return Ok(uuid.as_u128());
43        } else if s.starts_with("0x") || s.starts_with("0X") {
44            if let Ok(n) =
45                u128::from_str_radix(s.trim_start_matches("0x").trim_start_matches("0X"), 16)
46            {
47                return Ok(n);
48            }
49        } else if let Ok(n) = s.parse::<u128>() {
50            return Ok(n);
51        }
52    } else if let Some(n) = val.as_u64() {
53        return Ok(n as u128);
54    } else if let Some(n) = val.as_i64() {
55        if n < 0 {
56            return Err("message_id cannot be negative".to_string());
57        }
58        return Ok(n as u128);
59    } else if val.is_number() {
60        // Fallback for large numeric literals that don't fit in u64/i64
61        if let Ok(n) = serde_json::from_value::<u128>(val.clone()) {
62            return Ok(n);
63        }
64    } else if let Some(oid) = val.get("$oid").and_then(|v| v.as_str()) {
65        if let Ok(n) = u128::from_str_radix(oid, 16) {
66            return Ok(n);
67        }
68    }
69    Err("Invalid u128 format".to_string())
70}
71
72impl CanonicalMessage {
73    pub fn new(payload: Vec<u8>, message_id: Option<u128>) -> Self {
74        Self {
75            message_id: message_id.unwrap_or_else(fast_uuid_v7::gen_id_with_sub_ms_4),
76            payload: Bytes::from(payload),
77            metadata: HashMap::new(),
78        }
79    }
80
81    pub fn new_bytes(payload: Bytes, message_id: Option<u128>) -> Self {
82        Self {
83            message_id: message_id.unwrap_or_else(fast_uuid_v7::gen_id_with_sub_ms_4),
84            payload,
85            metadata: HashMap::new(),
86        }
87    }
88
89    pub fn from_type<T: Serialize>(data: &T) -> Result<Self, serde_json::Error> {
90        let bytes = serde_json::to_vec(data)?;
91        Ok(Self::new(bytes, None))
92    }
93
94    pub fn from_vec(payload: impl Into<Vec<u8>>) -> Self {
95        Self::new(payload.into(), None)
96    }
97
98    pub fn set_id(&mut self, id: u128) {
99        self.message_id = id;
100    }
101
102    pub fn from_json(payload: serde_json::Value) -> Result<Self, serde_json::Error> {
103        #[derive(Deserialize)]
104        struct IdExtractor {
105            #[serde(deserialize_with = "deserialize_u128")]
106            id: u128,
107        }
108
109        let mut message_id = None;
110        for key in ["message_id", "id", "_id"] {
111            if let Some(v) = payload.get(key) {
112                // Use from_value with a helper struct to leverage deserialize_u128
113                // and produce a proper serde_json::Error on failure.
114                let mut map = serde_json::Map::new();
115                map.insert("id".to_string(), v.clone());
116                let extractor: IdExtractor =
117                    serde_json::from_value(serde_json::Value::Object(map))?;
118                message_id = Some(extractor.id);
119                break;
120            }
121        }
122
123        let bytes = serde_json::to_vec(&payload)?;
124        Ok(Self::new(bytes, message_id))
125    }
126
127    pub fn parse<T: DeserializeOwned>(&self) -> Result<T, serde_json::Error> {
128        serde_json::from_slice(&self.payload)
129    }
130
131    /// Returns the payload as a UTF-8 lossy string.
132    pub fn get_payload_str(&self) -> std::borrow::Cow<'_, str> {
133        String::from_utf8_lossy(&self.payload)
134    }
135
136    /// Sets the payload of this message to the given string.
137    pub fn set_payload_str(&mut self, payload: impl Into<String>) {
138        self.payload = Bytes::from(payload.into());
139    }
140
141    pub fn with_metadata(mut self, metadata: HashMap<String, String>) -> Self {
142        self.metadata = metadata;
143        self
144    }
145
146    pub fn with_metadata_kv(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
147        self.metadata.insert(key.into(), value.into());
148        self
149    }
150
151    pub fn with_type_key(mut self, kind: impl Into<String>) -> Self {
152        self.metadata.insert(KIND_KEY.into(), kind.into());
153        self
154    }
155
156    pub fn with_raw_format(mut self) -> Self {
157        self.metadata
158            .insert("mq_bridge.original_format".to_string(), "raw".to_string());
159        self
160    }
161}
162
163impl From<&str> for CanonicalMessage {
164    fn from(s: &str) -> Self {
165        Self::new(s.as_bytes().into(), None)
166    }
167}
168
169impl From<String> for CanonicalMessage {
170    fn from(s: String) -> Self {
171        Self::new(s.into_bytes(), None)
172    }
173}
174
175impl From<Vec<u8>> for CanonicalMessage {
176    fn from(v: Vec<u8>) -> Self {
177        Self::new(v, None)
178    }
179}
180
181impl From<serde_json::Value> for CanonicalMessage {
182    fn from(v: serde_json::Value) -> Self {
183        Self::from_json(v).expect("Failed to serialize JSON value")
184    }
185}
186
187/// A context object that holds metadata and identification for a message,
188/// separated from the payload. Useful for typed handlers.
189#[derive(Debug, Clone)]
190pub struct MessageContext {
191    pub message_id: u128,
192    pub metadata: HashMap<String, String>,
193}
194
195impl From<CanonicalMessage> for MessageContext {
196    fn from(msg: CanonicalMessage) -> Self {
197        Self {
198            message_id: msg.message_id,
199            metadata: msg.metadata,
200        }
201    }
202}
203
204#[doc(hidden)]
205pub mod tracing_support {
206    use super::CanonicalMessage;
207
208    /// A helper struct to lazily format a slice of message IDs for tracing.
209    /// The collection and formatting only occurs if the trace is enabled.
210    pub struct LazyMessageIds<'a>(pub &'a [CanonicalMessage]);
211
212    impl<'a> std::fmt::Debug for LazyMessageIds<'a> {
213        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
214            let ids: Vec<String> = self
215                .0
216                .iter()
217                .map(|m| format!("{:032x}", m.message_id))
218                .collect();
219            f.debug_list().entries(ids).finish()
220        }
221    }
222}
223
224#[doc(hidden)]
225pub mod macro_support {
226    use super::CanonicalMessage;
227    use serde::Serialize;
228
229    pub trait Fallback {
230        fn convert(&self) -> CanonicalMessage;
231    }
232
233    impl<T: Serialize> Fallback for Wrap<T> {
234        fn convert(&self) -> CanonicalMessage {
235            CanonicalMessage::from_type(&self.0).expect("Serialization failed in msg! macro")
236        }
237    }
238
239    pub struct Wrap<T>(pub T);
240
241    impl<T> Wrap<T>
242    where
243        T: Into<CanonicalMessage> + Clone,
244    {
245        pub fn convert(&self) -> CanonicalMessage {
246            self.0.clone().into()
247        }
248    }
249}
250
251/// A macro to create a `CanonicalMessage` easily.
252///
253/// Examples:
254/// ```rust
255/// use mq_bridge::msg;
256///
257/// let m1 = msg!("hello");
258/// let m2 = msg!("hello", "greeting");
259/// let m3 = msg!("hello", "kind" => "greeting");
260///
261/// #[derive(serde::Serialize, Clone)]
262/// struct MyData { val: i32 }
263/// let m4 = msg!(MyData { val: 42 }, "my_type");
264/// ```
265#[macro_export]
266macro_rules! msg {
267    ($payload:expr $(, $key:expr => $val:expr)* $(,)?) => {
268        {
269            #[allow(unused_imports)]
270            use $crate::canonical_message::macro_support::{Wrap, Fallback};
271            #[allow(unused_mut)]
272            let mut message = Wrap($payload).convert();
273            $(
274                message = message.with_metadata_kv($key, $val);
275            )*
276            message
277        }
278    };
279    ($payload:expr, $kind:expr $(,)?) => {
280        {
281            #[allow(unused_imports)]
282            use $crate::canonical_message::macro_support::{Wrap, Fallback};
283            let mut message = Wrap($payload).convert();
284            message = message.with_type_key($kind);
285            message
286        }
287    };
288}
289
290#[cfg(test)]
291mod tests {
292    use super::*;
293    use serde_json::json;
294
295    #[test]
296    fn test_message_id_parsing() {
297        // String UUID
298        let uuid = "550e8400-e29b-41d4-a716-446655440000";
299        let msg = CanonicalMessage::from_json(json!({ "id": uuid })).unwrap();
300        assert_eq!(msg.message_id, 113059749145936325402354257176981405696);
301
302        // Hex string
303        let msg = CanonicalMessage::from_json(json!({ "id": "0xFF" })).unwrap();
304        assert_eq!(msg.message_id, 255);
305
306        // Numeric
307        let msg = CanonicalMessage::from_json(json!({ "id": 100 })).unwrap();
308        assert_eq!(msg.message_id, 100);
309
310        // Negative numeric
311        let msg_err = CanonicalMessage::from_json(json!({ "id": -1 }));
312        assert!(msg_err.is_err());
313
314        // Mongo OID
315        let oid = "507f1f77bcf86cd799439011";
316        let msg = CanonicalMessage::from_json(json!({ "_id": { "$oid": oid } })).unwrap();
317        let expected = u128::from_str_radix(oid, 16).unwrap();
318        assert_eq!(msg.message_id, expected);
319    }
320
321    #[test]
322    fn test_metadata_builder() {
323        let msg = CanonicalMessage::new(b"payload".to_vec(), None)
324            .with_metadata_kv("key1", "val1")
325            .with_type_key("my_type");
326
327        assert_eq!(msg.metadata.get("key1").map(|s| s.as_str()), Some("val1"));
328        assert_eq!(
329            msg.metadata.get("kind").map(|s| s.as_str()),
330            Some("my_type")
331        );
332    }
333}
mq_bridge/canonical_message.rs

mq_bridge/
canonical_message.rs
