Understanding Asynio Better?

[Jibun-no-Kage]

I am looking that the mqtt_as code, but my explicit understanding of asyncio is not complete. As each MQTT subscribed message is received, is it handled as a unique asyncio task? As say a routine/function defined async and called via create_task()?

For example I have a topic called 'then' that via a dispatch route calls DoThen(), DoThen in turn publishes a response back to the original sender all via MQTT. The publish is established via create_task(). So... which diagram below is correct or most accurate?

sender->then->ESP32->message->DoThen->publish
sender->then->ESP32->message->DoThen->publish

Or...

sender->then->ESP32->message->then->DoThen->publish
sender->then->ESP32-/ ->publish

If the message callback via the MQTT client is a separate task, then the first diagram is applicable. But if message callback is just working off a queue, for example, the second diagram is applicable.

Of course because I am calling publish via create_task() that is definitely parallel processing (as is such under asyncio).

In traditional python, having the MQTT message call back enqueue message, and a separate thread or process dequeue messages and routes to the respective routines, is not parallel processing per message, but is parallel processing in the fact that the main thread, client thread, and message dispatch thread handle messages in parallel to each other.

[peterhinch]

I'm afraid I don't understand your query.

From the point of view of the ESP32 running mqtt_as, if you're using the callback interface, the following occurs. You are subscribed to topic then. A message arrives on that topic and triggers a callback. If that (synchronous) callback wishes to publish, it uses asyncio.create_task(client.publish(...)) to schedule publication which occurs as a background task.

With the event based interface, incoming messages are queued and are handled by a task like this (from range.py):

async def messages(client):
    async for topic, msg, retained in client.queue:
        print(f'Topic: "{topic.decode()}" Message: "{msg.decode()}" Retained: {retained}')
        asyncio.create_task(pulse())

You would replace the pulse() task with client.publish(...).

If you want clarification about uasyncio, please see the tutorial.