MQTT: handle not spreading connections

rabbitmq · Sep 11, 2024 · 28bc164 · 28bc164
1 parent 4005ca6
commit 28bc164
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 11 deletions.
diff --git a/pkg/mqtt_client/consumer.go b/pkg/mqtt_client/consumer.go
@@ -74,12 +74,15 @@ func (c MqttConsumer) Start(ctx context.Context, subscribed chan bool) {
 		log.Info("consumer subscribed", "id", c.Id, "topic", c.Topic)
 	}
 
-	// TODO implement spreadConnections=false
-	for _, n := range utils.WrappedSequence(len(c.Config.ConsumerUri), c.Id-1) {
-		parsedUri := utils.ParseURI(c.Config.ConsumerUri[n], "mqtt", "1883")
-		opts.AddBroker(parsedUri.Broker).
-			SetUsername(parsedUri.Username).
-			SetPassword(parsedUri.Password)
+	var j int
+	for i, n := range utils.WrappedSequence(len(c.Config.ConsumerUri), c.Id-1) {
+		if c.Config.SpreadConnections {
+			j = n
+		} else {
+			j = i
+		}
+		parsedUri := utils.ParseURI(c.Config.ConsumerUri[j], "mqtt", "1883")
+		opts.AddBroker(parsedUri.Broker).SetUsername(parsedUri.Username).SetPassword(parsedUri.Password)
 	}
 
 	var token mqtt.Token

diff --git a/pkg/mqtt_client/publisher.go b/pkg/mqtt_client/publisher.go
@@ -36,11 +36,15 @@ func NewPublisher(cfg config.Config, id int) *MqttPublisher {
 		}).
 		SetProtocolVersion(4)
 
-	for _, n := range utils.WrappedSequence(len(cfg.PublisherUri), id-1) {
-		parsedUri := utils.ParseURI(cfg.PublisherUri[n], "mqtt", "1883")
-		opts.AddBroker(parsedUri.Broker).
-			SetUsername(parsedUri.Username).
-			SetPassword(parsedUri.Password)
+	var j int
+	for i, n := range utils.WrappedSequence(len(cfg.PublisherUri), id-1) {
+		if cfg.SpreadConnections {
+			j = n
+		} else {
+			j = i
+		}
+		parsedUri := utils.ParseURI(cfg.PublisherUri[j], "mqtt", "1883")
+		opts.AddBroker(parsedUri.Broker).SetUsername(parsedUri.Username).SetPassword(parsedUri.Password)
 	}
 
 	connection := mqtt.NewClient(opts)