Skip to content

Commit 171c2cc

Browse files
sbcd90sbcd90
committed
#9 Encoders for MQTT
1 parent 89a6702 commit 171c2cc

File tree

5 files changed

+99
-79
lines changed

5 files changed

+99
-79
lines changed

‎cloudevents-sdk-mqtt/Cargo.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@ description = "CloudEvents official Rust SDK - Mqtt integration"
99
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
1010

1111
[dependencies]
12-
cloudevents-sdk = { version = "0.2.0", path = ".." }
12+
cloudevents-sdk = { version = "0.3.0", path = ".." }
1313
lazy_static = "1.4.0"
1414
paho-mqtt = { path = "../../paho.mqtt.rust" }
1515
chrono = { version = "^0.4", features = ["serde"] }

‎cloudevents-sdk-mqtt/src/headers.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ fn attributes_to_headers(it: impl Iterator<Item = &'static str>) -> HashMap<&'st
1616
(s, attribute_name_to_header!(s))
1717
}
1818
})
19-
.collect()
19+
.collect()
2020
}
2121

2222
lazy_static! {
@@ -32,4 +32,4 @@ pub enum MqttVersion {
3232
V3_1,
3333
V3_1_1,
3434
V5,
35-
}
35+
}

‎cloudevents-sdk-mqtt/src/lib.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,13 @@
22
//! using the [paho.mqtt.rust](https://github.com/eclipse/paho.mqtt.rust) library.\\
33
#[macro_use]
44
mod headers;
5-
mod mqtt_producer_record;
65
mod mqtt_consumer_record;
6+
mod mqtt_producer_record;
77

88
pub use mqtt_consumer_record::record_to_event;
99
pub use mqtt_consumer_record::ConsumerMessageDeserializer;
1010
pub use mqtt_consumer_record::MessageExt;
1111

12+
pub use headers::MqttVersion;
1213
pub use mqtt_producer_record::MessageBuilderExt;
1314
pub use mqtt_producer_record::MessageRecord;
14-
pub use headers::MqttVersion;

‎cloudevents-sdk-mqtt/src/mqtt_consumer_record.rs‎

Lines changed: 27 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,9 @@
11
use crate::headers;
22
use cloudevents::event::SpecVersion;
3-
use cloudevents::message::{Result, BinarySerializer, BinaryDeserializer, MessageAttributeValue,
4-
MessageDeserializer, Encoding, StructuredSerializer, StructuredDeserializer};
3+
use cloudevents::message::{
4+
BinaryDeserializer, BinarySerializer, Encoding, MessageAttributeValue, MessageDeserializer,
5+
Result, StructuredDeserializer, StructuredSerializer,
6+
};
57
use cloudevents::{message, Event};
68
use paho_mqtt::{Message, PropertyCode};
79
use std::collections::HashMap;
@@ -37,7 +39,7 @@ impl ConsumerMessageDeserializer {
3739
impl BinaryDeserializer for ConsumerMessageDeserializer {
3840
fn deserialize_binary<R: Sized, V: BinarySerializer<R>>(mut self, mut visitor: V) -> Result<R> {
3941
if self.encoding() != Encoding::BINARY {
40-
return Err(message::Error::WrongEncoding {})
42+
return Err(message::Error::WrongEncoding {});
4143
}
4244

4345
let spec_version = SpecVersion::try_from(
@@ -124,9 +126,15 @@ impl MessageDeserializer for ConsumerMessageDeserializer {
124126

125127
pub fn record_to_event(msg: &Message, version: headers::MqttVersion) -> Result<Event> {
126128
match version {
127-
headers::MqttVersion::V5 => BinaryDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?),
128-
headers::MqttVersion::V3_1 => StructuredDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?),
129-
headers::MqttVersion::V3_1_1 => StructuredDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?),
129+
headers::MqttVersion::V5 => {
130+
BinaryDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?)
131+
}
132+
headers::MqttVersion::V3_1 => {
133+
StructuredDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?)
134+
}
135+
headers::MqttVersion::V3_1_1 => {
136+
StructuredDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?)
137+
}
130138
}
131139
}
132140

@@ -145,11 +153,12 @@ mod tests {
145153
use super::*;
146154
use crate::mqtt_producer_record::MessageRecord;
147155

156+
use crate::MessageBuilderExt;
148157
use chrono::Utc;
158+
use cloudevents::event::Data;
149159
use cloudevents::{EventBuilder, EventBuilderV10};
150-
use crate::MessageBuilderExt;
160+
use paho_mqtt::MessageBuilder;
151161
use serde_json::json;
152-
use cloudevents::event::Data;
153162

154163
#[test]
155164
fn test_binary_record() {
@@ -160,8 +169,10 @@ mod tests {
160169
.ty("example.test")
161170
.time(time)
162171
.source("http://localhost")
163-
.data("application/json",
164-
Data::Binary(String::from("{\"hello\":\"world\"}").into_bytes()))
172+
.data(
173+
"application/json",
174+
Data::Binary(String::from("{\"hello\":\"world\"}").into_bytes()),
175+
)
165176
.extension("someint", "10")
166177
.build()
167178
.unwrap();
@@ -178,7 +189,7 @@ mod tests {
178189
.unwrap(),
179190
headers::MqttVersion::V5,
180191
)
181-
.unwrap();
192+
.unwrap();
182193

183194
let msg = MessageBuilder::new()
184195
.topic("test")
@@ -220,6 +231,9 @@ mod tests {
220231
.qos(1)
221232
.finalize();
222233

223-
assert_eq!(msg.to_event(headers::MqttVersion::V3_1_1).unwrap(), expected)
234+
assert_eq!(
235+
msg.to_event(headers::MqttVersion::V3_1_1).unwrap(),
236+
expected
237+
)
224238
}
225-
}
239+
}

‎cloudevents-sdk-mqtt/src/mqtt_producer_record.rs‎

Lines changed: 67 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,11 @@
11
use super::headers;
2-
use paho_mqtt::{Properties, Property, PropertyCode, MessageBuilder};
3-
use cloudevents::message::{BinaryDeserializer, BinarySerializer, MessageAttributeValue, Result,
4-
StructuredDeserializer, StructuredSerializer, Error};
5-
use cloudevents::Event;
62
use cloudevents::event::SpecVersion;
3+
use cloudevents::message::{
4+
BinaryDeserializer, BinarySerializer, Error, MessageAttributeValue, Result,
5+
StructuredDeserializer, StructuredSerializer,
6+
};
7+
use cloudevents::Event;
8+
use paho_mqtt::{MessageBuilder, Properties, Property, PropertyCode};
79
use std::option::Option::Some;
810

911
pub struct MessageRecord {
@@ -22,68 +24,76 @@ impl MessageRecord {
2224

2325
pub fn from_event(event: Event, version: headers::MqttVersion) -> Result<Self> {
2426
match version {
25-
headers::MqttVersion::V5 => BinaryDeserializer::deserialize_binary(event, MessageRecord::new()),
26-
headers::MqttVersion::V3_1 => StructuredDeserializer::deserialize_structured(event, MessageRecord::new()),
27-
headers::MqttVersion::V3_1_1 => StructuredDeserializer::deserialize_structured(event, MessageRecord::new()),
27+
headers::MqttVersion::V5 => {
28+
BinaryDeserializer::deserialize_binary(event, MessageRecord::new())
29+
}
30+
headers::MqttVersion::V3_1 => {
31+
StructuredDeserializer::deserialize_structured(event, MessageRecord::new())
32+
}
33+
headers::MqttVersion::V3_1_1 => {
34+
StructuredDeserializer::deserialize_structured(event, MessageRecord::new())
35+
}
2836
}
2937
}
3038
}
3139

3240
impl BinarySerializer<MessageRecord> for MessageRecord {
3341
fn set_spec_version(mut self, spec_version: SpecVersion) -> Result<Self> {
34-
match Property::new_string_pair(PropertyCode::UserProperty, headers::SPEC_VERSION_HEADER,
35-
spec_version.as_str()) {
36-
Ok(property) => {
37-
match self.headers.push(property) {
38-
Err(e) => Err(Error::Other {
39-
source: Box::new(e)
40-
}),
41-
_ => Ok(self)
42-
}
42+
match Property::new_string_pair(
43+
PropertyCode::UserProperty,
44+
headers::SPEC_VERSION_HEADER,
45+
spec_version.as_str(),
46+
) {
47+
Ok(property) => match self.headers.push(property) {
48+
Err(e) => Err(Error::Other {
49+
source: Box::new(e),
50+
}),
51+
_ => Ok(self),
4352
},
4453
_ => Err(Error::UnrecognizedAttributeName {
45-
name: headers::SPEC_VERSION_HEADER.to_string()
46-
})
54+
name: headers::SPEC_VERSION_HEADER.to_string(),
55+
}),
4756
}
4857
}
4958

5059
fn set_attribute(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
51-
match Property::new_string_pair(PropertyCode::UserProperty, &headers::ATTRIBUTES_TO_MQTT_HEADERS
52-
.get(name)
53-
.ok_or(cloudevents::message::Error::UnrecognizedAttributeName {
54-
name: String::from(name),
55-
})?
56-
.clone()[..],
57-
&value.to_string()[..]) {
58-
Ok(property) => {
59-
match self.headers.push(property) {
60-
Err(e) => Err(Error::Other {
61-
source: Box::new(e)
62-
}),
63-
_ => Ok(self)
64-
}
60+
match Property::new_string_pair(
61+
PropertyCode::UserProperty,
62+
&headers::ATTRIBUTES_TO_MQTT_HEADERS
63+
.get(name)
64+
.ok_or(cloudevents::message::Error::UnrecognizedAttributeName {
65+
name: String::from(name),
66+
})?
67+
.clone()[..],
68+
&value.to_string()[..],
69+
) {
70+
Ok(property) => match self.headers.push(property) {
71+
Err(e) => Err(Error::Other {
72+
source: Box::new(e),
73+
}),
74+
_ => Ok(self),
6575
},
6676
_ => Err(Error::UnrecognizedAttributeName {
67-
name: headers::SPEC_VERSION_HEADER.to_string()
68-
})
77+
name: headers::SPEC_VERSION_HEADER.to_string(),
78+
}),
6979
}
7080
}
7181

7282
fn set_extension(mut self, name: &str, value: MessageAttributeValue) -> Result<Self> {
73-
match Property::new_string_pair(PropertyCode::UserProperty,
74-
&attribute_name_to_header!(name)[..],
75-
&value.to_string()[..]) {
76-
Ok(property) => {
77-
match self.headers.push(property) {
78-
Err(e) => Err(Error::Other {
79-
source: Box::new(e)
80-
}),
81-
_ => Ok(self)
82-
}
83+
match Property::new_string_pair(
84+
PropertyCode::UserProperty,
85+
&attribute_name_to_header!(name)[..],
86+
&value.to_string()[..],
87+
) {
88+
Ok(property) => match self.headers.push(property) {
89+
Err(e) => Err(Error::Other {
90+
source: Box::new(e),
91+
}),
92+
_ => Ok(self),
8393
},
8494
_ => Err(Error::UnrecognizedAttributeName {
85-
name: headers::SPEC_VERSION_HEADER.to_string()
86-
})
95+
name: headers::SPEC_VERSION_HEADER.to_string(),
96+
}),
8797
}
8898
}
8999

@@ -100,14 +110,15 @@ impl BinarySerializer<MessageRecord> for MessageRecord {
100110

101111
impl StructuredSerializer<MessageRecord> for MessageRecord {
102112
fn set_structured_event(mut self, bytes: Vec<u8>) -> Result<MessageRecord> {
103-
match Property::new_string_pair(PropertyCode::UserProperty,
104-
headers::CONTENT_TYPE, headers::CLOUDEVENTS_JSON_HEADER) {
105-
Ok(property) => {
106-
match self.headers.push(property) {
107-
_ => ()
108-
}
113+
match Property::new_string_pair(
114+
PropertyCode::UserProperty,
115+
headers::CONTENT_TYPE,
116+
headers::CLOUDEVENTS_JSON_HEADER,
117+
) {
118+
Ok(property) => match self.headers.push(property) {
119+
_ => (),
109120
},
110-
_ => ()
121+
_ => (),
111122
}
112123
self.payload = Some(bytes);
113124

@@ -116,16 +127,11 @@ impl StructuredSerializer<MessageRecord> for MessageRecord {
116127
}
117128

118129
pub trait MessageBuilderExt {
119-
fn message_record(
120-
self,
121-
message_record: & MessageRecord,
122-
) -> MessageBuilder;
130+
fn message_record(self, message_record: &MessageRecord) -> MessageBuilder;
123131
}
124132

125133
impl MessageBuilderExt for MessageBuilder {
126-
fn message_record(mut self,
127-
message_record: & MessageRecord
128-
) -> MessageBuilder {
134+
fn message_record(mut self, message_record: &MessageRecord) -> MessageBuilder {
129135
self = self.properties(message_record.headers.clone());
130136

131137
if let Some(s) = message_record.payload.as_ref() {
@@ -134,4 +140,4 @@ impl MessageBuilderExt for MessageBuilder {
134140

135141
self
136142
}
137-
}
143+
}

0 commit comments

Comments
 (0)