#9 Encoders for MQTT by sbcd90 · Pull Request #104 · cloudevents/sdk-rust

sbcd90 · 2020-11-08T18:28:27Z

No description provided.

cloudevents-sdk-mqtt/Cargo.toml

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

slinkydeveloper · 2020-11-10T20:15:54Z

I marked this as 0.4, hopefully paho.mqtt new version will be released before our new version and we can get this one in. I'll review this PR asap

slinkydeveloper · 2020-11-10T20:23:16Z

Seems like musl doesn't like paho.mqtt? Is there a way to fix it? I see this issue is open here: eclipse-paho/paho.mqtt.rust#57

One thing I wanted with the integration between sdk-rust and mqtt is a crate supporting no_std (given we're also investing a lot of time in supporting no std #94). Is there any particular reason why @sbcd90 you chose paho.mqtt? Did you considered other mqtt clients supporting no_std?

sbcd90 · 2020-11-10T22:07:49Z

I did evaluate https://github.com/bytebeamio/rumqtt . I can use any other library.
However, i went with paho mqtt because I was not sure of future maintenance guarantees for rumqtt library.

sbcd90 · 2020-11-10T22:12:31Z

Please let me know if i should try switching the library.

fpagliughi · 2020-11-30T01:00:47Z

A Paho release went out last week (v0.8).

Another (v0.9) is in active development and should be out by the end of the year. The initial purpose for the next release was to get HTTP/S proxy support for websocket connections, but a few other additions popped up, and now we're also going to push for better cross-compiling support and fully static linking. That includes (optional) static OpenSSL linking and getting that musl support stable.

slinkydeveloper · 2020-11-30T07:44:35Z

@sbcd90 I think we can go ahead and work on this one, but let's be explicit we're supporting paho, so maybe rename it to cloudevents-sdk-paho-mqtt. Can you set the version to 0.8?

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 · 2020-11-30T20:33:10Z

@sbcd90 I think we can go ahead and work on this one, but let's be explicit we're supporting paho, so maybe rename it to cloudevents-sdk-paho-mqtt. Can you set the version to 0.8?

@slinkydeveloper I have set version to 0.8

slinkydeveloper · 2020-12-01T07:43:37Z

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

+}
+
+impl ConsumerMessageDeserializer {
+    fn get_mqtt_headers(message: &Message) -> Result<HashMap<String, Vec<u8>>> {


Do we need the result return here? I don't see any failing stmt here

updated this

slinkydeveloper · 2020-12-01T07:45:45Z

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

+    fn encoding(&self) -> Encoding {
+        match (
+            self.headers
+                .get("content-type")


Can you use the constant in headers?

updated this

slinkydeveloper · 2020-12-01T07:47:26Z

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

+        headers::MqttVersion::V3_1_1 => {
+            StructuredDeserializer::into_event(ConsumerMessageDeserializer::new(msg)?)
+        }
+    }


I may miss some knowledge about mqtt here, but why do we need the headers version? Also, why in V5 only Binary Deserializer is supported (aka only binary mode is supported), more than the general MessageDeserializer?

Also, can you infer the headers version from Message more than providing it manually as parameter?

now, V5 should allow both structured & binary mode. V3 only structured mode.

slinkydeveloper

So i have some concerns about the whole v3/v5 difference, why did you defined an extension for it? do we need it?

slinkydeveloper · 2020-12-09T15:04:51Z

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

+//! This library provides Mqtt protocol bindings for CloudEvents
+//! using the [paho.mqtt.rust](https://github.com/eclipse/paho.mqtt.rust) library.\\


Can we have here a sample usage in the docs? (look at actix-web integration module for details on how)

slinkydeveloper · 2020-12-09T15:06:41Z

example-projects/paho-mqtt-example/src/main.rs

+use cloudevents::{EventBuilderV10, EventBuilder};
+use cloudevents_sdk_mqtt::{MessageRecord, MessageBuilderExt, MqttVersion, MessageExt};
+
+fn consume_v3(broker: &str, topic_name: &str) {


Can you make the version just an argument more than 2 separate functions?

slinkydeveloper · 2020-12-09T15:07:23Z

example-projects/paho-mqtt-example/src/main.rs

+        process::exit(1);
+    });
+
+    if let Err(err) = block_on(async {


I wonder, can you make the consume_v3 function async in order to push the "async concern" to the outer scope (caller of consume)

slinkydeveloper · 2020-12-09T15:07:54Z

example-projects/paho-mqtt-example/src/main.rs

+        process::exit(1);
+    });
+
+    if let Err(err) = block_on(async {


Same as above

slinkydeveloper · 2020-12-09T15:08:04Z

example-projects/paho-mqtt-example/src/main.rs

+    }
+}
+
+fn produce_v3(broker: &str, topic_name: &str) {


Same as above

slinkydeveloper · 2020-12-09T15:09:35Z

example-projects/paho-mqtt-example/src/main.rs

+                .long("mode")
+                .help("enter \"consmer\" or \"producer\"")
+                .takes_value(true)
+                .possible_values(&["consumerV3", "producerV3", "consumerV5", "producerV5"])


Can you make the version a separate arg and choose some default?

slinkydeveloper · 2020-12-09T15:24:41Z

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

+        let mut hm = HashMap::new();
+        let prop_iterator = message.properties().iter(PropertyCode::UserProperty);


Do we really need this copy? Can't you just access to the user properties in the deserialize_binary method?

slinkydeveloper · 2020-12-09T16:23:28Z

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

+            self.headers
+                .get(headers::MQTT_VERSION_HEADER)
+                .map(|s| String::from_utf8(s.to_vec()).ok())
+                .flatten()
+                .map(|s| s.eq(headers::MQTT_V5_BINARY))
+                .unwrap_or(false),


So do we expect incoming messages to have this header?

slinkydeveloper · 2020-12-09T16:23:47Z

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

+
+    pub fn from_event(event: Event) -> Result<Self> {
+        match event
+            .extension(headers::MQTT_VERSION_HEADER)


Is this part of the spec?

sbcd90 · 2020-12-09T19:44:14Z

So i have some concerns about the whole v3/v5 difference, why did you defined an extension for it? do we need it?

Hi @slinkydeveloper , currently, v3 supports only structured mode. v5 supports both structured as well as binary mode according to spec.
This is because mqtt v3 has no support for headers while mqtt v5 has.
Now, in the current approach the default mode is structured mode which is supported by both v3 & v5. If binary mode needs to be used, then we need a flag to tell us if the message has to be deserialized using BinaryDeserializer.

The spec doesnt define how to set this mqtt version info.

Earlier, I was passing the flag as a parameter, now I pass it as an extension field. paho-mqtt passes version info as a param.
See

sdk-rust/example-projects/paho-mqtt-example/src/main.rs

Line 80 in ba09b47

.mqtt_version(5)

Another approach, we can follow is, ignore the mqtt version, & just trust the Content-Type header. This leaves us with a risk that we're depending on library api behavior a little bit.

match (
            self.headers
                .get(headers::CONTENT_TYPE)
                .map(|s| String::from_utf8(s.to_vec()).ok())
                .flatten()
                .map(|s| s.starts_with(headers::CLOUDEVENTS_JSON_HEADER))
                .unwrap_or(false),
            self.headers.len() > 0,
        ) {
            // v5 & structured
            (true, true) => Encoding::STRUCTURED,
            // v3 & structured
            (false, false) => Encoding::STRUCTURED,
            // v5 & binary
            (false, true) => Encoding::BINARY,
            _ => Encoding::UNKNOWN
        }

Kindly let me know according to you which is the best approach.

slinkydeveloper · 2020-12-10T11:41:18Z

Ok so one step at the time.

MQTT -> CloudEvent

First of all, i think you should remove the headers map you create and you should directly access to the Property borrow, in order to avoid allocating intermediate memory, which in this case it's not required.

For figuring out v3 vs v5 messages, I think we should simply do best effort: from the paho.mqtt rustdocs i see that you can always get the properties from a message https://docs.rs/paho-mqtt/0.8.0/paho_mqtt/message/struct.Message.html#method.properties and in case the UserProperty is empty, then we can assume the properties are not there so it's not mqtt v5 (hence we should just deserialize as structured).

CloudEvent -> MQTT

In this case, we should have an additional input in the serialization that explicits the mqtt version, like an enum MqttVersion created by us with MQTT_3 and MQTT_5 variants. I thought there was such enum in paho.mqtt, but there isn't. No extensions, the event must not be mutated.

Also note that i don't expect the user providing MessageRecord, this should be done under the hood:

pub trait MessageBuilderExt {
    fn event(self, Event, MqttVersion) -> MessageBuilder;
}

And the implementation would look like:

impl MessageBuilderExt for MessageBuilder {
    fn event(self, Event, MqttVersion) -> MessageBuilder {
        // mutate directly MessageBuilder here using MessageRecord
    }
}

Check out the actix-web, because it's a better one (I see that you probably looked at the MessageRecord thingy from Kafka, but that's kinda bad because of the rdkafka apis themselves)

Does that makes sense? Did I understood the problems here?

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 · 2020-12-22T19:40:43Z

Hi @slinkydeveloper , Thanks a lot for your detailed review. Refactored the code accordingly.
Added a sample README.

Can you please have a look now?

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

slinkydeveloper

To me this PR sounds good, @sbcd90 can you fix MUSL tests? Is paho-mqtt compatible with musl?

@Lazzaretti can you review this PR?

slinkydeveloper · 2021-01-18T16:52:50Z

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

+//! This library provides Mqtt protocol bindings for CloudEvents
+//! using the [paho.mqtt.rust](https://github.com/eclipse/paho.mqtt.rust) library.\\


slinkydeveloper · 2021-01-18T16:58:17Z

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

+
+        match version {
+            MQTT_5 => {
+                self = self.properties(message_record.headers.clone());


You can avoid clone destructuring message_record

slinkydeveloper · 2021-01-18T16:59:19Z

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

+        if let Some(s) = message_record.payload.as_ref() {
+            self = self.payload(s.to_vec());
+        }


No need to perform this copy, after destructing (as in the above comment) use a pattern matching here so you get ownership of s

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 · 2021-01-25T17:15:32Z

hi @slinkydeveloper , i fixed the remaining issues.
For musl, i get the error

Make sure you also have the development packages of openssl installed.
  For example, `libssl-dev` on Ubuntu or `openssl-devel` on Fedora.

slinkydeveloper · 2021-01-26T07:10:53Z

@sbcd90 try to modify the github action here: https://github.com/cloudevents/sdk-rust/blob/master/.github/workflows/rust_tests.yml#L31 adding openssl dev package setup

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

slinkydeveloper · 2021-01-27T08:15:09Z

example-projects/paho-mqtt-example/Cargo.toml

+serde_json = "^1.0"
+futures = "^0.3"
+tokio = { version = "^0.2", features = ["full"] }
+clap = "2.33.1"


Here you need to add

[workspace]

Like https://github.com/cloudevents/sdk-rust/blob/master/example-projects/warp-example/Cargo.toml

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 · 2021-01-27T22:09:29Z

hi @slinkydeveloper , the builds are fixed now.

Lazzaretti · 2021-01-31T16:38:11Z

cloudevents-sdk-paho-mqtt/README.md

+- Start the MQTT V3 Consumer
+
+```
+run --package <package-name> --bin <binary-name> -- --mode consumerV3 --broker tcp://localhost:1883 --topic test


what is run? can we have some more specific examples?

Lazzaretti · 2021-01-31T16:40:36Z

Cargo.toml

+openssl-sys = "*"
+openssl = { version = "*", features = ["vendored"] }


may we should add some version restrictions?

Good catch! Why in the main toml we need openssl?

Lazzaretti · 2021-01-31T16:55:53Z

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

+    pub fn from_event(event: Event, version: &headers::MqttVersion) -> Result<Self> {
+        match version {
+            headers::MqttVersion::MQTT_5 => {
+                BinaryDeserializer::deserialize_binary(event, MessageRecord::new())
+            }


As I understood MQTTv5 should support Binary and Structured Content Mode.
https://github.com/cloudevents/spec/blob/v1.0.1/mqtt-protocol-binding.md#32-structured-content-mode

So I would add a parameter to define the content mode (Binary or Structured) and not the MQTT version directly.

More than version you would use the Encoding struct? sounds good

Lazzaretti requested changes Nov 8, 2020

View reviewed changes

cloudevents-sdk-mqtt/Cargo.toml Outdated Show resolved Hide resolved

sbcd90 force-pushed the mqtt_rust branch from 171c2cc to db5ebd1 Compare November 9, 2020 07:50

sbcd90 added 2 commits November 8, 2020 23:57

cloudevents#9 Encoders for MQTT

5d446f4

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

9 Encoders for MQTT

f5302ed

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from db5ebd1 to f5302ed Compare November 9, 2020 08:03

slinkydeveloper added the enhancement New feature or request label Nov 10, 2020

slinkydeveloper added this to the 0.4 milestone Nov 10, 2020

slinkydeveloper self-assigned this Nov 10, 2020

9 Encoders for MQTT

a3ad7a9

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from ac05689 to a3ad7a9 Compare November 30, 2020 20:18

slinkydeveloper reviewed Dec 1, 2020

View reviewed changes

sbcd90 force-pushed the mqtt_rust branch from 6405dbb to e09fd5c Compare December 9, 2020 08:05

slinkydeveloper reviewed Dec 9, 2020

View reviewed changes

sbcd90 force-pushed the mqtt_rust branch from ba09b47 to 689e2cf Compare December 22, 2020 19:33

cloudevents#9 Encoders for MQTT

dc04020

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from 689e2cf to dc04020 Compare December 22, 2020 19:38

cloudevents#9 Encoders for MQTT

4f39c79

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from fe06610 to 4f39c79 Compare December 22, 2020 20:03

slinkydeveloper reviewed Jan 18, 2021

View reviewed changes

cloudevents#9 Encoders for MQTT

f5c3e8a

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from 2abee1d to f5c3e8a Compare January 25, 2021 16:23

cloudevents#9 Encoders for MQTT

445ccd4

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from c9a9a2a to 445ccd4 Compare January 25, 2021 16:31

sbcd90 added 8 commits January 26, 2021 17:26

cloudevents#9 Encoders for MQTT

7b1bdd7

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

8015532

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

9763b2e

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

d736c11

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

Merge branch 'master' into mqtt_rust

0feadc3

cloudevents#9 Encoders for MQTT

2a681db

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

fa7fde9

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

9de2c81

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

sbcd90 force-pushed the mqtt_rust branch from 9b955ad to 9de2c81 Compare January 27, 2021 06:06

slinkydeveloper reviewed Jan 27, 2021

View reviewed changes

sbcd90 added 5 commits January 27, 2021 06:55

cloudevents#9 Encoders for MQTT

f6f3e83

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

995d221

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

374288b

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

b09a661

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

cloudevents#9 Encoders for MQTT

c52975f

Signed-off-by: Subhobrata Dey <sbcd90@gmail.com>

slinkydeveloper linked an issue Jan 28, 2021 that may be closed by this pull request

Encoders for MQTT #9

Open

Lazzaretti reviewed Jan 31, 2021

View reviewed changes

ctron added a commit to ctron/sdk-rust that referenced this pull request Mar 1, 2021

[cloudevents#104] Allowing handling URI-ref types

e21bcd1

slinkydeveloper removed this from the 0.4 milestone Jul 7, 2021

		//! This library provides Mqtt protocol bindings for CloudEvents
		//! using the [paho.mqtt.rust](https://github.com/eclipse/paho.mqtt.rust) library.\\

		let mut hm = HashMap::new();
		let prop_iterator = message.properties().iter(PropertyCode::UserProperty);

		openssl-sys = "*"
		openssl = { version = "*", features = ["vendored"] }

Comments

Conversation

sbcd90 commented Nov 8, 2020

Uh oh!

Uh oh!

slinkydeveloper commented Nov 10, 2020

Uh oh!

slinkydeveloper commented Nov 10, 2020

Uh oh!

sbcd90 commented Nov 10, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sbcd90 commented Nov 10, 2020

Uh oh!

fpagliughi commented Nov 30, 2020

Uh oh!

slinkydeveloper commented Nov 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sbcd90 commented Nov 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

slinkydeveloper left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sbcd90 commented Dec 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

slinkydeveloper commented Dec 10, 2020

MQTT -> CloudEvent

CloudEvent -> MQTT

Uh oh!

sbcd90 commented Dec 22, 2020

Uh oh!

slinkydeveloper left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sbcd90 commented Jan 25, 2021

Uh oh!

slinkydeveloper commented Jan 26, 2021

Uh oh!

Choose a reason for hiding this comment

sbcd90 commented Nov 10, 2020 •

edited

Loading

slinkydeveloper commented Nov 30, 2020 •

edited

Loading

sbcd90 commented Nov 30, 2020 •

edited

Loading

sbcd90 commented Dec 9, 2020 •

edited

Loading