Repository: lf-edge/ekuiper
Pull request #1353 (2022-08-09 09:44:36)

refactor(mqtt): default to MQTT 3.1.1(4) to connect to NanoMQ
also add debug log to paho if debug is enabled



Patch:
diff --git a/internal/topo/connection/clients/mqtt/mqtt.go b/internal/topo/connection/clients/mqtt/mqtt.go
index a3491d9c..7ebfeb40 100644
--- a/internal/topo/connection/clients/mqtt/mqtt.go
+++ b/internal/topo/connection/clients/mqtt/mqtt.go
@@ -73,10 +73,10 @@ func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
 	} else {
 		ms.clientid = cfg.ClientId
 	}
-
-	ms.pVersion = 3
-	if cfg.PVersion == "3.1.1" {
-		ms.pVersion = 4
+	// Default to MQTT 3.1.1 or NanoMQ cannot connect
+	ms.pVersion = 4
+	if cfg.PVersion == "3.1" {
+		ms.pVersion = 3
 	}
 
 	tlsOpts := cert.TlsConfigurationOptions{
@@ -98,8 +98,10 @@ func (ms *MQTTClient) CfgValidate(props map[string]interface{}) error {
 }
 
 func (ms *MQTTClient) Connect(connHandler MQTT.OnConnectHandler, lostHandler MQTT.ConnectionLostHandler) error {
-
-	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(ms.pVersion)
+	if conf.Config.Basic.Debug {
+		MQTT.DEBUG = conf.Log
+	}
+	opts := MQTT.NewClientOptions().AddBroker(ms.srv).SetProtocolVersion(4)
 
 	opts = opts.SetTLSConfig(ms.tls)
 
diff --git a/internal/topo/node/source_pool.go b/internal/topo/node/source_pool.go
index 15e65c50..c69a7831 100644
--- a/internal/topo/node/source_pool.go
+++ b/internal/topo/node/source_pool.go
@@ -24,6 +24,7 @@ import (
 	"github.com/lf-edge/ekuiper/pkg/api"
 	"github.com/lf-edge/ekuiper/pkg/infra"
 	"sync"
+	"time"
 )
 
 //// Package vars and funcs
@@ -273,15 +274,27 @@ func (ss *sourceSingleton) broadcastError(err error) {
 }
 
 func (ss *sourceSingleton) attach(instanceKey string, bl int) error {
-	ss.Lock()
-	defer ss.Unlock()
-	if _, ok := ss.outputs[instanceKey]; !ok {
-		ss.outputs[instanceKey] = newSourceInstanceChannels(bl)
-	} else {
-		// should not happen
-		return fmt.Errorf("fail to attach source instance, already has an output of the same key %s", instanceKey)
+	retry := 10
+	var err error
+	// retry multiple times in case the detach is still in progress
+	for i := 0; i < retry; i++ {
+		err = func() error {
+			ss.Lock()
+			defer ss.Unlock()
+			if _, ok := ss.outputs[instanceKey]; !ok {
+				ss.outputs[instanceKey] = newSourceInstanceChannels(bl)
+			} else {
+				// should not happen
+				return fmt.Errorf("fail to attach source instance, already has an output of the same key %s", instanceKey)
+			}
+			return nil
+		}()
+		if err == nil {
+			return nil
+		}
+		time.Sleep(time.Millisecond * 100)
 	}
-	return nil
+	return err
 }
 
 // detach Detach an instance and return if the singleton is ended


Test patch:
diff --git a/internal/topo/connection/clients/mqtt/client_mqtt_test.go b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
index 1899a494..ce73f2f3 100644
--- a/internal/topo/connection/clients/mqtt/client_mqtt_test.go
+++ b/internal/topo/connection/clients/mqtt/client_mqtt_test.go
@@ -131,7 +131,7 @@ func TestMQTTClient_CfgResult(t *testing.T) {
 	if !reflect.DeepEqual("clientid", ms.clientid) {
 		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.clientid)
 	}
-	if !reflect.DeepEqual(uint(3), ms.pVersion) {
+	if !reflect.DeepEqual(uint(4), ms.pVersion) {
 		t.Errorf("result mismatch:\n\n got=%#v\n\n", ms.pVersion)
 	}