MF-1079 - Add MQTT forwarder #1164

dborovcanin · 2020-05-05T09:35:46Z

What does this do?

This pull request introduces MQTT forwarder for multi-protocol support explained in #1079.

Which issue(s) does this PR fix/relate to?

This pull request fixes #1079.

List any changes that modify/break current functionality

Messages published from other protocols will be received on MQTT subs.

Have you included tests for your changes?

No.

Did you document any new/modified functionality?

This fix will make the MQTT adapter work as intended.

codecov-io · 2020-05-05T09:44:08Z

Codecov Report

Merging #1164 into master will not change coverage.
The diff coverage is n/a.

@@           Coverage Diff           @@
##           master    #1164   +/-   ##
=======================================
  Coverage   74.03%   74.03%           
=======================================
  Files         102      102           
  Lines        6883     6883           
=======================================
  Hits         5096     5096           
  Misses       1433     1433           
  Partials      354      354

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update f9432c6...fe27966. Read the comment docs.

blokovi · 2020-05-05T17:24:30Z

messaging/mqtt/subscriber.go

+package mqtt
+
+import (
+	"errors"


why not "github.com/mainflux/mainflux/errors" ?

manuio · 2020-05-05T17:29:16Z

cmd/mqtt/main.go

@@ -130,6 +136,22 @@ func main() {

 	cc := thingsapi.NewClient(conn, thingsTracer, cfg.thingsAuthTimeout)

+	ps, err := nats.NewPubSub(cfg.natsURL, "mqtt", logger)


I would use natsPubSub here instead of ps

Maybe nps can be a good compromise? Also mqttPub can become mp and pub below can become np.

In any case, using long or short names - we should be consistent - which is not currently the case.

manuio · 2020-05-05T17:34:39Z

cmd/mqtt/main.go

+		logger.Error(fmt.Sprintf("Failed to create MQTT publisher: %s", err))
+		os.Exit(1)
+	}
+	if err := mqtt.Forward(nats.SubjectAllChannels, ps, mqttPub); err != nil {


should we configure nats.SubjectAllChannels from pubsub?

manuio · 2020-05-05T17:53:11Z

mqtt/forwarder.go

+		if msg.Protocol == protocol {
+			return nil
+		}
+		topic := fmt.Sprintf("%s.%s.%s.%s", channels, msg.Channel, messages, msg.Subtopic)


Is topic ok if Subtopic is empty?

drasko · 2020-05-05T21:19:09Z

cmd/mqtt/main.go

@@ -130,6 +136,22 @@ func main() {

 	cc := thingsapi.NewClient(conn, thingsTracer, cfg.thingsAuthTimeout)

+	ps, err := nats.NewPubSub(cfg.natsURL, "mqtt", logger)


Maybe nps can be a good compromise? Also mqttPub can become mp and pub below can become np.

In any case, using long or short names - we should be consistent - which is not currently the case.

drasko · 2020-05-05T21:36:09Z

messaging/mqtt/pubsub.go

+		SetClientID(id).
+		SetCleanSession(false)
+	client := mqtt.NewClient(opts)
+	tkn := client.Connect()


What does tkn stand for?

It stands for token. That's the construct Paho internally returns on the async operations.

Why not token?

It is bit confusing IMHO, maybe comment could help?

@drasko why a comment?

why not token than?

drasko · 2020-05-05T21:37:57Z

messaging/mqtt/subscriber.go

+import (
+	"errors"
+	"fmt"
+


Kill this white line

drasko

LGTM

manuio · 2020-05-06T15:31:59Z

messaging/mqtt/publisher.go

+func (pub publisher) Publish(topic string, msg messaging.Message) error {
+	token := pub.client.Publish(topic, qos, false, msg.Payload)
+	ok := token.WaitTimeout(pub.timeout)
+	if ok && token.Error() != nil {


What if token.Error() != nil in the previous line?

dont you like this if token.Wait() && token.Error() != nil { better ?

manuio · 2020-05-06T15:33:32Z

mqtt/forwarder.go

+		// Use concatenation instead of mft.Sprintf for the
+		// sake of simplicity and performance.
+		topic := channels + "." + msg.Channel + "." + messages
+		if msg.Subtopic != "" {
+			topic += "." + msg.Subtopic
+		}
+		topic = strings.ReplaceAll(topic, ".", "/")


why not simply use /?
Also, use fmt.Sprint() to concatenate strings

Changed to /. String buffer would be an overkill here, and the comment explains why I don't want to use Sprintf.

Makes sense, and I think we should consider avoiding sprintf in general where + string concatenation can bring simplicity. We need to sync internally and understand when to use one and when another.

mteodor · 2020-05-06T15:59:53Z

messaging/mqtt/subscriber.go

+	token := sub.client.Subscribe(topic, qos, sub.mqttHandler(handler))
+	ok := token.WaitTimeout(sub.timeout)
+	if ok && token.Error() != nil {
+		return token.Error()


if we return token.Error() on multiple places it would be good to wrap with another error so that exact place can be more easily detected

I would leave error wrapping for the layer above MQTT Publisher or Subscriber. We only use it as a wrapper around MQTT client, so returning the real error is sufficient, IMHO. However, I'll improve the message of the errors we return if token is expired to be more specific ("failed to subscribe/unsubscribe due to timeout reached" is better than just "failed to subscribe/unsubscribe").

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

manuio · 2020-05-06T18:13:54Z

cmd/mqtt/main.go

@@ -16,6 +16,7 @@ import (
 	"github.com/mainflux/mainflux"
 	"github.com/mainflux/mainflux/logger"
 	"github.com/mainflux/mainflux/messaging"
+	mqttpub "github.com/mainflux/mainflux/messaging/mqtt"
 	"github.com/mainflux/mainflux/messaging/nats"
 	mqtt "github.com/mainflux/mainflux/mqtt"
 	mr "github.com/mainflux/mainflux/mqtt/redis"


I would use mqttRedis here.

messaging/mqtt/pubsub.go

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

manuio

LGTM

* Add MQTT forwarder Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Cleanup forwarder code Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Use MQTT Publisher in MQTT forwarder Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Cleanup MQTT messaging Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Add Paho client timeout errors Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Simplify MQTT fowarder Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Fix naming in main method Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Use interface and struct instead of function Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Use Mainflux errors package Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Rename `tkn` to `token` Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Use "/" instead of "." as topic separator Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Use async MQTT Publisher Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Fix timeout errors messages Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Add connect token check Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com> * Rename package alias Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

dborovcanin requested a review from a team as a code owner May 5, 2020 09:35

dborovcanin force-pushed the mqtt-forwarder branch from 327bf48 to ccbfd7c Compare May 5, 2020 13:31

dborovcanin changed the base branch from relay to master May 5, 2020 16:30

dborovcanin changed the title ~~NOISSUE - Add MQTT forwarder~~ MF-1079 - Add MQTT forwarder May 5, 2020

dborovcanin force-pushed the mqtt-forwarder branch from bfb762b to 3126070 Compare May 5, 2020 17:17

blokovi reviewed May 5, 2020

View reviewed changes

manuio reviewed May 5, 2020

View reviewed changes

manuio suggested changes May 5, 2020

View reviewed changes

drasko requested changes May 5, 2020

View reviewed changes

dborovcanin force-pushed the mqtt-forwarder branch from dd75b21 to be96166 Compare May 6, 2020 10:16

drasko previously approved these changes May 6, 2020

View reviewed changes

drasko mentioned this pull request May 6, 2020

MFX-1079 - Implement multi-protocol relay for MQTT #1155

Closed

manuio suggested changes May 6, 2020

View reviewed changes

dborovcanin dismissed drasko’s stale review via 13ea0ed May 6, 2020 15:38

mteodor reviewed May 6, 2020

View reviewed changes

dborovcanin force-pushed the mqtt-forwarder branch 2 times, most recently from f75d21a to fe27966 Compare May 6, 2020 17:40

dborovcanin added 10 commits May 6, 2020 19:41

Add MQTT forwarder

25755b8

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Cleanup forwarder code

b892220

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Use MQTT Publisher in MQTT forwarder

87c7abc

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Cleanup MQTT messaging

af3bb0c

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Add Paho client timeout errors

9b78705

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Simplify MQTT fowarder

0cdb5d8

10000

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Fix naming in main method

5021ae6

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Use interface and struct instead of function

03f4f6e

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Use Mainflux errors package

9565a85

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Rename tkn to token

3f0aa28

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

dborovcanin added 3 commits May 6, 2020 19:41

Use "/" instead of "." as topic separator

a6dc770

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Use async MQTT Publisher

bb3a5ed

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Fix timeout errors messages

fe27966

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

manuio reviewed May 6, 2020

View reviewed changes

manuio suggested changes May 6, 2020

View reviewed changes

messaging/mqtt/pubsub.go Show resolved Hide resolved

dborovcanin added 2 commits May 6, 2020 22:57

Add connect token check

0c3120d

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

Rename package alias

47c2eb6

Signed-off-by: Dušan Borovčanin <dusan.borovcanin@mainflux.com>

manuio approved these changes May 6, 2020

View reviewed changes

mteodor approved these changes May 7, 2020

View reviewed changes

manuio merged commit b8be181 into absmach:master May 7, 2020

dborovcanin deleted the mqtt-forwarder branch February 17, 2021 09:09

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

MF-1079 - Add MQTT forwarder #1164

MF-1079 - Add MQTT forwarder #1164

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

		@@ -130,6 +136,22 @@ func main() {

		cc := thingsapi.NewClient(conn, thingsTracer, cfg.thingsAuthTimeout)

		ps, err := nats.NewPubSub(cfg.natsURL, "mqtt", logger)

MF-1079 - Add MQTT forwarder #1164

MF-1079 - Add MQTT forwarder #1164

Uh oh!

Conversation

Uh oh!

What does this do?

Which issue(s) does this PR fix/relate to?

List any changes that modify/break current functionality

Have you included tests for your changes?

Did you document any new/modified functionality?

Uh oh!

Uh oh!

Codecov Report

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!

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!

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!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!