Mosquitto Python Paho How can i get all message since last connection

Question

I have a python client that posts to a mosquitto broker. This broker is linked to a second broker via a bridge. On deck I wish I have a python script that subscribes to the topic. I would like to be able to retrieve all the messages published between two client connections.

I just have the last message

mosquitto.conf broker

pid_file /run/mosquitto/mosquitto.pid

persistence true
persistence_location /var/lib/mosquitto/
log_type all
log_dest file /var/log/mosquitto/mosquitto.log
log_timestamp true
log_timestamp_format %Y-%m-%dT%H:%M:%S


include_dir /etc/mosquitto/conf.d

listener 1883
allow_anonymous true

persistent_client_expiration 14d

connection bridge-rpi
address 192.168.1.140:1884
topic # out 2
cleansession true
remote_clientid xxxx

mosquitto.conf bridge

listener 1884
allow_anonymous true
persistent_client_expiration 14d
persistence true
persistence_file mosquitto.db
persistence_location /var/lib/mosquitto
log_type all
log_dest file /var/log/mosquitto/mosquitto.log
log_timestamp true
log_timestamp_format %Y-%m-%dT%H:%M:%S

subscribe code :

#Configuration MQTT
MQTT_HOST = "localhost"
MQTT_PORT = 1884
MQTT_KEEPALIVE_INTERVAL = 45
MQTT_TOPIC = "cn"

#Configuration Influxdb
INFLUXDB_ADDRESS = 'localhost'
INFLUXDB_USER = 'xxxx'
INFLUXDB_PASSWORD = 'xxxx'
INFLUXDB_DATABASE = 'xxxx'

influxdb_client = InfluxDBClient(INFLUXDB_ADDRESS, 8086, INFLUXDB_USER, INFLUXDB_PASSWORD,  database=INFLUXDB_DATABASE)


# initialisation du client MQTT
mqttc = mqtt.Client("PythonMqttToInfluxDb")
#mqttc = mqtt.Client()

# connexion au broker MQTT
mqttc.connect(MQTT_HOST, MQTT_PORT, MQTT_KEEPALIVE_INTERVAL)


def on_connect(client, userdata, flags, rc):
    """ The callback for when the client connects to the broker."""
    print("Connecté au serveur MQTT avec le code de retour "+str(rc))

    # Subscribe to a topic
    client.subscribe(MQTT_TOPIC,qos=2)

def on_message(client, userdata, msg):
    """ The callback for when a PUBLISH message is received from the server."""
    print(msg.topic+" "+str(msg.payload))
    sensor_data = _parse_mqtt_message(msg.topic, msg.payload.decode('utf-8'))
    json_body = [
        {
            'measurement': "cn",
            'time': parsedJson["timestamp"],
            'tags': {
                'site': parsedJson["site"],
                'frequence': parsedJson["frequence"],
                'info': parsedJson["timestamp"]
            },
            'fields': {
                'cn': parsedJson["cn"]
                #'time': parsedJson["timestamp"]
            }
        }
    ]
    print(json_body)
    influxdb_client.write_points(json_body, time_precision='n')



## MQTT logic - Register callbacks and start MQTT client

mqttc.on_connect = on_connect
mqttc.on_message = on_message
mqttc.loop_forever()

Thanks for your help

hardillb · Answer

You need to make sure that the Clean Session flag is set to false to ensure that queued messages are delivered.

From the paho doc

Client()

Client(client_id="", clean_session=True, userdata=None, protocol=MQTTv311, transport="tcp")

The Client() constructor takes the following arguments:

client_id - the unique client id string used when connecting to the broker. If client_id is zero length or None, then one will be randomly generated. In this case the clean_session parameter must be True.

clean_session - a boolean that determines the client type. If True, the broker will remove all information about this client when it disconnects. If False, the client is a durable client and subscription information and queued messages will be retained when the client disconnects.

Note that a client will never discard its own outgoing messages on disconnect. Calling connect() or reconnect() will cause the messages to be resent. Use reinitialise() to reset a client to its original state.

userdata - user defined data of any type that is passed as the userdata parameter to callbacks. It may be updated at a later point with the user_data_set() function.

protocol - the version of the MQTT protocol to use for this client. Can be either MQTTv31 or MQTTv311 transport set to "websockets" to send MQTT over WebSockets. Leave at the default of "tcp" to use raw TCP.

e.g.

# initialisation du client MQTT
mqttc = mqtt.Client(client_id="PythonMqttToInfluxDb", clean_session=False)

Mosquitto Python Paho How can i get all message since last connection

1 Answers