How to use asynchronous MQTT on ESP8266/ESP32 with AsyncMqttClient

First, the main sketch is basically identical to what we had in the previous example. We have only changed the include of the libraries to the equivalents in the new example.

#include <WiFi.h>

#include <AsyncMqttClient.h>

#include "config.h"  // Replace with your network data
#include "MQTT.hpp"
#include "ESP8266_Utils.hpp"
#include "ESP32_Utils_MQTT_Async.hpp"

void setup()
{
  Serial.begin(115200);

  delay(500);

  WiFi.onEvent(WiFiEvent);
  InitMqtt();

  ConnectWiFi_STA();
}

void loop()
{
  delay(1000);
  
  PublisMqtt(millis());
}

On the other hand, we have the new file ESP32_Utils_MQTT_Async, which replaces ESP32_Utils_MQTT from the previous example. Here we have functions related to the use of MQTT, but that are relatively independent of our project. Therefore, they are easy to reuse in other projects without the need to modify them.

There is not much here either. Basically, the only new thing is that two FreeRTOS timers are used to manage the reconnection of MQTT and WiFi in case of a connection failure.

TimerHandle_t mqttReconnectTimer;
TimerHandle_t wifiReconnectTimer;

void ConnectToMqtt()
{
  Serial.println("Connecting to MQTT...");
  mqttClient.connect();
}

void WiFiEvent(WiFiEvent_t event)
{
  Serial.printf("[WiFi-event] event: %d\n", event);
  switch(event)
  {
  case SYSTEM_EVENT_STA_GOT_IP:
    Serial.println("WiFi connected");
    Serial.println("IP address: ");
    Serial.println(WiFi.localIP());
    ConnectToMqtt();
    break;
  case SYSTEM_EVENT_STA_DISCONNECTED:
    Serial.println("WiFi lost connection");
    xTimerStop(mqttReconnectTimer, 0); // ensure we don't reconnect to MQTT while reconnecting to Wi-Fi
    xTimerStart(wifiReconnectTimer, 0);
    break;
  }
}

void OnMqttConnect(bool sessionPresent)
{
  Serial.println("Connected to MQTT.");
  Serial.print("Session present: ");
  Serial.println(sessionPresent);
  SuscribeMqtt();
}

void OnMqttDisconnect(AsyncMqttClientDisconnectReason reason)
{
  Serial.println("Disconnected from MQTT.");

  if(WiFi.isConnected())
  {
    xTimerStart(mqttReconnectTimer, 0);
  }
}

void OnMqttSubscribe(uint16_t packetId, uint8_t qos)
{
  Serial.println("Subscribe acknowledged.");
  Serial.print("  packetId: ");
  Serial.println(packetId);
  Serial.print("  qos: ");
  Serial.println(qos);
}

void OnMqttUnsubscribe(uint16_t packetId)
{
  Serial.println("Unsubscribe acknowledged.");
  Serial.print("  packetId: ");
  Serial.println(packetId);
}

void OnMqttPublish(uint16_t packetId)
{
  Serial.println("Publish acknowledged.");
  Serial.print("  packetId: ");
  Serial.println(packetId);
}

void InitMqtt()
{
  mqttReconnectTimer = xTimerCreate("mqttTimer", pdMS_TO_TICKS(2000), pdFALSE, (void*)0, reinterpret_cast<TimerCallbackFunction_t>(ConnectToMqtt));
  wifiReconnectTimer = xTimerCreate("wifiTimer", pdMS_TO_TICKS(5000), pdFALSE, (void*)0, reinterpret_cast<TimerCallbackFunction_t>(ConnectWiFi_STA));

  mqttClient.onConnect(OnMqttConnect);
  mqttClient.onDisconnect(OnMqttDisconnect);

  mqttClient.onSubscribe(OnMqttSubscribe);
  mqttClient.onUnsubscribe(OnMqttUnsubscribe);

  mqttClient.onMessage(OnMqttReceived);
  mqttClient.onPublish(OnMqttPublish);

  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
}

In the last file ‘MQTT.hpp’, we have saved the functions related to MQTT that are intimately related to our project. Thus, we have a function to subscribe to topics, one to publish messages, and a callback function to execute when receiving a message.

That is, basically we have the same example as in the previous entry, with the same program structure and operation, but with the functions adapted to AsyncMqttClient.

const IPAddress MQTT_HOST(192, 168, 1, 150);
const int MQTT_PORT = 1883;

AsyncMqttClient mqttClient;

String GetPayloadContent(char* data, size_t len)
{
  String content = "";
  for(size_t i = 0; i < len; i++)
  {
    content.concat(data[i]);
  }
  return content;
}

void SuscribeMqtt()
{
  uint16_t packetIdSub = mqttClient.subscribe("hello/world", 0);
  Serial.print("Subscribing at QoS 2, packetId: ");
  Serial.println(packetIdSub);
}

String payload;
void PublisMqtt(unsigned long data)
{
  String payload = String(data);
  mqttClient.publish("hello/world", 0, true, (char*)payload.c_str());
}
void OnMqttReceived(char* topic, char* payload, AsyncMqttClientMessageProperties properties, size_t len, size_t index, size_t total)
{
  Serial.print("Received on ");
  Serial.print(topic);
  Serial.print(": ");

  String content = GetPayloadContent(payload, len);
  Serial.print(content);
  Serial.println();
}