9 Encoders for MQTT

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

Author: sbcd90

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

@@ -115,7 +115,8 @@ impl MessageDeserializer for ConsumerMessageDeserializer {
                 .flatten()
                 .map(|s| s.starts_with(headers::CLOUDEVENTS_JSON_HEADER))
                 .unwrap_or(false),
-            self.headers.get(headers::MQTT_VERSION_HEADER)
+            self.headers
+                .get(headers::MQTT_VERSION_HEADER)
                 .map(|s| String::from_utf8(s.to_vec()).ok())
                 .flatten()
                 .map(|s| s.eq(headers::MQTT_V5_BINARY))
@@ -130,7 +131,7 @@ impl MessageDeserializer for ConsumerMessageDeserializer {
 
 pub fn record_to_event(msg: &Message) -> Result<Event> {
     MessageDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?)
-/*    match version {
+    /*    match version {
         headers::MqttVersion::V5 => {
             BinaryDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?)
         }
@@ -190,7 +191,8 @@ mod tests {
                 .extension("mqttversion", headers::MQTT_V5_BINARY)
                 .data(
                     "application/octet-stream",
-                    Data::Binary(String::from("hello rust").into_bytes()))
+                    Data::Binary(String::from("hello rust").into_bytes()),
+                )
                 .build()
                 .unwrap(),
         )
@@ -234,9 +236,6 @@ mod tests {
             .qos(1)
             .finalize();
 
-        assert_eq!(
-            msg.to_event().unwrap(),
-            expected
-        )
+        assert_eq!(msg.to_event().unwrap(), expected)
     }
 }

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

@@ -23,11 +23,13 @@ impl MessageRecord {
     }
 
     pub fn from_event(event: Event) -> Result<Self> {
-        match event.extension(headers::MQTT_VERSION_HEADER)
+        match event
+            .extension(headers::MQTT_VERSION_HEADER)
             .map(|e| e.to_string().eq(headers::MQTT_V5_BINARY))
-            .unwrap_or(false) {
+            .unwrap_or(false)
+        {
             true => BinaryDeserializer::deserialize_binary(event, MessageRecord::new()),
-            _ => StructuredDeserializer::deserialize_structured(event, MessageRecord::new())
+            _ => StructuredDeserializer::deserialize_structured(event, MessageRecord::new()),
         }
     }
 }