ESPIoTBase/src/mqtt.ino

/*#include <Arduino.h>*/

#define MQTT_RECONNECT_INTERVAL_3_TRIES 15000
#define MQTT_RECONNECT_INTERVAL_10_TRIES 60000
#define MQTT_RECONNECT_INTERVAL_LONG 300000

bool mqtt_heartbeatReceived = false;

// MQTT callback
void mqttCallback(char *topic, unsigned char *payload, uint16_t length)
{
  //  Serial.print("MQTT payload arrived [");
  //  Serial.print(topic);
  //  Serial.print("] ");
  //  for (int i = 0; i < length; i++) {
  //    Serial.print((char)payload[i]);
  //  }
  //  Serial.println();

  //char tmp_topic_pub[51];
  //char tmp_payload_pub[51];

  if (strcmp(topic, mqtt_topic_in_cmd) == 0)
  { //if topic = mqtt_topic_in_cmd
    int _len;
    if (length < sizeof(cmdPayload))
      _len = length; // if input is bigger than dest buffer, cut
    else
      _len = sizeof(cmdPayload) - 1;

    for (int i = 0; i < _len; i++)
    {
      cmdPayload[i] = (char)payload[i];
    }

    cmdPayload[_len] = '\0';
    //    Serial.print("cmdPayload:");
    //    Serial.println(cmdPayload);
    //if (serialdebug)
    //{
    //  Serial.print("MQTT: received '");
    //  Serial.print(mqtt_topic_in_cmd);
    //  Serial.print("' - '");
    //  Serial.print(cmdPayload);
    //  Serial.println("'");
    //}
    //if (mqttdebug)
    //{
    //  sprintf(tmp_topic_pub, "%s/%s", confMqtt.mqtt_topic_out, "mqtt_received");
    //  sprintf(tmp_payload_pub, "%s: %s", mqtt_topic_in_cmd, cmdPayload);
    //  mqttclient.publish(tmp_topic_pub, tmp_payload_pub);
    //}

      if (strncmp(cmdPayload, "HEARTBEAT", 9) == 0)
      {
        mqtt_lastHeartbeat = millis();
        mqtt_heartbeatReceived = true;
        mqtt_connected = true;
      }
      else
      {
        cmdInQueue = true; // payload is processed in "commands"
      }
    } //if topic = mqtt_topic_in_cmd


  #ifdef FIRMWARE_VARIANT_THERMOSTAT
  if (strcmp(topic, mqtt_topic_in_setTemp) == 0) {
    int len;
    if (length < sizeof(cmdPayload))
      len = length; // if input is bigger than dest buffer, cut
    else
      len = sizeof(cmdPayload) - 1;

    for (int i = 0; i < len; i++)
    {
      cmdPayload[i] = (char)payload[i];
    }
    //cmdPayload[len + 1] = '\0';
    cmdPayload[len] = '\0';
    //    Serial.print("cmdPayload:");
    //    Serial.println(cmdPayload);

    char tmpPayload[11];
    strlcpy(tmpPayload, cmdPayload, 10);

    float valueFloat;
    valueFloat = round(atof(tmpPayload) * 2.0) / 2.0;
    thermostat_setTempTo(valueFloat);

    //if (serialdebug)
    //{
    //  Serial.print("MQTT: received '");
    //  Serial.print(mqtt_topic_in_setTemp);
    //  Serial.print("' - '");
    //  Serial.print(cmdPayload);
    //  Serial.println("'");
    //}
    //if (mqttdebug)
    //{
    //  sprintf(tmp_topic_pub, "%s/%s", confMqtt.mqtt_topic_out, "mqtt_received");
    //  sprintf(tmp_payload_pub, "%s: %s", mqtt_topic_in_setTemp, cmdPayload);
    //  mqttclient.publish(tmp_topic_pub, tmp_payload_pub);
    //}

    char buf[50];
    sprintf(buf, "MQTT IN: setTemp to %2.1f", valueFloat);
    sendLog(buf, LOGLEVEL_INFO);

    cmdInQueue = false; // payload is processed in "commands"
  }                     //if topic = mqtt_topic_in_setTemp

  if (strcmp(topic, mqtt_topic_in_setMode) == 0)
  { //if topic = mqtt_topic_in_setMode
    uint8_t len;

    if (length < sizeof(cmdPayload))
      len = length; // if input is bigger than dest buffer, cut
    else
      len = sizeof(cmdPayload) - 1;

    for (int i = 0; i < len; i++)
    {
      cmdPayload[i] = (char)payload[i];
    }
    //cmdPayload[len + 1] = '\0';
    cmdPayload[len] = '\0';
    //    Serial.print("cmdPayload:");
    //    Serial.println(cmdPayload);

    char tmpPayload[21];
    strlcpy(tmpPayload, cmdPayload, sizeof(tmpPayload));
    strlwr(tmpPayload);

    unsigned char tmpHeatMode = 100;

    char tmpModename0[15];
    char tmpModename1[15];
    strlcpy(tmpModename0, confTherm.modeName0, sizeof(tmpModename0));
    strlcpy(tmpModename1, confTherm.modeName1, sizeof(tmpModename1));
    strlwr(tmpModename0);
    strlwr(tmpModename1);

    if (strcmp(tmpPayload, tmpModename0) == 0)
      tmpHeatMode = 0;
    else if (strcmp(tmpPayload, tmpModename1) == 0)
      tmpHeatMode = 1;
    else if (strcmp(tmpPayload, "off") == 0)
      tmpHeatMode = 0;
    else if (strcmp(tmpPayload, "aus") == 0)
      tmpHeatMode = 0;
    else if (strcmp(tmpPayload, "on") == 0)
      tmpHeatMode = 1;
    else if (strcmp(tmpPayload, "ein") == 0)
      tmpHeatMode = 1;
    else if (strcmp(tmpPayload, "heat") == 0)
      tmpHeatMode = 1;
    else if (atoi(tmpPayload) == 0)
      tmpHeatMode = 0;
    else if (atoi(tmpPayload) == 1)
      tmpHeatMode = 1;

    if (tmpHeatMode == 0 || tmpHeatMode == 1)
    {
      thermostat_setHeatingmodeTo(tmpHeatMode);
      //if (serialdebug)
      //{
      //  Serial.print(F("set heatmode to: "));
      //  Serial.println(tmpHeatMode);
      //}
      //if (serialdebug)
      //{
      //  Serial.print("MQTT: received '");
      //  Serial.print(mqtt_topic_in_setMode);
      //  Serial.print("' - '");
      //  Serial.print(cmdPayload);
      //  Serial.println("'");
      //}
      //if (mqttdebug)
      //{
      //  sprintf(tmp_topic_pub, "%s/%s", confMqtt.mqtt_topic_out, "mqtt_received");
      //  sprintf(tmp_payload_pub, "%s: %s", mqtt_topic_in_setMode, cmdPayload);
      //  mqttclient.publish(tmp_topic_pub, tmp_payload_pub);
      //}
      char buf[50];
      sprintf(buf, "MQTT IN: setMode to %u", tmpHeatMode);
      sendLog(buf, LOGLEVEL_INFO);
    }

    cmdInQueue = false; // payload is processed in "commands"
  }                     //if topic = mqtt_topic_in_setMode

  if (strcmp(topic, mqtt_topic_in_setPause) == 0)
  { //if topic = mqtt_topic_in_setMode
    uint8_t len;
    uint8_t _setTo = 255;

    if (length < sizeof(cmdPayload))
      len = length; // if input is bigger than dest buffer, cut
    else
      len = sizeof(cmdPayload) - 1;

    for (int i = 0; i < len; i++)
    {
      cmdPayload[i] = (char)payload[i];
    }
    //cmdPayload[len + 1] = '\0';
    cmdPayload[len] = '\0';
    //    Serial.print("cmdPayload:");
    //    Serial.println(cmdPayload);

    char tmpPayload[21];
    strlcpy(tmpPayload, cmdPayload, sizeof(tmpPayload));
    strlwr(tmpPayload);

    
    if (strcmp(tmpPayload, "1") == 0) _setTo = 1;
    else if (strcmp(tmpPayload, "0") == 0) _setTo = 0;

    if(_setTo == 1 || _setTo == 0) {
      thermostat_setHeatingPause(_setTo);
      char buf[50];
      sprintf(buf, "MQTT IN: setPause to %u", _setTo);
      sendLog(buf, LOGLEVEL_INFO);
    }

    cmdInQueue = false; // payload is processed in "commands"
  }                     //if topic = mqtt_topic_in_setPause

  if (strcmp(topic, mqtt_topic_in_setPreset) == 0)
  { //if topic = mqtt_topic_in_setPreset
    char tmpPayload[21];
    char tmpPresetName0[15];
    char tmpPresetName1[15];
    char tmpPresetName2[15];
    uint16_t len;
    unsigned char tmpPreset = 100;

    if (length < sizeof(cmdPayload))
      len = length; // if input is bigger than dest buffer, cut
    else
      len = sizeof(cmdPayload) - 1;

    for (unsigned char i = 0; i < len; i++)
    {
      cmdPayload[i] = (char)payload[i];
    }
    //cmdPayload[len + 1] = '\0';
    cmdPayload[len] = '\0';
    //    Serial.print("cmdPayload:");
    //    Serial.println(cmdPayload);

    strlcpy(tmpPayload, cmdPayload, sizeof(tmpPayload));
    strlwr(tmpPayload);

    strlcpy(tmpPresetName0, confTherm.psetName0, sizeof(tmpPresetName0));
    strlcpy(tmpPresetName1, confTherm.psetName1, sizeof(tmpPresetName1));
    strlcpy(tmpPresetName2, confTherm.psetName2, sizeof(tmpPresetName2));
    strlwr(tmpPresetName0);
    strlwr(tmpPresetName1);
    strlwr(tmpPresetName2);

    if (strcmp(tmpPayload, tmpPresetName0) == 0)
      tmpPreset = 0;
    else if (strcmp(tmpPayload, tmpPresetName1) == 0)
      tmpPreset = 1;
    else if (strcmp(tmpPayload, tmpPresetName2) == 0)
      tmpPreset = 2;
    else if (strcmp(tmpPayload, "none") == 0)
      tmpPreset = 0;
    else if (strcmp(tmpPayload, "norm") == 0)
      tmpPreset = 0;
    else if (strcmp(tmpPayload, "red1") == 0)
      tmpPreset = 1;
    else if (strcmp(tmpPayload, "red2") == 0)
      tmpPreset = 2;
    else if (atoi(tmpPayload) == 0)
      tmpPreset = 0;
    else if (atoi(tmpPayload) == 1)
      tmpPreset = 1;
    else if (atoi(tmpPayload) == 2)
      tmpPreset = 2;

    if (tmpPreset <= 2)
    {
      thermostat_setPresetTo(tmpPreset);
      //if (serialdebug)
      //{
      //  Serial.print(F("set preset to: "));
      //  Serial.println(tmpPreset);
      //}
      //if (serialdebug)
      //{
      //  Serial.print(F("MQTT: received '"));
      //  Serial.print(mqtt_topic_in_setPreset);
      //  Serial.print("' - '");
      //  Serial.print(cmdPayload);
      //  Serial.println("'");
      //}
      //if (mqttdebug)
      //{
      //  sprintf(tmp_topic_pub, "%s/%s", confMqtt.mqtt_topic_out, "mqtt_received");
      //  sprintf(tmp_payload_pub, "%s: %s", mqtt_topic_in_setPreset, cmdPayload);
      //  mqttclient.publish(tmp_topic_pub, tmp_payload_pub);
      //}
      char buf[50];
      sprintf(buf, "MQTT IN: setPreset to %s", cmdPayload);
      sendLog(buf, LOGLEVEL_INFO);
    }
    cmdInQueue = false; // payload is processed in "commands"
  }                     //if topic = mqtt_topic_in_setPreset
  #endif // Thermostat

  // outside temp/hum via MQTT
  if (strcmp(topic, confAdd.outTemp_topic_in) == 0) {
    int len;
    if (length < 6)
      len = length; // if input is bigger than dest buffer, cut
    else
      len = 5;
    for (int i = 0; i < len; i++)
    {
      outTemp_newValue[i] = (char)payload[i];
    }
    //outTemp_newValue[len + 1] = '\0';
    outTemp_newValue[len] = '\0';
    outTemp_parseNewValue = true;
  } //if topic = outTemp_topic_in

  if (strcmp(topic, confAdd.outHum_topic_in) == 0) {
    int len;
    if (length < 4)
      len = length; // if input is bigger than dest buffer, cut
    else
      len = 3;
    for (int i = 0; i < len; i++)
    {
      outHum_newValue[i] = (char)payload[i];
    }
    //outHum_newValue[len + 1] = '\0';
    outHum_newValue[len] = '\0';
    outHum_parseNewValue = true;
  } //if topic = outHum_topic_in

} //mqttCallback

void mqttPrepareSubscribeTopics()
{
  //char tmp_topic_out[50];
  sprintf(mqtt_topic_in_cmd, "%s/%s", confMqtt.mqtt_topic_in, "cmd");
  
  #ifdef FIRMWARE_VARIANT_THERMOSTAT
  sprintf(mqtt_topic_in_setTemp, "%s/%s", mqtt_topic_in_cmd, "setTemp");
  sprintf(mqtt_topic_in_setMode, "%s/%s", mqtt_topic_in_cmd, "setMode");
  sprintf(mqtt_topic_in_setPreset, "%s/%s", mqtt_topic_in_cmd, "setPreset");
  sprintf(mqtt_topic_in_setPause, "%s/%s", mqtt_topic_in_cmd, "setPause");
  #endif
}

bool mqttReconnect(void)
{
  mqttclient.disconnect();
  delay(10);
  // Create MQTT client ID from device name
  //String mqttClientId = "ESP8266Client-";
  String mqttClientId = FIRMWARE_SHORTNAME;
  mqttClientId += "-";
  //mqttClientId += String(random(0xffff), HEX);   //or random
  mqttClientId += String(confDevWiFi.deviceName);

  //if (serialdebug)
  //  Serial.print(F("MQTT: connecting to broker '"));
  //  Serial.print(confMqtt.mqtt_server);
  //  Serial.print(F("' with "));

  boolean connRes;
  mqttReconnectAttempts++;
  char connectMode[20];

  if (strlen(confSecrets.mqtt_user) > 0 && strlen(confMqtt.mqtt_willTopic) == 0)
  {
    // user and password, no Last Will
    //if (serialdebug)
    //  Serial.print(F("Auth, no LWT"));
    strlcpy(connectMode, "Auth, no LWT", sizeof(connectMode));
    connRes = mqttclient.connect(mqttClientId.c_str(), confSecrets.mqtt_user, confSecrets.mqtt_pass);
  }
  else if (strlen(confSecrets.mqtt_user) > 0 && strlen(confMqtt.mqtt_willTopic) > 0)
  {
    // user, password and Last Will
    //if (serialdebug)
    //  Serial.print(F("Auth and LWT"));
    strlcpy(connectMode, "Auth and LWT", sizeof(connectMode));
    connRes = mqttclient.connect(mqttClientId.c_str(), confSecrets.mqtt_user, confSecrets.mqtt_pass, confMqtt.mqtt_willTopic, confMqtt.mqtt_willQos, confMqtt.mqtt_willRetain, confMqtt.mqtt_willMsg);
  }
  else if (strlen(confSecrets.mqtt_user) == 0 && strlen(confMqtt.mqtt_willTopic) > 0)
  {
    // Last Will but no user and password
    //if (serialdebug)
    //  Serial.print(F("LWT, no Auth"));
    strlcpy(connectMode, "LWT, no Auth", sizeof(connectMode));
    connRes = mqttclient.connect(mqttClientId.c_str(), confMqtt.mqtt_willTopic, confMqtt.mqtt_willQos, confMqtt.mqtt_willRetain, confMqtt.mqtt_willMsg);
  }
  else
  {
    // no user, password and no Last Will
    //if (serialdebug)
    //  Serial.print(F("no Auth, no LWT"));
    strlcpy(connectMode, "no Auth, no LWT", sizeof(connectMode));
    connRes = mqttclient.connect(mqttClientId.c_str());
  }

  //if (serialdebug)
  //{
  //  Serial.print(F(", attempt: "));
  //  Serial.print(mqttReconnectAttempts);
  //  Serial.println();
  //}

  snprintf(logBuf, LOG_BUFFER_SIZE, "MQTT: connecting to %s:%u w/ %s, attempt: %u", confMqtt.mqtt_server, confMqtt.mqtt_port, connectMode, mqttReconnectAttempts);
  sendLog(logBuf, LOGLEVEL_INFO);

  if (connRes)
  { // if connection was successful
    // if (serialdebug)
    // {
    //   Serial.print(F("MQTT: connected. Reconnects: "));
    //   Serial.println(mqtt_reconnects);
    // }
    //char logBuf2[50];
    //sprintf(logBuf2, "MQTT: connected to %s:%u, Reconnects: %u", confMqtt.mqtt_server, confMqtt.mqtt_port, mqtt_reconnects);
    //sendLog(logBuf2, LOGLEVEL_INFO);
    //sendLog(F("MQTT: connected. Reconnects: "), LOGLEVEL_INFO);

    mqtt_connected = true;
    mqtt_reconnects++;
    mqttOnConnect();

    //if (serialdebug)
    //Serial.println("MQTT: subscribed topics:");
    sendLog("MQTT: subscribed topics:", LOGLEVEL_INFO);

    if (strlen(mqtt_topic_in_cmd) > 0)
    {
      char mqtt_topic_in_subscribe[52];
      sprintf(mqtt_topic_in_subscribe, "%s/%s", mqtt_topic_in_cmd, "#");
      if (mqttclient.subscribe(mqtt_topic_in_subscribe))
      {
        //if (serialdebug) {
        //  Serial.print("    ");
        //  Serial.println(mqtt_topic_in_subscribe);
        //}
        char buf[60];
        sprintf(buf, "      %s", mqtt_topic_in_subscribe);
        sendLog(buf, LOGLEVEL_INFO);
        mqttInTopicSubscribed = true;
      }
    }

    if (strlen(confAdd.outTemp_topic_in) > 0)
    {
      if (mqttclient.subscribe(confAdd.outTemp_topic_in))
      {
        //if (serialdebug) {
        //  Serial.print("    ");
        //  Serial.println(confAdd.outTemp_topic_in);
        //}
        char buf[60];
        sprintf(buf, "      %s", confAdd.outTemp_topic_in);
        sendLog(buf, LOGLEVEL_INFO);
      }
    }

    if (strlen(confAdd.outHum_topic_in) > 0)
    {
      if (mqttclient.subscribe(confAdd.outHum_topic_in))
      {
        //if (serialdebug) {
        //  Serial.print("    ");
        //  Serial.println(confAdd.outHum_topic_in);
        //}
        char buf[60];
        sprintf(buf, "      %s", confAdd.outHum_topic_in);
        sendLog(buf, LOGLEVEL_INFO);
      }
    }
    mqttReconnectAttempts = 0;
    return mqttclient.connected();
  }
  else {
    int mqttConnRes = mqttclient.state();
    mqtt_updateCurrentStateName();
    sprintf_P(logBuf, "%s: %s, rc=%d (%s)", PGMStr_MQTT, PGMStr_connectedFailed, mqttConnRes, mqttCurrentStateName);
    sendLog(logBuf, LOGLEVEL_INFO);

    if (mqttReconnectAttempts >= 3 && (mqttConnRes == 4 || mqttConnRes == 5))
    {
      // MQTT credentials are invalid/rejected from the server - stop trying to reconnect until reboot
      mqtt_tempDisabled_credentialError = true;
      if (confLog.logLevelSerial >= LOGLEVEL_VERBOSE)
      {
        //Serial.println(F("MQTT disabled until reboot due to credential error"));
        sendLog(F("MQTT disabled until reboot due to credential error"), LOGLEVEL_ERROR);
      }
    }
    mqtt_connected = false;
    mqttOnDisconnect();
    return mqttclient.connected();
  }
} //mqttReconnect

void mqttClientInit()
{
  if (confMqtt.mqtt_enable)
  {
    mqttclient.setServer(confMqtt.mqtt_server, confMqtt.mqtt_port);
    mqttclient.setCallback(mqttCallback);
    mqttLastReconnectAttempt = 0;
    mqttReconnectAttempts = 0;
    mqtt_tempDisabled_credentialError = false;

    snprintf(mqtt_topic_log, 50, "%s/%s", confMqtt.mqtt_topic_in, "log");
  }
}

int _lastWifiStatus;

void mqttHandleConnection()
{
  if (mqtt_heartbeatReceived)
  {
    mqtt_heartbeatReceived = false;
    sendLog(F("MQTT: received HEARTBEAT - resetting timer..."), LOGLEVEL_VERBOSE);
  }

  int _currWifiStatus = WiFi.status();
  if (_currWifiStatus != _lastWifiStatus)
  {
    _lastWifiStatus = _currWifiStatus;
    //Serial.print("WiFi status changed to: ");
    //Serial.println(_currWifiStatus);
  }

  if (confMqtt.mqtt_enable && !mqtt_tempDisabled_credentialError && _currWifiStatus == WL_CONNECTED)
  {
    // MQTT reconnect if not connected (nonblocking)
    boolean _doReconnect = false;
    if (!mqttclient.connected())
      _doReconnect = true;

    if (mqttInTopicSubscribed && confMqtt.mqtt_heartbeat_maxage_reconnect > 0)
    { //  if mqtt_topic_in is subscribed
      unsigned long _maxage_reconn = confMqtt.mqtt_heartbeat_maxage_reconnect * 60000;
      if (confMqtt.mqtt_enable_heartbeat && confMqtt.mqtt_heartbeat_maxage_reconnect > 0 && ((millis() - mqtt_lastHeartbeat) > _maxage_reconn))
      { // also reconnect if no HEARTBEAT was received for some time
        _doReconnect = true;
        mqtt_connected = false;
        mqtt_lastHeartbeat = millis();
        //Serial.println(F("MQTT: HEARTBEAT overdue. Force-Reconnecting MQTT..."));
        sendLog(F("MQTT: HEARTBEAT overdue. Force-Reconnecting MQTT..."), LOGLEVEL_WARN);
      }
    }

    if (_doReconnect)
    {
      unsigned long mqttReconnectAttemptDelay;
      if (mqttReconnectAttempts < 3)
        mqttReconnectAttemptDelay = MQTT_RECONNECT_INTERVAL_3_TRIES; // if this is the 1-3rd attempt, try again in 15s
      else if (mqttReconnectAttempts < 10)
        mqttReconnectAttemptDelay = MQTT_RECONNECT_INTERVAL_10_TRIES; // if more than 3 attempts failed, try again every min
      else
        mqttReconnectAttemptDelay = MQTT_RECONNECT_INTERVAL_LONG; // if more than 10 attempts failed, try again every 5 min

      unsigned long _maxage_reboot = confMqtt.mqtt_heartbeat_maxage_reboot * 60000;
      if (confMqtt.mqtt_enable_heartbeat && confMqtt.mqtt_heartbeat_maxage_reboot > 0 && ((millis() - mqtt_lastHeartbeat) > _maxage_reboot))
      {
        // if no heartbeat was received for set time, reboot the ESP
        //Serial.println(F("MQTT: HEARTBEAT_MAXAGE_REBOOT overdue. Restarting..."));
        sendLog(F("MQTT: HEARTBEAT_MAXAGE_REBOOT overdue. Restarting..."), LOGLEVEL_WARN);
        restart();
      }

      if ((millis() - mqttLastReconnectAttempt) > mqttReconnectAttemptDelay)
      {
        mqttLastReconnectAttempt = millis();
        //#ifdef USE_MQTT_TLS
        //        verifytls();
        //#endif
        mqttReconnect();
      }
      mqttclient.loop();
    }
    else
    {
      mqttclient.loop();
    }
  }
}

void mqtt_publishStatus()
{
  if (confMqtt.mqtt_enable && mqttclient.state() == 0)
  {
    char outMsg[60];
    sprintf_P(outMsg, "%s: %s: %s:%u, %s: %d", PGMStr_MQTT, PGMStr_connectedTo, confMqtt.mqtt_server, confMqtt.mqtt_port, PGMStr_reconnects, mqtt_reconnects - 1);
    //mqttclient.publish(confMqtt.mqtt_topic_out, outMsg, confMqtt.mqtt_outRetain);
    sendLog(outMsg, LOGLEVEL_INFO);
  }
}

void mqttOnConnect()
{
  mqtt_publishConnectMsg();
  mqtt_publishStatus();

  #ifdef ENABLE_LCD_I2C
  if(!confMqtt.mqtt_reconnect_silent || mqtt_reconnects <= 1) {
    display_showMQTTConnected();
  }
  #endif
}

void mqttOnDisconnect()
{
  #ifdef ENABLE_LCD_I2C
  if(!confMqtt.mqtt_reconnect_silent || mqtt_reconnects <= 1) {
    display_showMQTTConnectionError();
  }
  #endif
}

void mqtt_publishConnectMsg()
{
  if (strlen(confMqtt.mqtt_willTopic) > 4)
  {
    if (confMqtt.mqtt_enable && mqttclient.state() == 0)
    {
      mqttclient.publish(confMqtt.mqtt_willTopic, confMqtt.mqtt_connMsg, confMqtt.mqtt_willRetain);
      //sendLog(F("MQTT: connected"), LOGLEVEL_INFO);
    }
  }
}

void mqttPublishHeartbeat()
{
  if (confMqtt.mqtt_enable && confMqtt.mqtt_enable_heartbeat && mqttclient.state() == 0)
  {
    mqttclient.publish(mqtt_topic_in_cmd, "HEARTBEAT", false); // publishes to IN-topic. MQTT will force-reconnect if it does not get a HEARTBEAT for 2 min
  }
}

void mqtt_updateCurrentStateName()
{
  if (confMqtt.mqtt_enable)
  {
    sprintf_P(mqttCurrentStateName, "%s", PGMStr_MQTTStates[mqttclient.state() + 4]);
  }
  else
    sprintf_P(mqttCurrentStateName, "%s", PGMStr_MQTTState_DIS);
}

void mqttResetConnection()
{
  sendLog(F("MQTT: resetting connection..."));
  if (!confMqtt.mqtt_enable)
    mqttclient.disconnect();
  else
  {
    mqttclient.disconnect();
    mqttPrepareSubscribeTopics();
    mqttClientInit();
  }
  configChangedMqttConnResetRequired = false;
}