Allow configuration of multiple MQTT brokers.

This contains the changes from #141 by @sa-wilson with some minor
modifications.

Closes #141.

Author: brocaar

cmd/chirpstack-gateway-bridge/cmd/configfile.go

@@ -220,8 +220,13 @@ marshaler="{{ .Integration.Marshaler }}"
 
     # Generic MQTT authentication.
     [integration.mqtt.auth.generic]
-    # MQTT server (e.g. scheme://host:port where scheme is tcp, ssl or ws)
-    server="{{ .Integration.MQTT.Auth.Generic.Server }}"
+    # MQTT servers.
+    #
+    # Configure one or multiple MQTT server to connect to. Each item must be in
+    # the following format: scheme://host:port where scheme is tcp, ssl or ws.
+    servers=[{{ range $index, $elm := .Integration.MQTT.Auth.Generic.Servers }}
+      "{{ $elm }}",{{ end }}
+    ]
 
     # Connect with the given username (optional)
     username="{{ .Integration.MQTT.Auth.Generic.Username }}"

cmd/chirpstack-gateway-bridge/cmd/root.go

@@ -57,7 +57,7 @@ func init() {
 	viper.SetDefault("integration.mqtt.command_topic_template", "gateway/{{ .GatewayID }}/command/#")
 	viper.SetDefault("integration.mqtt.max_reconnect_interval", 10*time.Minute)
 
-	viper.SetDefault("integration.mqtt.auth.generic.server", "tcp://127.0.0.1:1883")
+	viper.SetDefault("integration.mqtt.auth.generic.servers", []string{"tcp://127.0.0.1:1883"})
 	viper.SetDefault("integration.mqtt.auth.generic.clean_session", true)
 
 	viper.SetDefault("integration.mqtt.auth.gcp_cloud_iot_core.server", "ssl://mqtt.googleapis.com:8883")
@@ -127,6 +127,11 @@ func initConfig() {
 		config.C.Filters.NetIDs = config.C.Backend.BasicStation.Filters.NetIDs
 		config.C.Filters.JoinEUIs = config.C.Backend.BasicStation.Filters.JoinEUIs
 	}
+
+	// migrate server to servers
+	if config.C.Integration.MQTT.Auth.Generic.Server != "" {
+		config.C.Integration.MQTT.Auth.Generic.Servers = []string{config.C.Integration.MQTT.Auth.Generic.Server}
+	}
 }
 
 func viperBindEnvs(iface interface{}, parts ...string) {

docs/content/install/config.md

@@ -250,8 +250,13 @@ marshaler="protobuf"
 
     # Generic MQTT authentication.
     [integration.mqtt.auth.generic]
-    # MQTT server (e.g. scheme://host:port where scheme is tcp, ssl or ws)
-    server="tcp://127.0.0.1:1883"
+    # MQTT servers.
+    #
+    # Configure one or multiple MQTT server to connect to. Each item must be in
+    # the following format: scheme://host:port where scheme is tcp, ssl or ws.
+    servers=[
+      "tcp://127.0.0.1:1883",
+    ]
 
     # Connect with the given username (optional)
     username=""

internal/config/config.go

@@ -62,15 +62,16 @@ type Config struct {
 				Type string `mapstructure:"type"`
 
 				Generic struct {
-					Server       string `mapstructure:"server"`
-					Username     string `mapstructure:"username"`
-					Password     string `mapstrucure:"password"`
-					CACert       string `mapstructure:"ca_cert"`
-					TLSCert      string `mapstructure:"tls_cert"`
-					TLSKey       string `mapstructure:"tls_key"`
-					QOS          uint8  `mapstructure:"qos"`
-					CleanSession bool   `mapstructure:"clean_session"`
-					ClientID     string `mapstructure:"client_id"`
+					Server       string   `mapstructure:"server"`
+					Servers      []string `mapstructure:"servers"`
+					Username     string   `mapstructure:"username"`
+					Password     string   `mapstrucure:"password"`
+					CACert       string   `mapstructure:"ca_cert"`
+					TLSCert      string   `mapstructure:"tls_cert"`
+					TLSKey       string   `mapstructure:"tls_key"`
+					QOS          uint8    `mapstructure:"qos"`
+					CleanSession bool     `mapstructure:"clean_session"`
+					ClientID     string   `mapstructure:"client_id"`
 				} `mapstructure:"generic"`
 
 				GCPCloudIoTCore struct {

internal/integration/mqtt/auth/generic.go

@@ -12,7 +12,7 @@ import (
 
 // GenericAuthentication implements a generic MQTT authentication.
 type GenericAuthentication struct {
-	server       string
+	servers      []string
 	username     string
 	password     string
 	cleanSession bool
@@ -33,9 +33,8 @@ func NewGenericAuthentication(conf config.Config) (Authentication, error) {
 	}
 
 	return &GenericAuthentication{
-		tlsConfig: tlsConfig,
-
-		server:       conf.Integration.MQTT.Auth.Generic.Server,
+		tlsConfig:    tlsConfig,
+		servers:      conf.Integration.MQTT.Auth.Generic.Servers,
 		username:     conf.Integration.MQTT.Auth.Generic.Username,
 		password:     conf.Integration.MQTT.Auth.Generic.Password,
 		cleanSession: conf.Integration.MQTT.Auth.Generic.CleanSession,
@@ -45,7 +44,9 @@ func NewGenericAuthentication(conf config.Config) (Authentication, error) {
 
 // Init applies the initial configuration.
 func (a *GenericAuthentication) Init(opts *mqtt.ClientOptions) error {
-	opts.AddBroker(a.server)
+	for _, server := range a.servers {
+		opts.AddBroker(server)
+	}
 	opts.SetUsername(a.username)
 	opts.SetPassword(a.password)
 	opts.SetCleanSession(a.cleanSession)

internal/integration/mqtt/backend_test.go

@@ -57,7 +57,7 @@ func (ts *MQTTBackendTestSuite) SetupSuite() {
 	conf.Integration.MQTT.EventTopicTemplate = "gateway/{{ .GatewayID }}/event/{{ .EventType }}"
 	conf.Integration.MQTT.CommandTopicTemplate = "gateway/{{ .GatewayID }}/command/#"
 	conf.Integration.MQTT.Auth.Type = "generic"
-	conf.Integration.MQTT.Auth.Generic.Server = server
+	conf.Integration.MQTT.Auth.Generic.Servers = []string{server}
 	conf.Integration.MQTT.Auth.Generic.Username = username
 	conf.Integration.MQTT.Auth.Generic.Password = password
 	conf.Integration.MQTT.Auth.Generic.CleanSession = true