Asynchronous handlers

[DurandA]

Is there any plan to switch to asynchronous handlers?

I'd like to use an asynchronous function to store messages to a database using asyncio:

async def on_message(client, topic, payload, qos, properties):
    pass

mqtt_client.on_message = on_message

Indeed it is still possible to add a new task to the event loop; I just thought it would be nicer.

[Lenka42]

Hi, @DurandA
It's now possible to use asynchronous on_message callback (yes, only this one at the moment)
I will add comment about this to README in nearest future. You can see example usage here: https://github.com/wialon/gmqtt/blob/master/tests/test_mqtt5.py#L268 The only important thing here: asynchronous callback must return proper PUBACK code which will be used in PUBACK response to broker. So if everything went fine, just return 0. Full list of possible codes is here: https://github.com/wialon/gmqtt/blob/master/gmqtt/mqtt/constants.py#L69

[DurandA]

Awesome, thank you!

[DurandA]

@Lenka42 The error is still present when using functools.partial.

[Lenka42]

Could you please share minimal peace of code with your on_message handler?

[DurandA]

After some investigation, the issue is due to iscoroutinefunction() which does not work with functools.partial.

Here is a minimal example:

def on_connect(client, flags, rc, properties):
    client.subscribe('#')

async def on_message(client, topic, payload, qos, properties, loop=None):
    await asyncio.sleep(1, loop=loop)
    return 0

async def init(loop):
    mqtt_client = MQTTClient('') # anonymous
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = partial(on_message, loop=loop)

[Lenka42]

@DurandA I released fix for partial coroutines, please check newest version

[DurandA]

Many thanks!