Examples of interaction with amqtt broker

[hollymcr]

There are lots of examples of creating a broker with amqtt, but I cannot find any examples that interact with amqtt once the broker is up and running.

For example, if I simply wanted every value published to a topic to get also written to a database, how would I do that? Should I just add client code to the same application with a subscription to # - and if so how can I avoid any risk of missing any data (eg ensuring no data is lost before my client is connected)? What if I also wanted to record each broker connection being opened/closed - ie events that I can't just capture by subscribing to a topic?

[not-f-elsner]

AFAIK

To handle broker events, you can create a custom plugin for the broker which will get notified when any of the following events occur:

EVENT_BROKER_PRE_START = "broker_pre_start"
EVENT_BROKER_POST_START = "broker_post_start"
EVENT_BROKER_PRE_SHUTDOWN = "broker_pre_shutdown"
EVENT_BROKER_POST_SHUTDOWN = "broker_post_shutdown"
EVENT_BROKER_CLIENT_CONNECTED = "broker_client_connected"
EVENT_BROKER_CLIENT_DISCONNECTED = "broker_client_disconnected"
EVENT_BROKER_CLIENT_SUBSCRIBED = "broker_client_subscribed"
EVENT_BROKER_CLIENT_UNSUBSCRIBED = "broker_client_unsubscribed"
EVENT_BROKER_MESSAGE_RECEIVED = "broker_message_received"
EVENT_MQTT_PACKET_SENT = "mqtt_packet_sent"
EVENT_MQTT_PACKET_RECEIVED = "mqtt_packet_received"

You can see how that works in the amqtt/plugins/logging.py plugins for example.

In amqtt/plugins/persistence.py you can also find an example for hooking a database connection into the broker itself (although for handling sessions). This is probably the safest way to avoid missing any data at all.

Otherwise, you could also let an MQTT client run in a different process (or in a different task of the same process) and subscribe to the topic you need, like

await client.subscribe(
    [
        ("/#", QOS_2),
    ]
)

See also amqtt/samples/client_subscribe.py

But there might be some cases where data will be missed, e.g. before the client has connected once.

If your client is run separately, you will also want to use a fixed client_id so if your client disconnects, any messages which cannot be delivered will be retained for that client until the client reconnects.

[FlorianLudwig]

@hollymcr I converted your issue into a discussion. If the answer helped, please mark it :) thanks

[hollymcr]

I'd be happy to write it as a plugin, but I couldn't see how to bootstrap one. Admittedly it's been a couple of years since I last properly looked at it (I still need this functionality, I just can't remember everything I tried before), but I couldn't see how to actually write a plugin and have it loaded.

If there's a "how to create a plugin" readme somewhere please point me in that direction!