empicano
commented
6 months ago Hi there Stefan,
I'll respond to your question of handling incoming messages concurrently. We can also happily discuss how to make aiomqtt more elegant if you have specific ideas!
The second admonition in the "Message queue" section of our docs shows a small example of how to handle each message in a separate coroutine. However, as far as I understand your pseudocode, you'd like to have one coroutine per subscription instead of per message.
We'll need to dive a bit into asyncio for this 😋 The idea is to implement a "distributor" that sorts the incoming messages into different asyncio queues, which are then processed concurrently (but sequentially inside a subscription). Here's a minimal working example:
import asyncio
import aiomqtt
async def fast_producer(client: aiomqtt.Client):
while True:
await asyncio.sleep(0.2)
await client.publish("fast", "fast")
async def fast_consumer():
while True:
message = await fast_queue.get()
print(f"Fast consumer received: {message.payload}")
async def slow_producer(client: aiomqtt.Client):
while True:
await asyncio.sleep(2)
await client.publish("slow", "slow")
async def slow_consumer():
while True:
message = await slow_queue.get()
print(f"Slow consumer received: {message.payload}")
fast_queue = asyncio.Queue()
slow_queue = asyncio.Queue()
async def distributor(client: aiomqtt.Client):
async with client.messages() as messages:
await client.subscribe("fast")
await client.subscribe("slow")
# Sort messages into the appropriate queues
async for message in messages:
if message.topic.matches("fast"):
fast_queue.put_nowait(message)
elif message.topic.matches("slow"):
slow_queue.put_nowait(message)
async def main():
async with aiomqtt.Client("test.mosquitto.org") as client:
# Use a task group to manage and await all tasks
async with asyncio.TaskGroup() as tg:
tg.create_task(fast_producer(client))
tg.create_task(fast_consumer())
tg.create_task(slow_producer(client))
tg.create_task(slow_consumer())
tg.create_task(distributor(client))
asyncio.run(main())Does that make sense? 🙂 Issue #250 is similar and could have some more context.
skinkie
smurfix
The example denotes the following.
If a user wants to also test for temperature:
Is there any way to do this more elegantly? Maybe even in a concurrent way?
Pseudo: