impl From<AmqpMessage> for AmqpCloudEvent

Signed-off-by: minghuaw <[email protected]>

Author: minghuaw

src/binding/fe2o3_amqp/constants.rs

@@ -1,4 +1,3 @@
-
 // Required
 pub(super) const ID: &str = "id";
 pub(super) const SOURCE: &str = "source";
@@ -22,4 +21,4 @@ pub(super) mod prefixed {
     pub const DATASCHEMA: &str = "cloudEvents:dataschema";
     pub const SUBJECT: &str = "cloudEvents:subject";
     pub const TIME: &str = "cloudEvents:time";
-}
+}

src/binding/fe2o3_amqp/deserializer.rs

@@ -27,6 +27,8 @@ impl BinaryDeserializer for AmqpCloudEvent {
         let spec_version = {
             let value = self
                 .application_properties
+                .as_mut()
+                .ok_or(Error::WrongEncoding {})?
                 .remove(prefixed::SPECVERSION)
                 .ok_or(Error::WrongEncoding {})
                 .map(|val| match val {
@@ -47,14 +49,16 @@ impl BinaryDeserializer for AmqpCloudEvent {
         // remaining attributes
         let attributes = spec_version.attribute_names();
 
-        for (key, value) in self.application_properties.0.into_iter() {
-            if let Some(key) = key.strip_prefix(ATTRIBUTE_PREFIX) {
-                if attributes.contains(&key) {
-                    let value = MessageAttributeValue::try_from((key, value))?;
-                    serializer = serializer.set_attribute(key, value)?;
-                } else {
-                    let value = MessageAttributeValue::try_from(value)?;
-                    serializer = serializer.set_extension(key, value)?;
+        if let Some(application_properties) = self.application_properties {
+            for (key, value) in application_properties.0.into_iter() {
+                if let Some(key) = key.strip_prefix(ATTRIBUTE_PREFIX) {
+                    if attributes.contains(&key) {
+                        let value = MessageAttributeValue::try_from((key, value))?;
+                        serializer = serializer.set_attribute(key, value)?;
+                    } else {
+                        let value = MessageAttributeValue::try_from(value)?;
+                        serializer = serializer.set_extension(key, value)?;
+                    }
                 }
             }
         }
@@ -79,8 +83,7 @@ impl StructuredDeserializer for AmqpCloudEvent {
         let bytes = match self.body {
             Body::Data(data) => data.0.into_vec(),
             Body::Nothing => vec![],
-            Body::Sequence(_) 
-            | Body::Value(_) => return Err(Error::WrongEncoding {  }),
+            Body::Sequence(_) | Body::Value(_) => return Err(Error::WrongEncoding {}),
         };
         serializer.set_structured_event(bytes)
     }

src/binding/fe2o3_amqp/mod.rs

@@ -3,12 +3,12 @@
 use std::collections::HashMap;
 use std::convert::TryFrom;
 
-use chrono::{Utc, TimeZone};
+use chrono::{TimeZone, Utc};
 use fe2o3_amqp_lib::types::messaging::{ApplicationProperties, Body, Message, Properties};
 use fe2o3_amqp_lib::types::primitives::{Binary, SimpleValue, Symbol, Timestamp, Value};
 
 use crate::event::{AttributeValue, ExtensionValue};
-use crate::message::{Error, MessageAttributeValue};
+use crate::message::{BinaryDeserializer, Error, MessageAttributeValue};
 use crate::Event;
 
 use self::constants::{
@@ -39,13 +39,21 @@ pub type Extensions = HashMap<String, ExtensionValue>;
 /// structured mode, otherwise it defaults to binary mode.
 pub struct AmqpCloudEvent {
     content_type: Option<Symbol>,
-    application_properties: ApplicationProperties,
+    application_properties: Option<ApplicationProperties>,
     body: AmqpBody,
 }
 
 impl AmqpCloudEvent {
-    pub fn from_event(event: Event) -> Result<Self, Error> {
-        todo!()
+    fn new() -> Self {
+        Self {
+            content_type: None,
+            application_properties: None,
+            body: Body::Nothing,
+        }
+    }
+
+    pub fn from_binary_event(event: Event) -> Result<Self, Error> {
+        BinaryDeserializer::deserialize_binary(event, Self::new())
     }
 }
 
@@ -58,13 +66,24 @@ impl From<AmqpCloudEvent> for AmqpMessage {
             delivery_annotations: None,
             message_annotations: None,
             properties: Some(properties),
-            application_properties: Some(event.application_properties),
+            application_properties: event.application_properties,
             body: event.body,
             footer: None,
         }
     }
 }
 
+impl From<AmqpMessage> for AmqpCloudEvent {
+    fn from(message: AmqpMessage) -> Self {
+        let content_type = message.properties.map(|p| p.content_type).flatten();
+        Self {
+            content_type,
+            application_properties: message.application_properties,
+            body: message.body,
+        }
+    }
+}
+
 impl<'a> From<AttributeValue<'a>> for SimpleValue {
     fn from(value: AttributeValue) -> Self {
         match value {
@@ -149,10 +168,10 @@ impl TryFrom<SimpleValue> for MessageAttributeValue {
             SimpleValue::Timestamp(val) => {
                 let datetime = Utc.timestamp_millis(val.into_inner());
                 Ok(MessageAttributeValue::DateTime(datetime))
-            },
+            }
             SimpleValue::Binary(val) => Ok(MessageAttributeValue::Binary(val.into_vec())),
             SimpleValue::String(val) => Ok(MessageAttributeValue::String(val)),
-            _ => Err(Error::WrongEncoding {  })
+            _ => Err(Error::WrongEncoding {}),
         }
     }
 }
@@ -163,13 +182,13 @@ impl<'a> TryFrom<(&'a str, SimpleValue)> for MessageAttributeValue {
     fn try_from((key, value): (&'a str, SimpleValue)) -> Result<Self, Self::Error> {
         match key {
             // String
-            ID | prefixed::ID 
+            ID | prefixed::ID
             // String
-            | SPECVERSION | prefixed::SPECVERSION 
+            | SPECVERSION | prefixed::SPECVERSION
             // String
             | TYPE | prefixed::TYPE
             // String
-            | DATACONTENTTYPE 
+            | DATACONTENTTYPE
             // String
             | SUBJECT | prefixed::SUBJECT => {
                 let val = String::try_from(value).map_err(|_| Error::WrongEncoding {})?;

src/binding/fe2o3_amqp/serializer.rs

@@ -1,38 +1,49 @@
-use fe2o3_amqp_types::primitives::{SimpleValue, Symbol, Binary};
-use fe2o3_amqp_types::messaging::{Data as AmqpData};
+use fe2o3_amqp_types::messaging::{ApplicationProperties, Data as AmqpData};
+use fe2o3_amqp_types::primitives::{Binary, SimpleValue, Symbol};
 
 use crate::binding::header_prefix;
 use crate::message::StructuredSerializer;
-use crate::{message::{BinarySerializer, MessageAttributeValue, Error}, event::SpecVersion};
+use crate::{
+    event::SpecVersion,
+    message::{BinarySerializer, Error, MessageAttributeValue},
+};
 
 use super::constants::DATACONTENTTYPE;
-use super::{AmqpCloudEvent, ATTRIBUTE_PREFIX, AmqpBody};
+use super::{AmqpBody, AmqpCloudEvent, ATTRIBUTE_PREFIX};
 
 impl BinarySerializer<AmqpCloudEvent> for AmqpCloudEvent {
     fn set_spec_version(mut self, spec_version: SpecVersion) -> crate::message::Result<Self> {
         let key = String::from("cloudEvents:specversion");
         let value = String::from(spec_version.as_str());
-        self.application_properties.insert(key, SimpleValue::from(value));
+        self.application_properties
+            .get_or_insert(ApplicationProperties::default())
+            .insert(key, SimpleValue::from(value));
         Ok(self)
     }
 
-    fn set_attribute(mut self, name: &str, value: MessageAttributeValue) -> crate::message::Result<Self> {
+    fn set_attribute(
+        mut self,
+        name: &str,
+        value: MessageAttributeValue,
+    ) -> crate::message::Result<Self> {
         // For the binary mode, the AMQP content-type property field value maps directly to the
         // CloudEvents datacontenttype attribute.
-        // 
+        //
         // All CloudEvents attributes with exception of datacontenttype MUST be individually mapped
         // to and from the AMQP application-properties section.
         if name == DATACONTENTTYPE {
             self.content_type = match value {
                 MessageAttributeValue::String(s) => Some(Symbol::from(s)),
-                _ => return Err(Error::WrongEncoding {  })
+                _ => return Err(Error::WrongEncoding {}),
             }
         } else {
             // CloudEvent attributes are prefixed with "cloudEvents:" for use in the
             // application-properties section
             let key = header_prefix(ATTRIBUTE_PREFIX, name);
             let value = SimpleValue::from(value);
-            self.application_properties.insert(key, value);
+            self.application_properties
+                .get_or_insert(ApplicationProperties::default())
+                .insert(key, value);
         }
 
         Ok(self)
@@ -44,10 +55,16 @@ impl BinarySerializer<AmqpCloudEvent> for AmqpCloudEvent {
     // systems that also process the message. Extension specifications that do this SHOULD specify
     // how receivers are to interpret messages if the copied values differ from the cloud-event
     // serialized values.
-    fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> crate::message::Result<Self> {
+    fn set_extension(
+        mut self,
+        name: &str,
+        value: MessageAttributeValue,
+    ) -> crate::message::Result<Self> {
         let key = header_prefix(ATTRIBUTE_PREFIX, name);
         let value = SimpleValue::from(value);
-        self.application_properties.insert(key, value);
+        self.application_properties
+            .get_or_insert(ApplicationProperties::default())
+            .insert(key, value);
         Ok(self)
     }
 

src/binding/mod.rs

@@ -4,6 +4,8 @@
 pub mod actix;
 #[cfg(feature = "axum")]
 pub mod axum;
+#[cfg(feature = "fe2o3-amqp")]
+pub mod fe2o3_amqp;
 #[cfg(any(
     feature = "http-binding",
     feature = "actix",
@@ -23,8 +25,6 @@ pub mod rdkafka;
 pub mod reqwest;
 #[cfg(feature = "warp")]
 pub mod warp;
-#[cfg(feature = "fe2o3-amqp")]
-pub mod fe2o3_amqp;
 
 #[cfg(feature = "rdkafka")]
 pub(crate) mod kafka {