Change token.Wait to token.WaitTimeout.

Author: brocaar

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

@@ -62,6 +62,7 @@ func init() {
 	viper.SetDefault("integration.mqtt.state_retained", true)
 	viper.SetDefault("integration.mqtt.keep_alive", 30*time.Second)
 	viper.SetDefault("integration.mqtt.max_reconnect_interval", time.Minute)
+	viper.SetDefault("integration.mqtt.max_token_wait", time.Second)
 
 	viper.SetDefault("integration.mqtt.auth.generic.servers", []string{"tcp://127.0.0.1:1883"})
 	viper.SetDefault("integration.mqtt.auth.generic.clean_session", true)

internal/config/config.go

@@ -63,6 +63,7 @@ type Config struct {
 			KeepAlive               time.Duration `mapstructure:"keep_alive"`
 			MaxReconnectInterval    time.Duration `mapstructure:"max_reconnect_interval"`
 			TerminateOnConnectError bool          `mapstructure:"terminate_on_connect_error"`
+			MaxTokenWait            time.Duration `mapstructure:"max_token_wait"`
 
 			Auth struct {
 				Type string `mapstructure:"type"`

internal/integration/mqtt/backend.go

@@ -41,6 +41,7 @@ type Backend struct {
 	gatewaysSubscribed      map[lorawan.EUI64]struct{}
 	terminateOnConnectError bool
 	stateRetained           bool
+	maxTokenWait            time.Duration
 
 	qos                  uint8
 	eventTopicTemplate   *template.Template
@@ -62,6 +63,7 @@ func NewBackend(conf config.Config) (*Backend, error) {
 		gateways:                make(map[lorawan.EUI64]struct{}),
 		gatewaysSubscribed:      make(map[lorawan.EUI64]struct{}),
 		stateRetained:           conf.Integration.MQTT.StateRetained,
+		maxTokenWait:            conf.Integration.MQTT.MaxTokenWait,
 	}
 
 	switch conf.Integration.MQTT.Auth.Type {
@@ -283,7 +285,7 @@ func (b *Backend) subscribeGateway(gatewayID lorawan.EUI64) error {
 		"qos":   b.qos,
 	}).Info("integration/mqtt: subscribing to topic")
 
-	if token := b.conn.Subscribe(topic.String(), b.qos, b.handleCommand); token.Wait() && token.Error() != nil {
+	if token := b.conn.Subscribe(topic.String(), b.qos, b.handleCommand); token.WaitTimeout(b.maxTokenWait) && token.Error() != nil {
 		return errors.Wrap(token.Error(), "subscribe topic error")
 	}
 	return nil
@@ -298,7 +300,7 @@ func (b *Backend) unsubscribeGateway(gatewayID lorawan.EUI64) error {
 		"topic": topic.String(),
 	}).Info("integration/mqtt: unsubscribing from topic")
 
-	if token := b.conn.Unsubscribe(topic.String()); token.Wait() && token.Error() != nil {
+	if token := b.conn.Unsubscribe(topic.String()); token.WaitTimeout(b.maxTokenWait) && token.Error() != nil {
 		return errors.Wrap(token.Error(), "unsubscribe topic error")
 	}
 
@@ -351,7 +353,7 @@ func (b *Backend) PublishState(gatewayID lorawan.EUI64, state string, v proto.Me
 		"state":      state,
 		"gateway_id": gatewayID,
 	}).Info("integration/mqtt: publishing state")
-	if token := b.conn.Publish(topic.String(), b.qos, b.stateRetained, bytes); token.Wait() && token.Error() != nil {
+	if token := b.conn.Publish(topic.String(), b.qos, b.stateRetained, bytes); token.WaitTimeout(b.maxTokenWait) && token.Error() != nil {
 		return token.Error()
 	}
 	return nil
@@ -366,7 +368,7 @@ func (b *Backend) connect() error {
 	}
 
 	b.conn = paho.NewClient(b.clientOpts)
-	if token := b.conn.Connect(); token.Wait() && token.Error() != nil {
+	if token := b.conn.Connect(); token.WaitTimeout(b.maxTokenWait) && token.Error() != nil {
 		return token.Error()
 	}
 
@@ -659,7 +661,7 @@ func (b *Backend) publishEvent(gatewayID lorawan.EUI64, event string, fields log
 	fields["event"] = event
 
 	log.WithFields(fields).Info("integration/mqtt: publishing event")
-	if token := b.conn.Publish(topic.String(), b.qos, false, bytes); token.Wait() && token.Error() != nil {
+	if token := b.conn.Publish(topic.String(), b.qos, false, bytes); token.WaitTimeout(b.maxTokenWait) && token.Error() != nil {
 		return token.Error()
 	}
 	return nil

internal/integration/mqtt/backend_test.go

@@ -64,6 +64,7 @@ func (ts *MQTTBackendTestSuite) SetupSuite() {
 	conf.Integration.MQTT.Auth.Generic.Password = password
 	conf.Integration.MQTT.Auth.Generic.CleanSession = true
 	conf.Integration.MQTT.Auth.Generic.ClientID = ts.gatewayID.String()
+	conf.Integration.MQTT.MaxTokenWait = time.Second
 
 	var err error
 	ts.backend, err = NewBackend(conf)