Cómo usar MQTT asíncrono en el ESP8266/ESP32 con AsyncMqttClient

En primer lugar, el sketch principal básicamente es idéntico al que teníamos en el ejemplo anterior. Únicamente hemos cambiado el include de las librerías por los equivalentes en el nuevo ejemplo.

#include <WiFi.h>

#include <AsyncMqttClient.h>

#include "config.h"  // Sustituir con datos de vuestra red
#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());
}

Por otro lado, tenemos el nuevo fichero ESP32_Utils_MQTT_Async, que sustituye a ESP32_Utils_MQTT del ejemplo anterior. Aquí tenemos funciones relativas al uso de MQTT, pero que son relativamente independientes de nuestro proyecto. Por tanto, son sencillas de reusar en otros proyectos sin necesidad de modificarlas.

Tampoco es que haya tanto aquí. Básicamente lo único novedoso es que se emplean dos timers de FreeRTOS para gestionar la reconexión del MQTT y el WiFi, en caso de que haya un fallo de conexión

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);
}

En el último fichero ‘MQTT.hpp’, hemos guardado las funciones relativa al MQTT que sí están íntimamente relacionadas con nuestro proyecto. Así, tenemos una función para suscribirse a los topics, una para publicar mensajes, y otra de callback a ejecutar al recibir un mensaje.

Es decir, básicamente tenemos el mismo ejemplo que en la entrada anterior, con la misma estructura y funcionamiento de programa, pero con las funciones adaptadas a 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();
}