Skip to content

Commit 1214976

Browse files
minghuawminghuaw
committed
removed AmqpCloudEvent
1 parent 2593f81 commit 1214976

File tree

3 files changed

+67
-77
lines changed

3 files changed

+67
-77
lines changed

‎src/binding/fe2o3_amqp/deserializer.rs

Lines changed: 23 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,10 @@ use crate::{
1313

1414
use super::{
1515
constants::{prefixed, DATACONTENTTYPE},
16-
AmqpCloudEvent, ATTRIBUTE_PREFIX,
16+
ATTRIBUTE_PREFIX, AmqpMessage,
1717
};
1818

19-
impl BinaryDeserializer for AmqpCloudEvent {
19+
impl BinaryDeserializer for AmqpMessage {
2020
fn deserialize_binary<R: Sized, V: BinarySerializer<R>>(
2121
mut self,
2222
mut serializer: V,
@@ -27,6 +27,8 @@ impl BinaryDeserializer for AmqpCloudEvent {
2727
let spec_version = {
2828
let value = self
2929
.application_properties
30+
.as_mut()
31+
.ok_or(Error::WrongEncoding { })?
3032
.remove(prefixed::SPECVERSION)
3133
.ok_or(Error::WrongEncoding {})
3234
.map(|val| match val {
@@ -38,23 +40,24 @@ impl BinaryDeserializer for AmqpCloudEvent {
3840
serializer = serializer.set_spec_version(spec_version.clone())?;
3941

4042
// datacontenttype
41-
serializer = match self.content_type {
42-
Some(Symbol(content_type)) => serializer
43+
serializer = match self.properties.map(|p| p.content_type) {
44+
Some(Some(Symbol(content_type))) => serializer
4345
.set_attribute(DATACONTENTTYPE, MessageAttributeValue::String(content_type))?,
44-
None => serializer,
46+
_ => serializer,
4547
};
4648

4749
// remaining attributes
4850
let attributes = spec_version.attribute_names();
49-
50-
for (key, value) in self.application_properties.0.into_iter() {
51-
if let Some(key) = key.strip_prefix(ATTRIBUTE_PREFIX) {
52-
if attributes.contains(&key) {
53-
let value = MessageAttributeValue::try_from((key, value))?;
54-
serializer = serializer.set_attribute(key, value)?;
55-
} else {
56-
let value = MessageAttributeValue::try_from(value)?;
57-
serializer = serializer.set_extension(key, value)?;
51+
if let Some(application_properties) = self.application_properties {
52+
for (key, value) in application_properties.0.into_iter() {
53+
if let Some(key) = key.strip_prefix(ATTRIBUTE_PREFIX) {
54+
if attributes.contains(&key) {
55+
let value = MessageAttributeValue::try_from((key, value))?;
56+
serializer = serializer.set_attribute(key, value)?;
57+
} else {
58+
let value = MessageAttributeValue::try_from(value)?;
59+
serializer = serializer.set_extension(key, value)?;
60+
}
5861
}
5962
}
6063
}
@@ -70,7 +73,7 @@ impl BinaryDeserializer for AmqpCloudEvent {
7073
}
7174
}
7275

73-
impl StructuredDeserializer for AmqpCloudEvent {
76+
impl StructuredDeserializer for AmqpMessage {
7477
fn deserialize_structured<R: Sized, V: StructuredSerializer<R>>(
7578
self,
7679
serializer: V,
@@ -86,12 +89,14 @@ impl StructuredDeserializer for AmqpCloudEvent {
8689
}
8790
}
8891

89-
impl MessageDeserializer for AmqpCloudEvent {
92+
impl MessageDeserializer for AmqpMessage {
9093
fn encoding(&self) -> Encoding {
9194
match self
92-
.content_type
95+
.properties
9396
.as_ref()
94-
.map(|s| s.starts_with(CLOUDEVENTS_JSON_HEADER))
97+
.map(|p| p.content_type.as_ref()
98+
.map(|s| s.starts_with(CLOUDEVENTS_JSON_HEADER))
99+
).flatten()
95100
{
96101
Some(true) => Encoding::STRUCTURED,
97102
Some(false) => Encoding::BINARY,

‎src/binding/fe2o3_amqp/mod.rs

Lines changed: 5 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,13 @@
11
//! Implements AMQP 1.0 binding for CloudEvents
22
3-
use std::collections::HashMap;
43
use std::convert::TryFrom;
54

65
use chrono::{Utc, TimeZone};
7-
use fe2o3_amqp_lib::types::messaging::{ApplicationProperties, Body, Message, Properties};
8-
use fe2o3_amqp_lib::types::primitives::{Binary, SimpleValue, Symbol, Timestamp, Value};
6+
use fe2o3_amqp_lib::types::messaging::{Body, Message};
7+
use fe2o3_amqp_lib::types::primitives::{Binary, SimpleValue, Timestamp, Value};
98

10-
use crate::event::{AttributeValue, ExtensionValue};
9+
use crate::event::{AttributeValue};
1110
use crate::message::{Error, MessageAttributeValue};
12-
use crate::Event;
1311

1412
use self::constants::{
1513
prefixed, DATACONTENTTYPE, DATASCHEMA, ID, SOURCE, SPECVERSION, SUBJECT, TIME, TYPE,
@@ -27,43 +25,8 @@ mod constants;
2725
/// The generic parameter can be anything that implements `Serialize` and `Deserialize` but is of
2826
/// no importance because all CloudEvents are using the `Body::Data` as the body section type. For
2927
/// convenience, this type alias chose `Value` as the value of the generic parameter
30-
pub type AmqpMessage = Message<Value>;
31-
32-
pub type AmqpBody = Body<Value>;
33-
34-
pub type Extensions = HashMap<String, ExtensionValue>;
35-
36-
/// The receiver of the event can distinguish between the two modes by inspecting the content-type
37-
/// message property field. If the value is prefixed with the CloudEvents media type
38-
/// application/cloudevents, indicating the use of a known event format, the receiver uses
39-
/// structured mode, otherwise it defaults to binary mode.
40-
pub struct AmqpCloudEvent {
41-
content_type: Option<Symbol>,
42-
application_properties: ApplicationProperties,
43-
body: AmqpBody,
44-
}
45-
46-
impl AmqpCloudEvent {
47-
pub fn from_event(event: Event) -> Result<Self, Error> {
48-
todo!()
49-
}
50-
}
51-
52-
impl From<AmqpCloudEvent> for AmqpMessage {
53-
fn from(event: AmqpCloudEvent) -> Self {
54-
let mut properties = Properties::default();
55-
properties.content_type = event.content_type;
56-
Message {
57-
header: None,
58-
delivery_annotations: None,
59-
message_annotations: None,
60-
properties: Some(properties),
61-
application_properties: Some(event.application_properties),
62-
body: event.body,
63-
footer: None,
64-
}
65-
}
66-
}
28+
type AmqpMessage = Message<Value>;
29+
type AmqpBody = Body<Value>;
6730

6831
impl<'a> From<AttributeValue<'a>> for SimpleValue {
6932
fn from(value: AttributeValue) -> Self {

‎src/binding/fe2o3_amqp/serializer.rs

Lines changed: 39 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,37 +1,51 @@
1-
use fe2o3_amqp_types::primitives::{SimpleValue, Symbol, Binary};
2-
use fe2o3_amqp_types::messaging::{Data as AmqpData};
1+
use fe2o3_amqp_types::messaging::{Data as AmqpData, Properties, ApplicationProperties};
2+
use fe2o3_amqp_types::primitives::{Binary, SimpleValue, Symbol};
33

4+
use crate::binding::header_prefix;
45
use crate::message::StructuredSerializer;
5-
use crate::{message::{BinarySerializer, MessageAttributeValue, Error}, event::SpecVersion};
6+
use crate::{
7+
event::SpecVersion,
8+
message::{BinarySerializer, Error, MessageAttributeValue},
9+
};
610

711
use super::constants::DATACONTENTTYPE;
8-
use super::{AmqpCloudEvent, ATTRIBUTE_PREFIX, AmqpBody};
12+
use super::{AmqpBody, AmqpMessage, ATTRIBUTE_PREFIX};
913

10-
impl BinarySerializer<AmqpCloudEvent> for AmqpCloudEvent {
14+
impl BinarySerializer<AmqpMessage> for AmqpMessage {
1115
fn set_spec_version(mut self, spec_version: SpecVersion) -> crate::message::Result<Self> {
1216
let key = String::from("cloudEvents:specversion");
1317
let value = String::from(spec_version.as_str());
14-
self.application_properties.insert(key, SimpleValue::from(value));
18+
self.application_properties
19+
.get_or_insert(ApplicationProperties::default())
20+
.insert(key, SimpleValue::from(value));
1521
Ok(self)
1622
}
1723

18-
fn set_attribute(mut self, name: &str, value: MessageAttributeValue) -> crate::message::Result<Self> {
24+
fn set_attribute(
25+
mut self,
26+
name: &str,
27+
value: MessageAttributeValue,
28+
) -> crate::message::Result<Self> {
1929
// For the binary mode, the AMQP content-type property field value maps directly to the
2030
// CloudEvents datacontenttype attribute.
21-
//
31+
//
2232
// All CloudEvents attributes with exception of datacontenttype MUST be individually mapped
2333
// to and from the AMQP application-properties section.
2434
if name == DATACONTENTTYPE {
25-
self.content_type = match value {
35+
self.properties
36+
.get_or_insert(Properties::default())
37+
.content_type = match value {
2638
MessageAttributeValue::String(s) => Some(Symbol::from(s)),
27-
_ => return Err(Error::WrongEncoding { })
39+
_ => return Err(Error::WrongEncoding {}),
2840
}
2941
} else {
3042
// CloudEvent attributes are prefixed with "cloudEvents:" for use in the
3143
// application-properties section
32-
let key = format!("{}:{}", ATTRIBUTE_PREFIX, name);
44+
let key = header_prefix(ATTRIBUTE_PREFIX, name);
3345
let value = SimpleValue::from(value);
34-
self.application_properties.insert(key, value);
46+
self.application_properties
47+
.get_or_insert(ApplicationProperties::default())
48+
.insert(key, value);
3549
}
3650

3751
Ok(self)
@@ -43,10 +57,16 @@ impl BinarySerializer<AmqpCloudEvent> for AmqpCloudEvent {
4357
// systems that also process the message. Extension specifications that do this SHOULD specify
4458
// how receivers are to interpret messages if the copied values differ from the cloud-event
4559
// serialized values.
46-
fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> crate::message::Result<Self> {
47-
let key = format!("{}:{}", ATTRIBUTE_PREFIX, name);
60+
fn set_extension(
61+
mut self,
62+
name: &str,
63+
value: MessageAttributeValue,
64+
) -> crate::message::Result<Self> {
65+
let key = header_prefix(ATTRIBUTE_PREFIX, name);
4866
let value = SimpleValue::from(value);
49-
self.application_properties.insert(key, value);
67+
self.application_properties
68+
.get_or_insert(ApplicationProperties::default())
69+
.insert(key, value);
5070
Ok(self)
5171
}
5272

@@ -61,9 +81,11 @@ impl BinarySerializer<AmqpCloudEvent> for AmqpCloudEvent {
6181
}
6282
}
6383

64-
impl StructuredSerializer<AmqpCloudEvent> for AmqpCloudEvent {
84+
impl StructuredSerializer<AmqpMessage> for AmqpMessage {
6585
fn set_structured_event(mut self, bytes: Vec<u8>) -> crate::message::Result<Self> {
66-
self.content_type = Some(Symbol::from("application/cloudevents+json; charset=utf-8"));
86+
self.properties
87+
.get_or_insert(Properties::default())
88+
.content_type = Some(Symbol::from("application/cloudevents+json; charset=utf-8"));
6789
self.body = AmqpBody::Data(AmqpData(Binary::from(bytes)));
6890
Ok(self)
6991
}

0 commit comments

Comments
 (0)