initial impl of deserializers

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

Author: minghuaw

Cargo.toml

@@ -54,7 +54,7 @@ http-body = { version = "^0.4", optional = true }
 poem-lib = { version = "=1.2.34", optional = true, package = "poem" }
 nats-lib = { version = "0.21.0", optional = true, package = "nats" }
 fe2o3-amqp-lib = { version = "0.4.0", optional = true, package = "fe2o3-amqp" }
-fe2o3-amqp-types = { version = "0.3.3", optional = true }
+fe2o3-amqp-types = { version = "0.3.4", optional = true }
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies]
 hostname = "^0.3"

src/binding/fe2o3_amqp/constants.rs

@@ -0,0 +1,25 @@
+
+// Required
+pub(super) const ID: &str = "id";
+pub(super) const SOURCE: &str = "source";
+pub(super) const SPECVERSION: &str = "specversion";
+pub(super) const TYPE: &str = "type";
+
+// Optional
+pub(super) const DATACONTENTTYPE: &str = "datacontenttype";
+pub(super) const DATASCHEMA: &str = "dataschema";
+pub(super) const SUBJECT: &str = "subject";
+pub(super) const TIME: &str = "time";
+
+pub(super) mod prefixed {
+    // Required
+    pub const ID: &str = "cloudEvents:id";
+    pub const SOURCE: &str = "cloudEvents:source";
+    pub const SPECVERSION: &str = "cloudEvents:specversion";
+    pub const TYPE: &str = "cloudEvents:type";
+
+    // Optional
+    pub const DATASCHEMA: &str = "cloudEvents:dataschema";
+    pub const SUBJECT: &str = "cloudEvents:subject";
+    pub const TIME: &str = "cloudEvents:time";
+}

src/binding/fe2o3_amqp/deserializer.rs

@@ -0,0 +1,101 @@
+use std::convert::TryFrom;
+
+use fe2o3_amqp_types::primitives::{SimpleValue, Symbol};
+
+use crate::{
+    binding::CLOUDEVENTS_JSON_HEADER,
+    event::SpecVersion,
+    message::{
+        BinaryDeserializer, BinarySerializer, Encoding, Error, MessageAttributeValue,
+        MessageDeserializer, Result, StructuredDeserializer, StructuredSerializer,
+    },
+};
+
+use super::{
+    constants::{self, prefixed, DATACONTENTTYPE},
+    AmqpCloudEvent, ATTRIBUTE_PREFIX,
+};
+
+impl BinaryDeserializer for AmqpCloudEvent {
+    fn deserialize_binary<R: Sized, V: BinarySerializer<R>>(
+        mut self,
+        mut serializer: V,
+    ) -> Result<R> {
+        use fe2o3_amqp_types::messaging::Body;
+
+        // specversion
+        let spec_version = {
+            let value = self
+                .application_properties
+                .remove(prefixed::SPECVERSION)
+                .ok_or(Error::WrongEncoding {})
+                .map(|val| match val {
+                    SimpleValue::String(s) => Ok(s),
+                    _ => Err(Error::WrongEncoding {}),
+                })??;
+            SpecVersion::try_from(&value[..])?
+        };
+        serializer = serializer.set_spec_version(spec_version.clone())?;
+
+        // datacontenttype
+        serializer = match self.content_type {
+            Some(Symbol(content_type)) => serializer
+                .set_attribute(DATACONTENTTYPE, MessageAttributeValue::String(content_type))?,
+            None => serializer,
+        };
+
+        // 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)?;
+                }
+            }
+        }
+
+        match self.body {
+            Body::Data(data) => {
+                let bytes = data.0.into_vec();
+                serializer.end_with_data(bytes)
+            }
+            Body::Nothing => serializer.end(),
+            Body::Sequence(_) | Body::Value(_) => Err(Error::WrongEncoding {}),
+        }
+    }
+}
+
+impl StructuredDeserializer for AmqpCloudEvent {
+    fn deserialize_structured<R: Sized, V: StructuredSerializer<R>>(
+        self,
+        serializer: V,
+    ) -> Result<R> {
+        use fe2o3_amqp_types::messaging::Body;
+        let bytes = match self.body {
+            Body::Data(data) => data.0.into_vec(),
+            Body::Nothing => vec![],
+            Body::Sequence(_) 
+            | Body::Value(_) => return Err(Error::WrongEncoding {  }),
+        };
+        serializer.set_structured_event(bytes)
+    }
+}
+
+impl MessageDeserializer for AmqpCloudEvent {
+    fn encoding(&self) -> Encoding {
+        match self
+            .content_type
+            .as_ref()
+            .map(|s| s.starts_with(CLOUDEVENTS_JSON_HEADER))
+        {
+            Some(true) => Encoding::STRUCTURED,
+            Some(false) => Encoding::BINARY,
+            None => Encoding::UNKNOWN,
+        }
+    }
+}