MQTT com ESP32 envio de dados binários

Exemplo de envio de dados usando MQTT com a ESP32. Exemplo prático (funcional) do MQTT com ESP32 Async

Usei a lib Async MQTT Client

#define MQTT_MAX_PACKET_SIZE 30000  // Increase the maximum packet size
 
#include <WiFi.h>
#include <AsyncMqttClient.h>
 
#define WIFI_SSID "my-wifi"          // Replace with your Wi-Fi SSID
#define WIFI_PASSWORD "my-password"  // Replace with your Wi-Fi password
 
#define MQTT_HOST IPAddress(0, 0, 0, )  // Replace with your MQTT broker's IP`
#define MQTT_PORT 1883                         // Replace with your MQTT broker's port if different
 
AsyncMqttClient mqttClient;
TimerHandle_t wifiReconnectTimer;
TimerHandle_t mqttReconnectTimer;
 
const int ARRAY_SIZE = 5000;
int randomArray[ARRAY_SIZE];
 
void connectToWifi() {
  Serial.println("Connecting to Wi-Fi...");
  WiFi.begin(WIFI_SSID, WIFI_PASSWORD);
}
 
void connectToMqtt() {
  Serial.println("Connecting to MQTT...");
  mqttClient.connect();
}
 
void WiFiEvent(WiFiEvent_t event) {
  switch (event) {
    case ARDUINO_EVENT_WIFI_STA_GOT_IP:
      Serial.println("WiFi connected");
      Serial.print("IP address: ");
      Serial.println(WiFi.localIP());
      connectToMqtt();
      break;
    case ARDUINO_EVENT_WIFI_STA_DISCONNECTED:
      Serial.println("WiFi lost connection");
      xTimerStop(mqttReconnectTimer, 0);
      xTimerStart(wifiReconnectTimer, 0);
      break;
  }
}
 
void onMqttConnect(bool sessionPresent) {
  Serial.println("Connected to MQTT.");
 
  // Generate random numbers
  Serial.println("Generating random numbers...");
  for (int i = 0; i < ARRAY_SIZE; i++) {
    randomArray[i] = random(0, 1000000);  // Generates random integers between 0 and 999,999
  }
 
  // Send the array via MQTT
  Serial.println("Publishing message...");
 
  size_t payloadSize = sizeof(randomArray);
  uint16_t packetIdPub = mqttClient.publish(
    "test/randomArray",  // Topic name
    0,                   // QoS level
    false,               // Retain flag
    (const char*)randomArray,
    payloadSize
  );
 
  Serial.print("Publishing at QoS 0, packetId: ");
  Serial.println(packetIdPub);
}
 
void onMqttDisconnect(AsyncMqttClientDisconnectReason reason) {
  Serial.println("Disconnected from MQTT.");
 
  if (WiFi.isConnected()) {
    xTimerStart(mqttReconnectTimer, 0);
  }
}
 
void setup() {
  Serial.begin(115200);
 
  // Seed the random number generator
  randomSeed(analogRead(0));
 
  wifiReconnectTimer = xTimerCreate(
    "wifiTimer",
    pdMS_TO_TICKS(2000),
    pdFALSE,
    (void*)0,
    reinterpret_cast<TimerCallbackFunction_t>(connectToWifi)
  );
  mqttReconnectTimer = xTimerCreate(
    "mqttTimer",
    pdMS_TO_TICKS(2000),
    pdFALSE,
    (void*)0,
    reinterpret_cast<TimerCallbackFunction_t>(connectToMqtt)
  );
 
  WiFi.onEvent(WiFiEvent);
 
  mqttClient.onConnect(onMqttConnect);
  mqttClient.onDisconnect(onMqttDisconnect);
  mqttClient.setCredentials("username", "password");
  mqttClient.setServer(MQTT_HOST, MQTT_PORT);
 
  connectToWifi();
}
 
void loop() {
  // Nothing to do here since we're using asynchronous callbacks
}

Leitura dos dados recebidos no Python usando Paho Client:

import paho.mqtt.client as mqtt
import struct
 
# Define MQTT broker details
broker = 'broker-host'  # Replace with your broker's address
port = 1883  # Typically 1883 for non-SSL, 8883 for SSL
topic = 'test/randomArray'  # Topic to subscribe/publish
username = 'username'  # Replace with your username
password = 'password'  # Replace with your password
 
# Define callback for connection
def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("Connected to broker")
        client.subscribe(topic)  # Subscribe to the topic upon successful connection
    else:
        print(f"Connection failed with code {rc}")
 
# Define callback for received messages
def on_message(client, userdata, msg):
    print(f"Message received: {msg.payload} on topic {msg.topic}")
 
    payload = msg.payload
 
    print(f"Received msg on topic {msg.topic} with payload size {len(payload)} bytes")
 
    # Calculate the number of integers in the payload
    num_integers = len(payload) // 4  # 4 bytes per integer
 
    # Check if the payload size is as expected
    if num_integers != 5000:
        print(f"Warning: Expected 5000 integers, but received {num_integers}")
 
    # Unpack the binary data into a tuple of integers
    # Use '>' for big-endian or '<' for little-endian depending on how the data was sent
    integers = struct.unpack('<' + 'i' * num_integers, payload)  # '<' for little-endian
 
    # Now 'integers' is a tuple containing the integers
    print("First 10 integers received:", integers[:10])
 
    # If you want to convert the tuple to a list
    integers_list = list(integers)
    print('Integers list:',integers_list)
 
# Initialize the MQTT client
client = mqtt.Client()
 
# Set username and password
client.username_pw_set(username, password)
 
# Attach callbacks
client.on_connect = on_connect
client.on_message = on_message
 
# Connect to the broker
client.connect(broker, port, 60)
 
# Function to publish a message
def publish_message(client, topic, message):
    client.publish(topic, message)
    print(f"Message '{message}' published on topic '{topic}'")
 
# Start the MQTT client loop to handle reconnects, subscriptions, etc.
client.loop_start()
 
# Example of publishing a message
publish_message(client, topic, "Hello, MQTT!")
 
# Keep the script running for a while to receive messages
try:
    while True:
        pass
except KeyboardInterrupt:
    print("Disconnecting from broker")
    client.loop_stop()
    client.disconnect()