test(mqtt): fix mqtt client connect error

config.default.yaml

@@ -35,6 +35,15 @@ connector:
   mqttBroker:
     tcpPort: 1883
     wsPort: 8083
+    # mqtt storage, if left blank, the storage is disabled
+    # https://github.com/mochi-mqtt/server?tab=readme-ov-file#badger-db
+    storage: 
+      type: file  # file or redis
+      path: "./embed-mqtt.db" # when type is file
+      redis:
+        addr: 127.0.0.1:6379
+        db: 1
+        password:
     superUsers:
       - name: $tio
         password: public

connector/mqtt/client/client.go

@@ -51,7 +51,7 @@ func NewClient(cfg config.MqttClientConfig) Client {
 	opts.SetClientID(cfg.ClientId)
 	opts.SetUsername(cfg.User)
 	opts.SetPassword(cfg.Password)
-	cleanSession := false
+	cleanSession := true
 	if cfg.CleanSession != nil {
 		cleanSession = *cfg.CleanSession
 	}

connector/mqtt/embed/embed_broker_test.go

@@ -3,10 +3,11 @@ package embed_test
 import (
 	"context"
 	"encoding/json"
-	"ruff.io/tio/connector"
 	"testing"
 	"time"
 
+	"ruff.io/tio/connector"
+
 	"ruff.io/tio/connector/mqtt/client"
 	"ruff.io/tio/connector/mqtt/embed"
 
@@ -25,7 +26,7 @@ func TestEmbedBrokerConnectivity(t *testing.T) {
 	port := 21883
 	brk := embed.InitBroker(embed.MochiConfig{
 		TcpPort: port,
-		AuthzFn: func(user, password string) bool {
+		AuthzFn: func(embed.ConnectParams) bool {
 			return true
 		},
 		AclFn: func(user string, topic string, write bool) bool {

integration_tests/setup_test.go

@@ -6,9 +6,10 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
-	connector2 "ruff.io/tio/connector"
 	"testing"
 
+	connector2 "ruff.io/tio/connector"
+
 	"ruff.io/tio/connector/mqtt/embed"
 
 	"ruff.io/tio/connector/mqtt/client"