Message Serializer/Deserializer abstraction (#25)

* Started working on message module

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* just to align with other sdks (#21)

Signed-off-by: Doug Davis <dug@us.ibm.com>

* add link to email (#22)

Signed-off-by: Doug Davis <dug@us.ibm.com>

* Implemented custom deserialization process

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Implemented custom serialization process

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* V0.3 implementation (#24)

* Added String variant to Data

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Started V0.3 work

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Reworked EventDeserializer trait

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Now event parsing works with v1 and changes

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Cargo fmt

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Reorganized test data

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Fixed serde for v03

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Implemented spec version conversion

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* cargo fmt

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* WIP implementing serialization/deserialization for messages

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Adjusted some wrong merge changes

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* cargo fmt

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

* Rename

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>

Co-authored-by: Doug Davis <dug@us.ibm.com>

Cargo.lock

@@ -52,6 +52,7 @@ dependencies = [
  "serde",
  "serde-value",
  "serde_json",
+ "snafu",
  "uuid",
 ]
 
@@ -66,6 +67,12 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "doc-comment"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fea41bba32d969b513997752735605054bc0dfa92b4c56bf1189f2e174be7a10"
+
 [[package]]
 name = "getrandom"
 version = "0.1.14"
@@ -282,6 +289,27 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "snafu"
+version = "0.6.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b1ec0ae2ed980f26e1ad62e717feb01df90731df56887b5391a2c79f9f6805be"
+dependencies = [
+ "doc-comment",
+ "snafu-derive",
+]
+
+[[package]]
+name = "snafu-derive"
+version = "0.6.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0ec32ba84a7a86aeb0bc32fd0c46d31b0285599f68ea72e87eff6127889d99e1"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "syn"
 version = "1.0.16"

Cargo.toml

@@ -19,6 +19,7 @@ delegate = "^0.4"
 uuid = { version = "^0.8", features = ["serde", "v4"] }
 hostname = "^0.1"
 base64 = "^0.12"
+snafu = "^0.6"
 
 [dev-dependencies]
 rstest = "0.6"

src/event/data.rs

@@ -31,6 +31,18 @@ impl Data {
     {
         Ok(base64::decode(&i)?.into())
     }
+
+    pub fn from_binary<I>(content_type: Option<&str>, i: I) -> Result<Self, serde_json::Error>
+    where
+        I: AsRef<[u8]>,
+    {
+        let is_json = is_json_content_type(content_type.unwrap_or("application/json"));
+        if is_json {
+            serde_json::from_slice::<serde_json::Value>(i.as_ref()).map(Data::Json)
+        } else {
+            Ok(Data::Binary(i.as_ref().to_vec()))
+        }
+    }
 }
 
 pub(crate) fn is_json_content_type(ct: &str) -> bool {

src/event/deserializer.rs

@@ -0,0 +1,109 @@
+use super::Data;
+use super::Event;
+use super::{Attributes, AttributesReader};
+use crate::event::SpecVersion;
+use crate::message::{
+    BinaryDeserializer, BinaryVisitor, DeserializationResult, MessageAttributeValue,
+    SerializationResult, StructuredDeserializer, StructuredVisitor,
+};
+use std::borrow::Borrow;
+use std::io::Read;
+
+impl StructuredDeserializer for Event {
+    fn deserialize_structured<V: StructuredVisitor>(
+        self,
+        visitor: &mut V,
+    ) -> DeserializationResult {
+        let vec: Vec<u8> = serde_json::to_vec(&self)?;
+        visitor.set_structured_event::<&[u8]>(vec.borrow())
+    }
+}
+
+impl BinaryDeserializer for Event {
+    fn deserialize_binary<V: BinaryVisitor>(self, visitor: &mut V) -> DeserializationResult {
+        visitor.set_spec_version(self.get_specversion())?;
+        self.attributes.deserialize_attributes(visitor)?;
+        for (k, v) in self.extensions.into_iter() {
+            visitor.set_extension(&k, v.into())?;
+        }
+        match self.data.as_ref() {
+            Some(Data::String(s)) => visitor.set_body(s.as_bytes()),
+            Some(Data::Binary(v)) => visitor.set_body::<&[u8]>(v),
+            Some(Data::Json(j)) => {
+                let vec: Vec<u8> = serde_json::to_vec(j)?;
+                visitor.set_body::<&[u8]>(vec.borrow())
+            }
+            None => Ok(()),
+        }
+    }
+}
+
+pub(crate) trait AttributesDeserializer {
+    fn deserialize_attributes<V: BinaryVisitor>(self, visitor: &mut V) -> DeserializationResult;
+}
+
+pub(crate) trait AttributesSerializer {
+    fn serialize_attribute(
+        &mut self,
+        name: &str,
+        value: MessageAttributeValue,
+    ) -> SerializationResult;
+}
+
+impl AttributesDeserializer for Attributes {
+    fn deserialize_attributes<V: BinaryVisitor>(self, visitor: &mut V) -> DeserializationResult {
+        match self {
+            Attributes::V03(v03) => v03.deserialize_attributes(visitor),
+            Attributes::V10(v10) => v10.deserialize_attributes(visitor),
+        }
+    }
+}
+
+impl AttributesSerializer for Attributes {
+    fn serialize_attribute(
+        &mut self,
+        name: &str,
+        value: MessageAttributeValue,
+    ) -> SerializationResult {
+        match self {
+            Attributes::V03(v03) => v03.serialize_attribute(name, value),
+            Attributes::V10(v10) => v10.serialize_attribute(name, value),
+        }
+    }
+}
+
+impl StructuredVisitor for Event {
+    fn set_structured_event<R: Read>(&mut self, reader: R) -> SerializationResult {
+        let new_event: Event = serde_json::from_reader(reader)?;
+        self.attributes = new_event.attributes;
+        self.data = new_event.data;
+        self.extensions = new_event.extensions;
+        Ok(())
+    }
+}
+
+impl BinaryVisitor for Event {
+    fn set_spec_version(&mut self, spec_version: SpecVersion) -> SerializationResult {
+        match spec_version {
+            SpecVersion::V03 => self.attributes = self.attributes.clone().into_v03(),
+            SpecVersion::V10 => self.attributes = self.attributes.clone().into_v10(),
+        }
+        Ok(())
+    }
+
+    fn set_attribute(&mut self, name: &str, value: MessageAttributeValue) -> SerializationResult {
+        self.attributes.serialize_attribute(name, value)
+    }
+
+    fn set_extension(&mut self, name: &str, value: MessageAttributeValue) -> SerializationResult {
+        self.extensions.insert(name.to_string(), value.into());
+        Ok(())
+    }
+
+    fn set_body<R: Read>(&mut self, mut reader: R) -> SerializationResult {
+        let mut v = Vec::new();
+        let _ = reader.read_to_end(&mut v)?;
+        self.data = Some(Data::from_binary(self.get_datacontenttype(), v)?);
+        Ok(())
+    }
+}

src/event/mod.rs

@@ -5,6 +5,7 @@ mod event;
 mod extensions;
 #[macro_use]
 mod serde;
+mod deserializer;
 mod spec_version;
 
 pub use attributes::Attributes;

src/event/serde.rs

@@ -8,8 +8,6 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_value::Value;
 use std::collections::{BTreeMap, HashMap};
 
-const SPEC_VERSIONS: [&'static str; 2] = ["0.3", "1.0"];
-
 macro_rules! parse_optional_field {
     ($map:ident, $name:literal, $value_variant:ident, $error:ty) => {
         $map.remove($name)
@@ -135,7 +133,10 @@ impl<'de> Deserialize<'de> for Event {
         match parse_field!(map, "specversion", String, <D as Deserializer<'de>>::Error)?.as_str() {
             "0.3" => EventDeserializerV03::deserialize_event(map),
             "1.0" => EventDeserializerV10::deserialize_event(map),
-            s => Err(D::Error::unknown_variant(s, &SPEC_VERSIONS)),
+            s => Err(D::Error::unknown_variant(
+                s,
+                &super::spec_version::SPEC_VERSIONS,
+            )),
         }
     }
 }

src/event/spec_version.rs

@@ -1,6 +1,8 @@
 use std::convert::TryFrom;
 use std::fmt;
 
+pub(crate) const SPEC_VERSIONS: [&'static str; 2] = ["0.3", "1.0"];
+
 #[derive(PartialEq, Debug, Clone)]
 pub enum SpecVersion {
     V03,

src/event/v03/message.rs

@@ -0,0 +1,58 @@
+use crate::message::{
+    BinaryVisitor, DeserializationResult, Error, MessageAttributeValue, SerializationResult,
+};
+use std::convert::TryInto;
+
+impl crate::event::deserializer::AttributesDeserializer for super::Attributes {
+    fn deserialize_attributes<V: BinaryVisitor>(self, visitor: &mut V) -> DeserializationResult {
+        visitor.set_attribute("id", MessageAttributeValue::String(self.id))?;
+        visitor.set_attribute("type", MessageAttributeValue::String(self.ty))?;
+        visitor.set_attribute("source", MessageAttributeValue::UriRef(self.source))?;
+        if self.datacontenttype.is_some() {
+            visitor.set_attribute(
+                "datacontenttype",
+                MessageAttributeValue::String(self.datacontenttype.unwrap()),
+            )?;
+        }
+        if self.schemaurl.is_some() {
+            visitor.set_attribute(
+                "schemaurl",
+                MessageAttributeValue::Uri(self.schemaurl.unwrap()),
+            )?;
+        }
+        if self.subject.is_some() {
+            visitor.set_attribute(
+                "subject",
+                MessageAttributeValue::String(self.subject.unwrap()),
+            )?;
+        }
+        if self.time.is_some() {
+            visitor.set_attribute("time", MessageAttributeValue::DateTime(self.time.unwrap()))?;
+        }
+        Ok(())
+    }
+}
+
+impl crate::event::deserializer::AttributesSerializer for super::Attributes {
+    fn serialize_attribute(
+        &mut self,
+        name: &str,
+        value: MessageAttributeValue,
+    ) -> SerializationResult {
+        match name {
+            "id" => self.id = value.to_string(),
+            "type" => self.ty = value.to_string(),
+            "source" => self.source = value.to_string(),
+            "datacontenttype" => self.datacontenttype = Some(value.to_string()),
+            "schemaurl" => self.schemaurl = Some(value.to_string()),
+            "subject" => self.subject = Some(value.to_string()),
+            "time" => self.time = Some(value.try_into()?),
+            _ => {
+                return Err(Error::UnrecognizedAttributeName {
+                    name: name.to_string(),
+                })
+            }
+        };
+        Ok(())
+    }
+}

src/event/v03/mod.rs

@@ -1,5 +1,6 @@
 mod attributes;
 mod builder;
+mod message;
 mod serde;
 
 pub(crate) use crate::event::v03::serde::EventDeserializer;

src/event/v10/message.rs

@@ -0,0 +1,58 @@
+use crate::message::{
+    BinaryVisitor, DeserializationResult, Error, MessageAttributeValue, SerializationResult,
+};
+use std::convert::TryInto;
+
+impl crate::event::deserializer::AttributesDeserializer for super::Attributes {
+    fn deserialize_attributes<V: BinaryVisitor>(self, visitor: &mut V) -> DeserializationResult {
+        visitor.set_attribute("id", MessageAttributeValue::String(self.id))?;
+        visitor.set_attribute("type", MessageAttributeValue::String(self.ty))?;
+        visitor.set_attribute("source", MessageAttributeValue::UriRef(self.source))?;
+        if self.datacontenttype.is_some() {
+            visitor.set_attribute(
+                "datacontenttype",
+                MessageAttributeValue::String(self.datacontenttype.unwrap()),
+            )?;
+        }
+        if self.dataschema.is_some() {
+            visitor.set_attribute(
+                "dataschema",
+                MessageAttributeValue::Uri(self.dataschema.unwrap()),
+            )?;
+        }
+        if self.subject.is_some() {
+            visitor.set_attribute(
+                "subject",
+                MessageAttributeValue::String(self.subject.unwrap()),
+            )?;
+        }
+        if self.time.is_some() {
+            visitor.set_attribute("time", MessageAttributeValue::DateTime(self.time.unwrap()))?;
+        }
+        Ok(())
+    }
+}
+
+impl crate::event::deserializer::AttributesSerializer for super::Attributes {
+    fn serialize_attribute(
+        &mut self,
+        name: &str,
+        value: MessageAttributeValue,
+    ) -> SerializationResult {
+        match name {
+            "id" => self.id = value.to_string(),
+            "type" => self.ty = value.to_string(),
+            "source" => self.source = value.to_string(),
+            "datacontenttype" => self.datacontenttype = Some(value.to_string()),
+            "dataschema" => self.dataschema = Some(value.to_string()),
+            "subject" => self.subject = Some(value.to_string()),
+            "time" => self.time = Some(value.try_into()?),
+            _ => {
+                return Err(Error::UnrecognizedAttributeName {
+                    name: name.to_string(),
+                })
+            }
+        }
+        Ok(())
+    }
+}

src/event/v10/mod.rs

@@ -1,5 +1,6 @@
 mod attributes;
 mod builder;
+mod message;
 mod serde;
 
 pub(crate) use crate::event::v10::serde::EventDeserializer;

src/lib.rs

@@ -1,8 +1,10 @@
 extern crate serde;
 extern crate serde_json;
 extern crate serde_value;
+extern crate snafu;
 
 pub mod event;
+pub mod message;
 
 pub use event::Event;
 pub use event::EventBuilder;

src/message/deserializer.rs

@@ -0,0 +1,112 @@
+use super::{Encoding, MessageAttributeValue};
+use crate::event::SpecVersion;
+use crate::Event;
+use snafu::Snafu;
+use std::io::Read;
+
+#[derive(Debug, Snafu)]
+pub enum Error {
+    #[snafu(display("Unrecognized attribute name: {}", name))]
+    UnrecognizedAttributeName { name: String },
+    #[snafu(display("Error while decoding base64: {}", source))]
+    #[snafu(context(false))]
+    ParseTimeError { source: chrono::ParseError },
+    #[snafu(display("Error while decoding base64: {}", source))]
+    #[snafu(context(false))]
+    Base64DecodingError { source: base64::DecodeError },
+    #[snafu(display("Error while serializing/deserializing to json: {}", source))]
+    #[snafu(context(false))]
+    SerdeJsonError { source: serde_json::Error },
+    #[snafu(display("IO Error: {}", source))]
+    #[snafu(context(false))]
+    IOError { source: std::io::Error },
+}
+
+pub type SerializationResult = Result<(), Error>;
+pub type DeserializationResult = Result<(), Error>;
+
+pub trait StructuredDeserializer
+where
+    Self: Sized,
+{
+    fn deserialize_structured<V: StructuredVisitor>(self, visitor: &mut V)
+        -> DeserializationResult;
+
+    fn into_event(self) -> Result<Event, Error> {
+        let mut ev = Event::default();
+        self.deserialize_structured(&mut ev)?;
+        Ok(ev)
+    }
+}
+
+pub trait StructuredVisitor {
+    fn set_structured_event<R: Read>(&mut self, reader: R) -> SerializationResult;
+}
+
+pub trait BinaryDeserializer
+where
+    Self: Sized,
+{
+    fn deserialize_binary<V: BinaryVisitor>(self, visitor: &mut V) -> DeserializationResult;
+
+    fn into_event(self) -> Result<Event, Error> {
+        let mut ev = Event::default();
+        self.deserialize_binary(&mut ev)?;
+        Ok(ev)
+    }
+}
+
+pub trait BinaryVisitor {
+    fn set_spec_version(&mut self, spec_version: SpecVersion) -> SerializationResult;
+
+    fn set_attribute(&mut self, name: &str, value: MessageAttributeValue) -> SerializationResult;
+
+    fn set_extension(&mut self, name: &str, value: MessageAttributeValue) -> SerializationResult;
+
+    fn set_body<R: Read>(&mut self, reader: R) -> SerializationResult;
+}
+
+pub trait MessageDeserializer
+where
+    Self: StructuredDeserializer + BinaryDeserializer + Sized,
+{
+    fn encoding(&self) -> Encoding;
+
+    fn into_event(self) -> Result<Event, Error> {
+        let mut ev = Event::default();
+        self.deserialize_to(&mut ev)?;
+        Ok(ev)
+    }
+
+    fn deserialize_to_binary<T: BinaryVisitor>(self, visitor: &mut T) -> DeserializationResult {
+        if self.encoding() == Encoding::BINARY {
+            return self.deserialize_binary(visitor);
+        }
+
+        let ev = MessageDeserializer::into_event(self)?;
+        return ev.deserialize_binary(visitor);
+    }
+
+    fn deserialize_to_structured<T: StructuredVisitor>(
+        self,
+        visitor: &mut T,
+    ) -> DeserializationResult {
+        if self.encoding() == Encoding::STRUCTURED {
+            return self.deserialize_structured(visitor);
+        }
+
+        let ev = MessageDeserializer::into_event(self)?;
+        return ev.deserialize_structured(visitor);
+    }
+
+    fn deserialize_to<T: BinaryVisitor + StructuredVisitor>(
+        self,
+        visitor: &mut T,
+    ) -> DeserializationResult {
+        if self.encoding() == Encoding::STRUCTURED {
+            self.deserialize_structured(visitor)
+        } else {
+            self.deserialize_binary(visitor)
+        }
+    }
+}

src/message/encoding.rs

@@ -0,0 +1,5 @@
+#[derive(PartialEq)]
+pub enum Encoding {
+    STRUCTURED,
+    BINARY,
+}

src/message/mod.rs

@@ -0,0 +1,7 @@
+mod deserializer;
+mod encoding;
+mod types;
+
+pub use deserializer::*;
+pub use encoding::*;
+pub use types::MessageAttributeValue;

src/message/types.rs

@@ -0,0 +1,60 @@
+use crate::event::ExtensionValue;
+use chrono::{DateTime, Utc};
+use std::convert::TryInto;
+
+pub enum MessageAttributeValue {
+    Boolean(bool),
+    Integer(i64),
+    String(String),
+    Binary(Vec<u8>),
+    Uri(String),
+    UriRef(String),
+    DateTime(DateTime<Utc>),
+}
+
+impl TryInto<DateTime<Utc>> for MessageAttributeValue {
+    type Error = super::Error;
+
+    fn try_into(self) -> Result<DateTime<Utc>, Self::Error> {
+        match self {
+            MessageAttributeValue::DateTime(d) => Ok(d),
+            v => Ok(DateTime::<Utc>::from(DateTime::parse_from_rfc3339(
+                v.to_string().as_ref(),
+            )?)),
+        }
+    }
+}
+
+impl ToString for MessageAttributeValue {
+    fn to_string(&self) -> String {
+        match self {
+            MessageAttributeValue::Boolean(b) => b.to_string(),
+            MessageAttributeValue::Integer(i) => i.to_string(),
+            MessageAttributeValue::String(s) => s.clone(),
+            MessageAttributeValue::Binary(v) => base64::encode(v),
+            MessageAttributeValue::Uri(s) => s.clone(),
+            MessageAttributeValue::UriRef(s) => s.clone(),
+            MessageAttributeValue::DateTime(d) => d.to_rfc3339(),
+        }
+    }
+}
+
+impl Into<MessageAttributeValue> for ExtensionValue {
+    fn into(self) -> MessageAttributeValue {
+        match self {
+            ExtensionValue::String(s) => MessageAttributeValue::String(s),
+            ExtensionValue::Boolean(b) => MessageAttributeValue::Boolean(b),
+            ExtensionValue::Integer(i) => MessageAttributeValue::Integer(i),
+        }
+    }
+}
+
+impl Into<ExtensionValue> for MessageAttributeValue {
+    fn into(self) -> ExtensionValue {
+        match self {
+            MessageAttributeValue::Integer(i) => ExtensionValue::Integer(i),
+            MessageAttributeValue::Boolean(b) => ExtensionValue::Boolean(b),
+            v => ExtensionValue::String(v.to_string()),
+        }
+    }
+}

tests/message.rs

@@ -0,0 +1,71 @@
+mod test_data;
+use cloudevents::message::{
+    BinaryDeserializer, BinaryVisitor, DeserializationResult, Error, MessageAttributeValue,
+    MessageDeserializer, SerializationResult, StructuredDeserializer,
+};
+use cloudevents::Event;
+use test_data::*;
+
+#[test]
+fn message_v03_roundtrip_structured() -> DeserializationResult {
+    assert_eq!(
+        v03::full_json_data(),
+        StructuredDeserializer::into_event(v03::full_json_data())?
+    );
+    Ok(())
+}
+
+#[test]
+fn message_v03_roundtrip_binary() -> DeserializationResult {
+    assert_eq!(
+        v03::full_json_data(),
+        BinaryDeserializer::into_event(v03::full_json_data())?
+    );
+    Ok(())
+}
+
+#[test]
+fn message_v10_roundtrip_structured() -> DeserializationResult {
+    assert_eq!(
+        v10::full_json_data(),
+        StructuredDeserializer::into_event(v10::full_json_data())?
+    );
+    Ok(())
+}
+
+#[test]
+fn message_v10_roundtrip_binary() -> DeserializationResult {
+    assert_eq!(
+        v10::full_json_data(),
+        BinaryDeserializer::into_event(v10::full_json_data())?
+    );
+    Ok(())
+}
+
+#[test]
+fn message_v03_invalid_attribute_name() {
+    assert_eq!(
+        Error::UnrecognizedAttributeName {
+            name: "dataschema".to_string()
+        }
+        .to_string(),
+        v03::full_json_data()
+            .set_attribute("dataschema", MessageAttributeValue::Boolean(true))
+            .unwrap_err()
+            .to_string()
+    )
+}
+
+#[test]
+fn message_v10_invalid_attribute_name() {
+    assert_eq!(
+        Error::UnrecognizedAttributeName {
+            name: "schemaurl".to_string()
+        }
+        .to_string(),
+        v10::full_json_data()
+            .set_attribute("schemaurl", MessageAttributeValue::Boolean(true))
+            .unwrap_err()
+            .to_string()
+    )
+}