* restart mqtt connection in mqttmonitor on connection loss

* adjusted mqttmock structure to be compatible

Signed-off-by: Jürgen Eckel <juergen@riddleandcode.com>

monitor/mqtt_monitor.go

@@ -153,6 +153,8 @@ func (mms *MqttMonitor) MqttMsgHandler(_ mqtt.Client, msg mqtt.Message) {
 
 func (mms *MqttMonitor) MonitorActiveParticipants() {
 	LazyLoadMonitorMQTTClient()
+	for {
+		if !MonitorMQTTClient.IsConnected() {
 			if token := MonitorMQTTClient.Connect(); token.Wait() && token.Error() != nil {
 				mms.Log("[Monitor] error connecting to mqtt: " + token.Error().Error())
 				panic(token.Error())
@@ -167,6 +169,9 @@ func (mms *MqttMonitor) MonitorActiveParticipants() {
 				panic(token.Error())
 			}
 		}
+		time.Sleep(30 * time.Second)
+	}
+}
 
 func (mms *MqttMonitor) Log(msg string) {
 	mms.contextMutex.Lock()

util/mocks/mqtt.go

@@ -14,6 +14,8 @@ type MockMQTTClient struct {
 	PublishFunc     func(topic string, qos byte, retained bool, payload interface{}) mqtt.Token
 	SubscribeFunc   func(topic string, qos byte, callback mqtt.MessageHandler) mqtt.Token
 	UnsubscribeFunc func(topics ...string) mqtt.Token
+	IsConnectedFunc func() bool
+	connected       bool
 }
 
 // GetConnectFunc fetches the mock client's `Connect` func
@@ -96,6 +98,7 @@ func GetUnsubscribeFunc(_ ...string) mqtt.Token {
 
 // Connect is the mock client's `Disconnect` func
 func (m *MockMQTTClient) Connect() mqtt.Token {
+	m.connected = true
 	return GetConnectFunc()
 }
 
@@ -116,3 +119,7 @@ func (m *MockMQTTClient) Subscribe(topic string, qos byte, callback mqtt.Message
 func (m *MockMQTTClient) Unsubscribe(topics ...string) mqtt.Token {
 	return GetUnsubscribeFunc(topics...)
 }
+
+func (m *MockMQTTClient) IsConnected() bool {
+	return m.connected
+}

util/mqtt.go

@@ -21,6 +21,7 @@ type MQTTClientI interface {
 	Publish(topic string, qos byte, retained bool, payload interface{}) mqtt.Token
 	Subscribe(topic string, qos byte, callback mqtt.MessageHandler) mqtt.Token
 	Unsubscribe(topics ...string) mqtt.Token
+	IsConnected() bool
 }
 
 var (