bkanuka / trio-paho-mqtt

trio async MQTT client that wraps paho-mqtt

trio_paho_mqtt

trio specific async MQTT client. The goal of this project is to have all the MQTT protocol stuff handled by paho-mqtt, while the async loopy stuff is handled by trio. This keeps our MQTT communication async, without having to shove paho-mqtt into a thread, and without having to reimplement the MQTT protocol.

All callbacks are also removed to be more trio-like. Messages are accessed through an async for loop.

This is meant to be a light-as-possible wrapper around paho-mqtt. That way if something goes wrong with MQTT message parsing, it's not my fault ;-)

Usage

First, see the example.py and paho-mqtt documentation. The goal is to expose all the same methods as the paho-mqtt Client. If any methods (besides callbacks) are missing or do something unexpected, this is a bug. Please open an issue.

AsyncClient is initialized with a sync mqtt.Client and a trio.Nursery but then otherwise should operate much like the sync Client. Methods are not coroutines, they are non-blocking functions, so you do not need to call await. The only exception is accessing messages through async for.

Example

import paho.mqtt.client as mqtt
import trio
from trio_paho_mqtt import AsyncClient


async def handle_messages(client):
    async for msg in client.messages():
        print(f"Received msg: {msg}")

async with trio.open_nursery() as nursery:
    sync_client = mqtt.Client()
    client = AsyncClient(sync_client, nursery)
    client.connect('mqtt.eclipse.org', 1883, 60)
    client.subscribe(topic)
    client.publish(topic, msg)

    nursery.start_soon(handle_messages, client)

Related projects

• gmqtt - Uses asyncio and callbacks. Isn't compatible with trio.
• hbmqtt - Uses asyncio. Reimplements the MQTT protocol.
• aiomqtt - Uses asyncio and wraps paho, but still uses the loop from paho-mqtt. I believe all operations are put into a worker thread.
• distmqtt - anyio-ified hbmqtt fork. Works with trio.
• asyncio-mqtt - Same idea as this lib (a light wrapper for paho mqtt), but for asyncio.

trio-paho-mqtt's Issues

Deploy latest version to https://pypi.org/

As of trio 0.18.0, the hazmat package has been replaced with the lowlevel package. You already made this change in your code, you just need to upload a new version to pypi.

Might be worth checking out guest mode?

I haven't dug into how y'all are wrapping paho.mqtt but thought I'd mention that you can run trio on an asyncio loop now using the new guest mode (and maybe even more easily using the even newer inside-out-guest-mode).

I need a trio-ized mqtt client for my own work namely to pull real-time data feeds from a broker: https://github.com/tedchou12#64

subscribe doesn't take the same parameters that paho takes

Here is an example:

client.subscribe("temp/#", qos=0, client_id='furnace_receive')

TypeError: subscribe() got an unexpected keyword argument 'client_id'

'pip install trio-paho-mqtt' doesn't install trio_paho_mqtt

Running 'pip install trio-paho-mqtt' doesn't install trio_paho_mqtt directory or the trio_paho_mqtt.py file.
This is with python 3.7.5 on macos

Subscriptions are dropped following reconnect

Hi,

Thank you for trio-paho-mqtt!

Following the pattern in the README.md example, I did something equivalent to:

async with trio.open_nursery() as nursery:
    sync_client = mqtt.Client()
    client = AsyncClient(sync_client, nursery)
    client.connect(...)
    client.subscribe(topic)
    async for message in client.messages():
        ...

However I seemed to stop receiving messages after a while. I think this is because the connection timed out, the (sync) client reconnected automatically, but the subscription wasn't recreated.

https://github.com/eclipse/paho.mqtt.python/blob/master/README.rst says:

# The callback for when the client receives a CONNACK response from the server.
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))

    # Subscribing in on_connect() means that if we lose the connection and
    # reconnect then subscriptions will be renewed.
    client.subscribe("$SYS/#")

However I can't straightforwardly do this in trio-paho-mqtt because you're already using the sync client on_connect method for internal purposes, and aren't exposing that to the API user.

I wonder if you could provide on_connect functionality directly from AsyncClient please?