Eclipse Paho MQTT Python Client

The network loop functions are essential for processing incoming/outgoing data and dispatching callbacks. Do not mix different loop functions. There are three primary ways to manage the loop:

1. loop_start() / loop_stop(): Implements a threaded interface. loop_start() runs a background thread to call loop() automatically, freeing the main thread for other tasks. It also handles automatic reconnection. Use loop_stop() to terminate the thread.
2. loop_forever(): A blocking call that processes network traffic and handles reconnections. It will not return until disconnect() is called. Use retry_first_connection=True to retry the initial connection attempt.
3. loop(): A manual, non-blocking (up to a timeout) call. You must call this regularly in a loop to process network events. If using this method, you are responsible for implementing your own reconnection strategy.

# Threaded interface example
mqttc.loop_start()

while True:
    temperature = sensor.blocking_read()
    mqttc.publish("paho/temperature", temperature)

mqttc.loop_stop()

When using clean_session=False, the client session is stored only in memory and is not persisted to disk. If the client object is recreated (e.g., the program restarts), the session is lost, which can lead to message loss in the following scenarios:

• QoS 2 messages received but not acknowledged: The client may lose messages that were received from the server but not fully acknowledged.
• QoS 1 and QoS 2 messages sent but not acknowledged: Messages passed to publish() may be lost if the client restarts before acknowledgment.

Note on Compliance: When clean_session=True, the library will republish QoS > 0 messages across network reconnections to prevent loss. This deviates from the MQTT standard (which suggests discarding such messages) and may result in QoS 2 messages being received twice. To ensure the QoS 2 guarantee of exactly one delivery, set clean_session=False.

If you are upgrading to the new callback API in version 2.0, you must update your callback signatures. The new signatures provide more consistent arguments across MQTTv3 and MQTTv5, including reason_code and properties.

on_connect

For MQTTv3, the integer rc is replaced by reason_code (an instance of ReasonCode). It is highly recommended to compare reason_code against string values rather than integers to ensure compatibility.

on_disconnect

Signature changes to include flags, reason_code, and properties.

on_subscribe

Signature changes to include reason_codes (always a list) and properties.

on_unsubscribe

Signature changes to include reason_codes (always a list) and properties.

on_publish

Signature changes to include reason_codes and properties.

on_message

No changes required; the signature remains the same.

# NEW on_connect for both MQTTv3 and MQTTv5
def on_connect(client, userdata, flags, reason_code, properties):
    if flags.session_present:
        # ...
    if reason_code == 0:
        # success connect
    if reason_code > 0:
        # error processing

# Recommended way to handle specific errors using strings
def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code == "Unsupported protocol version":
        # handle bad protocol version
    if reason_code == "Client identifier not valid":
        # handle bad identifier

# NEW on_disconnect
def on_disconnect(client, userdata, flags, reason_code, properties):
    if reason_code == 0:
        # success disconnect
    if reason_code > 0:
        # error processing

# NEW on_subscribe
def on_subscribe(client, userdata, mid, reason_codes, properties):
    for sub_result in reason_codes:
        if sub_result == 1:
            # process QoS == 1
        if sub_result >= 128:
            # error processing

# NEW on_unsubscribe
def on_unsubscribe(client, userdata, mid, reason_codes, properties):
    for unsub_result in reason_codes:
        if unsub_result >= 128:
            # error processing

# NEW on_publish
def on_publish(client, userdata, mid, reason_codes, properties):
    # ...

# on_message remains unchanged
def on_message(client, userdata, message):
    # ...

To create a basic MQTT client that subscribes to a topic and prints messages, follow these steps:

1. Define an on_connect callback to handle the connection and perform subscriptions. Subscribing inside on_connect ensures subscriptions are renewed if the client reconnects.
2. Define an on_message callback to process incoming messages.
3. Instantiate mqtt.Client using a CallbackAPIVersion (use VERSION2 for modern MQTT 3.x/5.x support).
4. Assign the callbacks to the client instance.
5. Connect to the broker using connect().
6. Start the network loop using loop_forever() to process traffic and handle reconnections.

import paho.mqtt.client as mqtt
from paho.mqtt.enums import CallbackAPIVersion

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Connected with result code {reason_code}")
    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("$SYS/#")

# The callback for when a PUBLISH message is received from the server.
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))

mqttc = mqtt.Client(CallbackAPIVersion.VERSION2)
mqttc.on_connect = on_connect
mqttc.on_message = on_message

mqttc.connect("test.mosquitto.org", 1883, 60)

# Blocking call that processes network traffic, dispatches callbacks and
# handles reconnecting.
mqttc.loop_forever()