replace mqtt library "eclipse/paho" with "goiiot/libmqtt"

Author: rainu

actor/mqtt.go

@@ -2,12 +2,16 @@ package actor
 
 import (
 	"bytes"
-	mqtt "github.com/eclipse/paho.mqtt.golang"
+	"context"
 	"io"
 )
 
+type MQTTPublisher interface {
+	Publish(ctx context.Context, topic string, qos byte, retained bool, payload []byte) error
+}
+
 type Mqtt struct {
-	Client   mqtt.Client
+	Client   MQTTPublisher
 	Topic    string
 	QOS      byte
 	Retained bool
@@ -21,19 +25,5 @@ func (m *Mqtt) Do(ctx Context) error {
 		return err
 	}
 
-	token := m.Client.Publish(m.Topic, m.QOS, m.Retained, byteBuff.Bytes())
-
-	waitChan := make(chan error, 1)
-
-	go func() {
-		token.Wait()
-		waitChan <- token.Error()
-	}()
-
-	select {
-	case err := <-waitChan:
-		return err
-	case <-ctx.Context.Done():
-		return ctx.Context.Err()
-	}
+	return m.Client.Publish(ctx.Context, m.Topic, m.QOS, m.Retained, byteBuff.Bytes())
 }

cmd/lst/config/setup.go

@@ -1,6 +1,7 @@
 package config
 
 import (
+	"context"
 	"github.com/rainu/launchpad-super-trigger/config"
 	configActor "github.com/rainu/launchpad-super-trigger/config/actor"
 	connectionMqtt "github.com/rainu/launchpad-super-trigger/config/connection/mqtt"
@@ -28,8 +29,8 @@ func ConfigureDispatcher(configReader ...io.Reader) (*pad.TriggerDispatcher, map
 
 	//establish mqtt connections after callbacks were registered
 	for _, connection := range connections {
-		if token := connection.Connect(); token.Wait() && token.Error() != nil {
-			zap.L().Fatal("Error while connecting to mqtt broker: %s", zap.Error(token.Error()))
+		if err := connection.Connect(context.Background()); err != nil {
+			zap.L().Fatal("Error while connecting to mqtt broker: %s", zap.Error(err))
 		}
 	}
 

cmd/lst/main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	MQTT "github.com/eclipse/paho.mqtt.golang"
 	"github.com/rainu/launchpad-super-trigger/cmd/lst/config"
 	triggerConf "github.com/rainu/launchpad-super-trigger/config"
 	launchpad "github.com/rainu/launchpad-super-trigger/pad"
@@ -31,11 +30,6 @@ func main() {
 	zap.ReplaceGlobals(logger)
 	defer zap.L().Sync()
 
-	MQTT.ERROR, _ = zap.NewStdLogAt(zap.L(), zap.ErrorLevel)
-	MQTT.CRITICAL, _ = zap.NewStdLogAt(zap.L(), zap.ErrorLevel)
-	MQTT.WARN, _ = zap.NewStdLogAt(zap.L(), zap.WarnLevel)
-	MQTT.DEBUG, _ = zap.NewStdLogAt(zap.L(), zap.DebugLevel)
-
 	//start pprof if needed
 	if *Args.DebugPort > 0 {
 		zap.L().Info(fmt.Sprintf("Start pprof debug endpoint :%d", *Args.DebugPort))

config/connection/mqtt/builder.go

@@ -1,12 +1,16 @@
 package mqtt
 
 import (
-	MQTT "github.com/eclipse/paho.mqtt.golang"
+	"context"
+	MQTT "github.com/goiiot/libmqtt"
 	"github.com/rainu/launchpad-super-trigger/config"
+	"go.uber.org/zap"
 )
 
 type Client interface {
-	MQTT.Client
+	Connect(ctx context.Context) error
+	Publish(ctx context.Context, topic string, qos byte, retained bool, payload []byte) error
+	Subscribe(topic string, qos byte, clb func(string, byte, []byte))
 	AddListener(cl ConnectionListener)
 }
 
@@ -25,17 +29,18 @@ func buildMqttConnection(connection config.MQTTConnection) Client {
 		conObserver: connectionObserver{},
 	}
 
-	opts := MQTT.NewClientOptions()
-	opts.AddBroker(connection.Broker)
-
-	opts.SetAutoReconnect(true)
-	opts.SetClientID(connection.ClientId)
-	opts.SetUsername(connection.Username)
-	opts.SetPassword(connection.Password)
-	opts.SetOnConnectHandler(result.conObserver.OnConnectHandler())
-	opts.SetConnectionLostHandler(result.conObserver.ConnectionLostHandler())
+	client, err := MQTT.NewClient(
+		MQTT.WithClientID(connection.ClientId),
+		MQTT.WithIdentity(connection.Username, connection.Password),
+		MQTT.WithCleanSession(true),
+		MQTT.WithAutoReconnect(true),
+	)
+	if err != nil {
+		zap.L().Fatal("Error build connecting to mqtt broker: %s", zap.Error(err))
+	}
 
-	result.client = MQTT.NewClient(opts)
+	result.broker = connection.Broker
+	result.client = client
 	result.conObserver.AddListener(&logConnectionListener{}) //for logging purposes
 
 	return result

config/connection/mqtt/client.go

@@ -1,48 +1,69 @@
 package mqtt
 
-import MQTT "github.com/eclipse/paho.mqtt.golang"
+import (
+	"context"
+	MQTT "github.com/goiiot/libmqtt"
+)
 
 type mqttClient struct {
+	broker      string
 	client      MQTT.Client
 	conObserver connectionObserver
 }
 
-func (m *mqttClient) IsConnected() bool {
-	return m.client.IsConnected()
+type connectionListener struct {
+	errChan chan error
 }
 
-func (m *mqttClient) IsConnectionOpen() bool {
-	return m.client.IsConnectionOpen()
+func (c *connectionListener) OnConnect(client MQTT.Client) {
+	c.errChan <- nil
 }
 
-func (m *mqttClient) Connect() MQTT.Token {
-	return m.client.Connect()
+func (c *connectionListener) OnConnectionLost(client MQTT.Client, err error) {
+	c.errChan <- err
 }
 
-func (m *mqttClient) Disconnect(quiesce uint) {
-	m.client.Disconnect(quiesce)
-}
+func (m *mqttClient) Connect(ctx context.Context) error {
+	errChan := make(chan error, 1)
+	defer close(errChan)
 
-func (m *mqttClient) Publish(topic string, qos byte, retained bool, payload interface{}) MQTT.Token {
-	return m.client.Publish(topic, qos, retained, payload)
-}
+	listener := &connectionListener{errChan}
+	defer func() {
+		//we can remove our listener -> the only purpose was to wait for connection!
+		m.conObserver.RemoveListener(listener)
+	}()
 
-func (m *mqttClient) Subscribe(topic string, qos byte, callback MQTT.MessageHandler) MQTT.Token {
-	return m.client.Subscribe(topic, qos, callback)
-}
-
-func (m *mqttClient) SubscribeMultiple(filters map[string]byte, callback MQTT.MessageHandler) MQTT.Token {
-	return m.client.SubscribeMultiple(filters, callback)
-}
+	m.conObserver.AddListener(listener)
+	err := m.client.ConnectServer(m.broker, MQTT.WithConnHandleFunc(m.conObserver.OnConnectHandler()))
+	if err != nil {
+		return err
+	}
 
-func (m *mqttClient) Unsubscribe(topics ...string) MQTT.Token {
-	return m.client.Unsubscribe(topics...)
+	//wait until connection established
+	select {
+	case err := <-errChan:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
 }
 
-func (m *mqttClient) AddRoute(topic string, callback MQTT.MessageHandler) {
-	m.client.AddRoute(topic, callback)
+func (m *mqttClient) Subscribe(topic string, qos byte, clb func(string, byte, []byte)) {
+	m.client.HandleTopic(topic, func(client MQTT.Client, topic string, qos MQTT.QosLevel, msg []byte) {
+		clb(topic, qos, msg)
+	})
+	m.client.Subscribe(&MQTT.Topic{
+		Name: topic,
+		Qos:  qos,
+	})
 }
 
-func (m *mqttClient) OptionsReader() MQTT.ClientOptionsReader {
-	return m.client.OptionsReader()
+func (m *mqttClient) Publish(ctx context.Context, topic string, qos byte, retained bool, payload []byte) error {
+	m.client.Publish(&MQTT.PublishPacket{
+		TopicName: topic,
+		Qos:       qos,
+		IsRetain:  retained,
+		Payload:   payload,
+	})
+	return nil
 }

config/connection/mqtt/log.go

@@ -1,7 +1,7 @@
 package mqtt
 
 import (
-	MQTT "github.com/eclipse/paho.mqtt.golang"
+	MQTT "github.com/goiiot/libmqtt"
 	"go.uber.org/zap"
 )
 

config/connection/mqtt/observer.go

@@ -1,7 +1,7 @@
 package mqtt
 
 import (
-	MQTT "github.com/eclipse/paho.mqtt.golang"
+	MQTT "github.com/goiiot/libmqtt"
 	"sync"
 )
 
@@ -17,38 +17,41 @@ type ConnectionListener interface {
 type connectionObserver struct {
 	listenerMutex sync.RWMutex
 	eventMutex    sync.Mutex
-	listener      []ConnectionListener
+	listener      map[interface{}]ConnectionListener
 }
 
 func (c *connectionObserver) AddListener(cl ConnectionListener) {
 	c.listenerMutex.Lock()
 	defer c.listenerMutex.Unlock()
 
-	c.listener = append(c.listener, cl)
-}
+	if c.listener == nil {
+		c.listener = map[interface{}]ConnectionListener{}
+	}
 
-func (c *connectionObserver) OnConnectHandler() MQTT.OnConnectHandler {
-	return func(client MQTT.Client) {
-		c.listenerMutex.RLock()
-		c.eventMutex.Lock()
-		defer c.listenerMutex.RUnlock()
-		defer c.eventMutex.Unlock()
+	c.listener[cl] = cl
+}
+func (c *connectionObserver) RemoveListener(cl ConnectionListener) {
+	c.listenerMutex.Lock()
+	defer c.listenerMutex.Unlock()
 
-		for _, listener := range c.listener {
-			listener.OnConnect(client)
-		}
-	}
+	delete(c.listener, cl)
 }
 
-func (c *connectionObserver) ConnectionLostHandler() MQTT.ConnectionLostHandler {
-	return func(client MQTT.Client, err error) {
+func (c *connectionObserver) OnConnectHandler() MQTT.ConnHandleFunc {
+	return func(client MQTT.Client, server string, code byte, err error) {
 		c.listenerMutex.RLock()
 		c.eventMutex.Lock()
 		defer c.listenerMutex.RUnlock()
 		defer c.eventMutex.Unlock()
 
-		for _, listener := range c.listener {
-			listener.OnConnectionLost(client, err)
+		if code == MQTT.CodeSuccess {
+			for _, listener := range c.listener {
+				listener.OnConnect(client)
+			}
+		} else {
+			for _, listener := range c.listener {
+				listener.OnConnectionLost(client, err)
+			}
 		}
 	}
 }

config/sensor/mqtt.go

@@ -2,7 +2,7 @@ package sensor
 
 import (
 	"fmt"
-	MQTT "github.com/eclipse/paho.mqtt.golang"
+	MQTT "github.com/goiiot/libmqtt"
 	"github.com/rainu/launchpad-super-trigger/config"
 	"github.com/rainu/launchpad-super-trigger/config/connection/mqtt"
 	"github.com/rainu/launchpad-super-trigger/sensor"

go.mod

@@ -4,8 +4,8 @@ go 1.14
 
 require (
 	github.com/boltdb/bolt v1.3.1
-	github.com/eclipse/paho.mqtt.golang v1.2.0
 	github.com/go-playground/universal-translator v0.17.0 // indirect
+	github.com/goiiot/libmqtt v0.9.5
 	github.com/itchyny/gojq v0.11.0
 	github.com/leodido/go-urn v1.2.0 // indirect
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect