update mq client check

mq/mq.go

@@ -131,7 +131,7 @@ func Keepalive(ctx context.Context) {
 		case <-ctx.Done():
 			return
 		case <-time.After(time.Second * KEEPALIVE_TIMEOUT):
-			if mqclient == nil {
+			if mqclient == nil || !mqclient.IsConnectionOpen() {
 				SetupMQTT(false)
 			}
 			serverStatusUpdate()

mq/publishers.go

@@ -339,7 +339,7 @@ func serverStatusUpdate() error {
 			return err
 		}
 
-		if mqclient == nil || !mqclient.IsConnected() {
+		if mqclient == nil || !mqclient.IsConnectionOpen() {
 			return errors.New("cannot publish ... mqclient not connected")
 		}
 
@@ -454,7 +454,7 @@ func PushMetricsToExporter(metrics models.Metrics) error {
 	if err != nil {
 		return errors.New("failed to marshal metrics: " + err.Error())
 	}
-	if mqclient == nil || !mqclient.IsConnected() {
+	if mqclient == nil || !mqclient.IsConnectionOpen() {
 		return errors.New("cannot publish ... mqclient not connected")
 	}
 	if token := mqclient.Publish("metrics_exporter", 0, true, data); !token.WaitTimeout(MQ_TIMEOUT*time.Second) || token.Error() != nil {

mq/util.go

@@ -79,7 +79,7 @@ func publish(host *models.Host, dest string, msg []byte) error {
 	if encryptErr != nil {
 		return encryptErr
 	}
-	if mqclient == nil || !mqclient.IsConnected() {
+	if mqclient == nil || !mqclient.IsConnectionOpen() {
 		return errors.New("cannot publish ... mqclient not connected")
 	}