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

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

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

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

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

In the case of MCUs, fortunately, integrating MQTT into a processor like Arduino is very simple thanks to the existence of various libraries. The most popular and well-known library is the brilliant 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, as you will see, many examples and tutorials refer to these shields.

However, at this point, it is more common to use PubSubClient in more advanced processors that incorporate WiFi natively. 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 changing the MQTT_VERSION variable in the PubSubClient.h file. It allows subscribing to messages in QoS 0 or QoS 1, although it is only possible to publish messages in QoS 0.

The maximum message size to be sent, including the header, is 256 bytes. This is sufficient for most projects. However, it is possible to vary it by changing 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 using 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 library manager of the Arduino IDE.

Using the PubSub library

Constructors

We can initialize the library using one of its constructors

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

Being:

• Client, communication interface that we are using
• Server, broker name and its IP
• Port, broker port, default 1883
• Callback, function to execute when receiving a message
• Stream, a stream instance to store the 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 to a topic, 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 transmissions.

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 of 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 many comments in the code to explain it, but some noteworthy points are:

• 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 the received messages by 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 the 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 that we saw in this post.

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

The next entries on 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 to perform the MQTT connection asynchronously. See you in the next entry.

Download the code

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