MQTT: handle subscription errors

Author: mkuratczyk

go.sum

@@ -219,14 +219,10 @@ github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+W
 github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
-github.com/onsi/ginkgo/v2 v2.28.1 h1:S4hj+HbZp40fNKuLUQOYLDgZLwNUVn19N3Atb98NCyI=
-github.com/onsi/ginkgo/v2 v2.28.1/go.mod h1:CLtbVInNckU3/+gC8LzkGUb9oF+e8W8TdUsxPwvdOgE=
 github.com/onsi/ginkgo/v2 v2.28.3 h1:4JvMdwtFU0imd8fHx25OJXoDMRexnf8v5NHKYSTTji4=
 github.com/onsi/ginkgo/v2 v2.28.3/go.mod h1:+aXOY+vzZ5mu2iI2HpTZUPmM//oQfsNFX6gU9kNcA44=
 github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
-github.com/onsi/gomega v1.39.1 h1:1IJLAad4zjPn2PsnhH70V4DKRFlrCzGBNrNaru+Vf28=
-github.com/onsi/gomega v1.39.1/go.mod h1:hL6yVALoTOxeWudERyfppUcZXjMwIMLnuSfruD2lcfg=
 github.com/onsi/gomega v1.40.0 h1:Vtol0e1MghCD2ZVIilPDIg44XSL9l2QAn8ZNaljWcJc=
 github.com/onsi/gomega v1.40.0/go.mod h1:M/Uqpu/8qTjtzCLUA2zJHX9Iilrau25x1PdoSRbWh5A=
 github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0=

pkg/mqtt/consumer.go

@@ -74,7 +74,19 @@ func (c MqttConsumer) Start(cosumerReady chan bool) {
 			token := client.Subscribe(topic, byte(c.Config.MqttConsumer.QoS), handler)
 			token.Wait()
 			if token.Error() != nil {
-				log.Error("failed to subscribe", "id", c.Id, "error", token.Error())
+				log.Error("failed to subscribe, reconnecting", "id", c.Id, "topic", topic, "error", token.Error())
+				go func() {
+					select {
+					case <-c.ctx.Done():
+						return
+					case <-time.After(config.ReconnectDelay):
+					}
+					client.Disconnect(250)
+					if t := client.Connect(); t.Wait() && t.Error() != nil {
+						log.Error("consumer reconnect failed", "id", c.Id, "error", t.Error())
+					}
+				}()
+				return
 			}
 			log.Info("consumer subscribed", "id", c.Id, "topic", topic)
 		}

pkg/mqtt/consumer_v5.go

@@ -106,12 +106,34 @@ func (c Mqtt5Consumer) Start(consumerReady chan bool) {
 					Topic: topic,
 					QoS:   byte(c.Config.MqttConsumer.QoS),
 				})
-				log.Info("consumer subscribing", "id", c.Id, "topic", topic)
 			}
-			if _, err := cm.Subscribe(context.Background(), &paho.Subscribe{
+			if _, err := cm.Subscribe(c.ctx, &paho.Subscribe{
 				Subscriptions: subscriptions,
 			}); err != nil {
-				fmt.Printf("failed to subscribe (%s). This is likely to mean no messages will be received.", err)
+				log.Error("failed to subscribe, retrying", "id", c.Id, "error", err)
+				go func() {
+					for {
+						select {
+						case <-c.ctx.Done():
+							return
+						case <-time.After(config.ReconnectDelay):
+						}
+						if _, retryErr := cm.Subscribe(c.ctx, &paho.Subscribe{
+							Subscriptions: subscriptions,
+						}); retryErr == nil {
+							for _, sub := range subscriptions {
+								log.Info("consumer subscribed", "id", c.Id, "topic", sub.Topic)
+							}
+							return
+						} else {
+							log.Error("failed to subscribe, retrying", "id", c.Id, "error", retryErr)
+						}
+					}
+				}()
+			} else {
+				for _, sub := range subscriptions {
+					log.Info("consumer subscribed", "id", c.Id, "topic", sub.Topic)
+				}
 			}
 		},
 		OnConnectError: func(err error) {