Better handling of concurrency in MQTT
This helps avoiding issues when multiple subscribers are added in a short period of time
@@ -202,9 +202,9 @@ func (m *mqttPubSub) Subscribe(req pubsub.SubscribeRequest, handler pubsub.Handl
defer m.topicsLock.Unlock()
// reset synchronization
- if m.consumer != nil {
+ if m.consumer != nil && m.consumer.IsConnectionOpen() {
m.logger.Infof("re-initializing the subscriber to add topic %s", req.Topic)
- m.consumer.Disconnect(0)
+ m.consumer.Disconnect(5)
m.consumer = nil
} else {
m.logger.Infof("initializing the subscriber with topic %s", req.Topic)
@@ -227,7 +227,6 @@ func (m *mqttPubSub) Subscribe(req pubsub.SubscribeRequest, handler pubsub.Handl
subscribeTopics[k] = m.metadata.qos
}
- go func() {
token := m.consumer.SubscribeMultiple(
subscribeTopics,
func(client mqtt.Client, mqttMsg mqtt.Message) {
@@ -270,10 +269,17 @@ func (m *mqttPubSub) Subscribe(req pubsub.SubscribeRequest, handler pubsub.Handl
}
},
)
+ subscribeCtx, subscribeCancel := context.WithTimeout(m.ctx, defaultWait)
+ defer subscribeCancel()
+ select {
+ case <-token.Done():
+ // Subscription went through
+ case <-subscribeCtx.Done():
+ return subscribeCtx.Err()
+ }
if err := token.Error(); err != nil {
- m.logger.Errorf("mqtt error from subscribe: %v", err)
+ return fmt.Errorf("mqtt error from subscribe: %v", err)
}
- }()
return nil
}
@@ -348,9 +354,9 @@ func (m *mqttPubSub) Close() error {
m.cancel()
if m.consumer != nil {
- m.consumer.Disconnect(0)
+ m.consumer.Disconnect(5)
}
- m.producer.Disconnect(0)
+ m.producer.Disconnect(5)
return nil
}
