Send and receive messages via MQTT with Arduino and the PubSubClient library

We continue with the posts dedicated to MQTT communication by seeing how to send or receive MQTT messages from a processor like Arduino thanks to the PubSubClient library.

We already have several posts within the MQTT dedicated series, seeing What is MQTT and its importance in IoT, What are Topics and how to use them and learned how to install Mosquitto, one of the most popular MQTT brokers.

Being this blog what it is, it was clear from afar that the ultimate goal was to apply it to a microprocessor. After all, one of MQTT’s main functions is to be applied to MCUs to make IoT devices.

If you think it’s going to be very difficult, you can start losing your fear! MQTT communication is one of the easiest communication methods to use that we can employ in our projects. In fact, as we saw when presenting the protocol, ease and robustness are among MQTT’s strong points.

In the case of MCUs, fortunately, integrating MQTT into a processor like Arduino is very simple thanks to the existence of several libraries. The most popular and well-known is the great PubSubClient library developed by Nick O’Leary.

PubSubClient is compatible with a wide variety of devices and web interfaces. When it was developed, it was mainly intended for a combination of Arduino with an Ethernet shield or a WiFi shield. Therefore, you’ll see that many examples and tutorials refer to these shields.

However, nowadays, it is more common to use PubSubClient on more advanced processors that natively incorporate WiFi. Such as, for example, our beloved ESP8266 and ESP32, which are well-known on the blog in their own section.

PubSubClient is an MQTT client for microprocessors and IoT devices. By default, it uses MQTT 3.1.1, although it can be changed by modifying the MQTT_VERSION variable in the PubSubClient.h file. It allows subscribing to messages at QoS 0 or QoS 1, although it is only possible to publish messages at QoS 0.

The maximum message size to send, including the header, is 256 bytes. This is sufficient for most projects. However, it is possible to change it by modifying the MQTT_MAX_PACKET_SIZE constant in the PubSubClient.h file.

The KeepAlive interval is 15 seconds by default, although it is also possible to change it via the MQTT_KEEPALIVE constant, or by calling the static function PubSubClient::setKeepAlive(keepAlive).

The PubSubClient library is Open Source, and all the code is available at https://github.com/knolleary/pubsubclient. You also have it available through the Arduino IDE library manager.

Using the PubSub Library

Constructors

We can initialize the library using one of its constructors

PubSubClient ()
PubSubClient (client)
PubSubClient (server, port, [callback], client, [stream])

Where:

• Client, the communication interface we are using
• Server, the broker’s name and its IP
• Port, the broker’s port, default 1883
• Callback, function to execute when a message is received
• Stream, a stream instance to store messages

Subscribe to a topic

To subscribe to a topic we use the following function:

mqttClient.subscribe("hello/world");

Send a message

To send a message on a topic we simply do,

mqttClient.publish("hello/world", (char*)payload.c_str());

Important functions

The other fundamental function of the pubsub library is loop(), which must be called frequently to allow the client to handle MQTT requests and sends.

mqttClient.loop();

Other interesting functions of the PubSubClient object are:

boolean connect(const char* id);
boolean connect(const char* id, const char* user, const char* pass);
void disconnect();

boolean publish(const char* topic, const char* payload);

boolean subscribe(const char* topic);
boolean subscribe(const char* topic, uint8_t qos); // qos is 0 or 1, default 0
boolean unsubscribe(const char* topic);

PubSubClient& setServer(IPAddress ip, uint16_t port);
PubSubClient& setCallback(MQTT_CALLBACK_SIGNATURE);
PubSubClient& setClient(Client& client);
PubSubClient& setStream(Stream& stream);
PubSubClient& setKeepAlive(uint16_t keepAlive);
PubSubClient& setSocketTimeout(uint16_t timeout);

// callback function prototype
void OnMqttReceived(char* topic, byte* payload, unsigned int length) 

Code Example

Next, we have a code example, based on the examples from the library itself. This example is based on an Ethernet Shield, but the operation is similar with any other connection.

#include <Ethernet.h>

#include <PubSubClient.h>

// IP addresses, server, and MAC
// necessary for Ethernet Shield
IPAddress ip(172, 16, 0, 100);
IPAddress server(172, 16, 0, 2);
byte mac[] = {
    0xDE,
    0xED,
    0xBA,
    0xFE,
    0xFE,
    0xED};

// instantiate objects
EthernetClient ethClient;
PubSubClient client(ethClient);

// MQTT constants
// broker address, port, and client name
const char *MQTT_BROKER_ADRESS = "192.168.1.150";
const uint16_t MQTT_PORT = 1883;
const char *MQTT_CLIENT_NAME = "ArduinoClient_1";

// subscribes to the topic
// in this example, only to 'hello/world'
void SuscribeMqtt()
{
    mqttClient.subscribe("hello/world");
}

// callback to execute when a message is received
// in this example, shows the received message by serial
void OnMqttReceived(char *topic, byte *payload, unsigned int length)
{
    Serial.print("Received on ");
    Serial.print(topic);
    Serial.print(": ");

    String content = "";
    for (size_t i = 0; i < length; i++)
    {
        content.concat((char)payload[i]);
    }
    Serial.print(content);
    Serial.println();
}

// starts the MQTT communication
// starts setting the server and the callback when receiving a message
void InitMqtt()
{
    mqttClient.setServer(MQTT_BROKER_ADRESS, MQTT_PORT);
    mqttClient.setCallback(OnMqttReceived);
}

// connects or reconnects to MQTT
// connects -> subscribes to topic and publishes a message
// no -> waits 5 seconds
void ConnectMqtt()
{
    Serial.print("Starting MQTT connection...");
    if (mqttClient.connect(MQTT_CLIENT_NAME))
    {
        SuscribeMqtt();
        client.publish("connected","hello/world");
    }
    else
    {
        Serial.print("Failed MQTT connection, rc=");
        Serial.print(mqttClient.state());
        Serial.println(" try again in 5 seconds");

        delay(5000);
    }
}

// manages MQTT communication
// checks if the client is connected
// no -> tries to reconnect
// yes -> calls the MQTT loop
void HandleMqtt()
{
    if (!mqttClient.connected())
    {
        ConnectMqtt();
    }
    mqttClient.loop();
}

// only starts serial, ethernet, and MQTT
void setup()
{
    Serial.begin(9600);

    Ethernet.begin(mac, ip);
    delay(1500);
    InitMqtt();
}

// only calls HandleMqtt
void loop()
{
    HandleMqtt();
}

I have put quite a few comments in the code to explain it, but as highlights:

• We have divided the code into “more or less” reusable functions
• In the setup, it is only necessary to call the InitMqtt function
• InitMqtt sets the OnMqttReceived callback, which displays received messages via serial
• On the other hand, in the loop, we call the HandleMqtt function
• The HandleMqtt function reconnects the client if necessary, otherwise it executes the client’s loop
• If it is necessary to connect/reconnect, it calls ConnectMqtt
• ConnectMqtt connects the client and subscribes to topics with SuscribeMqtt

If we run it, we will see the received messages on the serial port. For testing, for example, we can use MQTT Explorer, a generic client we saw in this post.

That easy! As we anticipated, it’s not at all complicated to connect a microprocessor to an MQTT network. In fact, it’s very simple, and this is one of the advantages and strengths of this system.

The next posts about MQTT will be in the ESP8266 and ESP32 section. Although we will refer to this post when necessary, we will also see that these two have another library for performing MQTT connection asynchronously. See you in the next post.

Download the code

All the code from this post is available for download on Github.