fix(io): mqtt subscription concurrent map

internal/io/mqtt/client/client.go

@@ -17,6 +17,7 @@
 import (
 	"crypto/tls"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -41,7 +42,7 @@
 	scHandler api.StatusChangeHandler
 	conf      *ConnectionConfig
 	// key is the topic. Each topic will have only one connector
-	subscriptions map[string]*subscriptionInfo
+	subscriptions sync.Map
 }
 
 type ConnectionConfig struct {
@@ -61,7 +62,7 @@
 
 func CreateConnection(_ api.StreamContext) modules.Connection {
 	return &Connection{
-		subscriptions: make(map[string]*subscriptionInfo),
+		subscriptions: sync.Map{},
 	}
 }
 
@@ -129,12 +130,17 @@
 		conn.logger.Warnf("sc handler has not set yet")
 	}
 	conn.logger.Infof("The connection to mqtt broker is established")
-	for topic, info := range conn.subscriptions {
-		err := conn.Subscribe(topic, info.Qos, info.Handler)
-		if err != nil { // should never happen. If happens because of connection, it will retry later
-			conn.logger.Errorf("Failed to subscribe topic %s: %v", topic, err)
+	conn.subscriptions.Range(func(key, value any) bool {
+		topic, ok1 := key.(string)
+		info, ok2 := value.(*subscriptionInfo)
+		if ok1 && ok2 {
+			err := conn.Subscribe(topic, info.Qos, info.Handler)
+			if err != nil { // should never happen. If happens because of connection, it will retry later
+				conn.logger.Errorf("Failed to subscribe topic %s: %v", topic, err)
+			}
 		}
-	}
+		return true
+	})
 }
 
 func (conn *Connection) onConnectLost(_ pahoMqtt.Client, err error) {
@@ -159,7 +165,7 @@
 	if err != nil {
 		return
 	}
-	delete(conn.subscriptions, topic)
+	conn.subscriptions.Delete(topic)
 	conn.Client.Unsubscribe(topic)
 }
 
@@ -190,10 +196,10 @@
 }
 
 func (conn *Connection) Subscribe(topic string, qos byte, callback pahoMqtt.MessageHandler) error {
-	conn.subscriptions[topic] = &subscriptionInfo{
+	conn.subscriptions.Store(topic, &subscriptionInfo{
 		Qos:     qos,
 		Handler: callback,
-	}
+	})
 	token := conn.Client.Subscribe(topic, qos, callback)
 	return handleToken(token)
 }