impl conversion btwn AmqpMessage & AmqpCloudEvent

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

Author: minghuaw

Cargo.toml

@@ -25,6 +25,7 @@ warp = ["warp-lib", "bytes", "http", "hyper"]
 axum = ["bytes", "http", "hyper", "axum-lib", "http-body", "async-trait"]
 poem = ["bytes", "http", "poem-lib", "hyper", "async-trait"]
 nats = ["nats-lib"]
+fe2o3-amqp = ["fe2o3-amqp-lib"]
 
 [dependencies]
 serde = { version = "^1.0", features = ["derive"] }
@@ -52,6 +53,7 @@ axum-lib = { version = "^0.5", optional = true , package="axum"}
 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" }
 
 [target."cfg(not(target_arch = \"wasm32\"))".dependencies]
 hostname = "^0.3"

src/binding/fe2o3_amqp/mod.rs

@@ -0,0 +1,54 @@
+//! Implements AMQP 1.0 binding for CloudEvents
+
+use std::convert::TryFrom;
+
+use fe2o3_amqp_lib::types::messaging::{ApplicationProperties, Message, Body, Data as AmqpData, Properties};
+use fe2o3_amqp_lib::types::primitives::{Value, Binary};
+
+use crate::Event;
+use crate::message::Error;
+
+/// Type alias for an AMQP 1.0 message
+/// 
+/// The generic parameter can be anything that implements `Serialize` and `Deserialize` but is of
+/// no importance because all CloudEvents are using the `Body::Data` as the body section type. For 
+/// convenience, this type alias chose `Value` as the value of the generic parameter
+pub type AmqpMessage = Message<Value>;
+
+pub struct AmqpCloudEvent {
+    properties: Properties,
+    application_properties: ApplicationProperties,
+    data: Binary,
+}
+
+impl From<AmqpCloudEvent> for AmqpMessage {
+    fn from(event: AmqpCloudEvent) -> Self {
+        Message::builder()
+            .properties(event.properties)
+            .application_properties(event.application_properties)
+            .data(event.data)
+            .build()
+    }
+}
+
+impl TryFrom<AmqpMessage> for AmqpCloudEvent {
+    type Error = Error;
+
+    fn try_from(value: AmqpMessage) -> Result<Self, Self::Error> {
+        let data = match value.body {
+            Body::Data(AmqpData(data)) => data,
+            _ => return Err(Error::WrongEncoding {  })
+        };
+        let properties = value.properties
+            .ok_or(Error::WrongEncoding {  })?;
+        let application_properties = value.application_properties
+            .ok_or(Error::WrongEncoding {  })?;
+        Ok(Self {
+            properties,
+            application_properties,
+            data,
+        })
+    }
+}
+
+

src/binding/mod.rs

@@ -23,6 +23,8 @@ 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 {