cloudevents#9 Encoders for MQTT

Author: sbcd90

cloudevents-sdk-paho-mqtt/src/mqtt_consumer_record.rs

@@ -101,8 +101,8 @@ impl MessageExt for Message {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::mqtt_producer_record::MessageRecord;
 
+    use crate::headers::MqttVersion::{MQTT_3, MQTT_5};
     use crate::MessageBuilderExt;
     use chrono::Utc;
     use cloudevents::event::Data;
@@ -127,27 +127,23 @@ mod tests {
             .build()
             .unwrap();
 
-        let message_record = MessageRecord::from_event(
-            EventBuilderV10::new()
-                .id("0001")
-                .ty("example.test")
-                .time(time)
-                .source("http://localhost")
-                .extension("someint", "10")
-                .data("application/json", json!({"hello": "world"}))
-                .build()
-                .unwrap(),
-            headers::MqttVersion::V5,
-        )
-        .unwrap();
+        let event = EventBuilderV10::new()
+            .id("0001")
+            .ty("example.test")
+            .time(time)
+            .source("http://localhost")
+            .extension("someint", "10")
+            .data("application/json", json!({"hello": "world"}))
+            .build()
+            .unwrap();
 
         let msg = MessageBuilder::new()
             .topic("test")
-            .message_record(&message_record)
+            .event(event, MQTT_5)
             .qos(1)
             .finalize();
 
-        assert_eq!(msg.to_event(headers::MqttVersion::V5).unwrap(), expected)
+        assert_eq!(msg.to_event().unwrap(), expected)
     }
 
     #[test]
@@ -172,18 +168,12 @@ mod tests {
             .build()
             .unwrap();
 
-        let serialized_event =
-            StructuredDeserializer::deserialize_structured(input, MessageRecord::new()).unwrap();
-
         let msg = MessageBuilder::new()
             .topic("test")
-            .message_record(&serialized_event)
+            .event(input, MQTT_3)
             .qos(1)
             .finalize();
 
-        assert_eq!(
-            msg.to_event(headers::MqttVersion::V3_1_1).unwrap(),
-            expected
-        )
+        assert_eq!(msg.to_event().unwrap(), expected)
     }
 }

cloudevents-sdk-paho-mqtt/src/mqtt_producer_record.rs

@@ -1,4 +1,5 @@
 use super::headers;
+use crate::headers::MqttVersion::MQTT_5;
 use cloudevents::event::SpecVersion;
 use cloudevents::message::{
     BinaryDeserializer, BinarySerializer, Error, MessageAttributeValue, Result,
@@ -22,7 +23,7 @@ impl MessageRecord {
         }
     }
 
-    pub fn from_event(event: Event, version: headers::MqttVersion) -> Result<Self> {
+    pub fn from_event(event: Event, version: &headers::MqttVersion) -> Result<Self> {
         match version {
             headers::MqttVersion::MQTT_5 => {
                 BinaryDeserializer::deserialize_binary(event, MessageRecord::new())
@@ -130,9 +131,14 @@ pub trait MessageBuilderExt {
 impl MessageBuilderExt for MessageBuilder {
     fn event(mut self, event: Event, version: headers::MqttVersion) -> MessageBuilder {
         let message_record =
-            MessageRecord::from_event(event, version).expect("error while serializing the event");
+            MessageRecord::from_event(event, &version).expect("error while serializing the event");
 
-        self = self.properties(message_record.headers.clone());
+        match version {
+            MQTT_5 => {
+                self = self.properties(message_record.headers.clone());
+            }
+            _ => (),
+        }
 
         if let Some(s) = message_record.payload.as_ref() {
             self = self.payload(s.to_vec());