Hide navigation sidebar
Hide table of contents sidebar
Skip to content
Back to top
View this page
Toggle table of contents sidebar

# Subscribing to a topic

To receive messages for a topic, we need to subscribe to it. Incoming messages are queued internally. You can use the `Client.messages` generator to iterate over incoming messages. This is a minimal working example that listens for messages to the `temperature/#` wildcard:

    import asyncio
    import aiomqtt

    async def main():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            async for message in client.messages:
                print(message.payload)

    asyncio.run(main())

Now you can use the minimal publisher example to publish a message to `temperature/outside` and see it appear in the console.

Tip

You can set the Quality of Service of the subscription by passing the `qos` parameter to `Client.subscribe()`.

## Filtering messages

Imagine that we measure temperature and humidity on the outside and inside. We want to receive all types of measurements but handle them differently.

You can filter messages with `Topic.matches()`. Similar to `Client.subscribe()`, `Topic.matches()` also accepts wildcards (e.g. `temperature/#`):

    import asyncio
    import aiomqtt

    async def main():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            await client.subscribe("humidity/#")
            async for message in client.messages:
                if message.topic.matches("humidity/inside"):
                    print("A:", message.payload)
                if message.topic.matches("+/outside"):
                    print("B:", message.payload)
                if message.topic.matches("temperature/#"):
                    print("C:", message.payload)

    asyncio.run(main())

Note

In our example, messages to `temperature/outside` are handled twice!

Tip

For details on the `+` and `#` wildcards and what topics they match, see the OASIS specification.

## The message queue

Messages are queued internally and returned sequentially from `Client.messages`.

The default queue is `asyncio.Queue` which returns messages on a FIFO (“first in first out”) basis. You can pass other types of asyncio queues as `queue_type` to the `Client` to modify the order in which messages are returned, e.g. `asyncio.LifoQueue`.

You can subclass `asyncio.PriorityQueue` to queue based on priority. Messages are returned ascendingly by their priority values. In the case of ties, messages with lower message identifiers are returned first.

Let’s say we measure temperature and humidity again, but we want to prioritize humidity:

    import asyncio
    import aiomqtt

    class CustomPriorityQueue(asyncio.PriorityQueue):
        def _put(self, item):
            priority = 2
            if item.topic.matches("humidity/#"):  # Assign priority
                priority = 1
            super()._put((priority, item))

        def _get(self):
            return super()._get()[1]

    async def main():
        async with aiomqtt.Client(
            "test.mosquitto.org", queue_type=CustomPriorityQueue
        ) as client:
            await client.subscribe("temperature/#")
            await client.subscribe("humidity/#")
            async for message in client.messages:
                print(message.payload)

    asyncio.run(main())

Tip

By default, the size of the queue is unlimited. You can set a limit through the client’s `max_queued_incoming_messages` argument.

## Processing concurrently

Messages are queued internally and returned sequentially from `Client.messages`. If a message takes a long time to handle, it blocks the handling of other messages.

You can handle messages concurrently by using multiple worker tasks like so:

    import asyncio
    import aiomqtt

    async def work(client):
        async for message in client.messages:
            await asyncio.sleep(5)  # Simulate some I/O-bound work
            print(message.payload)

    async def main():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            # Use a task group to manage and await all worker tasks
            async with asyncio.TaskGroup() as tg:
                for _ in range(2):  # You can use more than two workers here
                    tg.create_task(work(client))

    asyncio.run(main())

Important

Coroutines only make sense if your message handling is I/O-bound. If it’s CPU-bound, you should spawn multiple processes instead.

## Listening without blocking

When you run the minimal example for subscribing and listening for messages, you’ll notice that the program doesn’t finish. Waiting for messages through the `Client.messages()` generator blocks the execution of everything that comes afterward.

In case you want to run other code after starting your listener, this is not very practical.

You can use `asyncio.TaskGroup` (or `asyncio.gather` for Python `<3.11`) to safely run other tasks alongside the MQTT listener:

    import asyncio
    import aiomqtt

    async def sleep(seconds):
        # Some other task that needs to run concurrently
        await asyncio.sleep(seconds)
        print(f"Slept for {seconds} seconds!")

    async def listen():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            async for message in client.messages:
                print(message.payload)

    async def main():
        # Use a task group to manage and await all tasks
        async with asyncio.TaskGroup() as tg:
            tg.create_task(sleep(2))
            tg.create_task(listen())  # Start the listener task
            tg.create_task(sleep(3))
            tg.create_task(sleep(1))

    asyncio.run(main())

In case task groups are not an option (e.g. because you run aiomqtt alongside a web framework) you can start the listener in a fire-and-forget way. The idea is to use asyncio’s `create_task` without awaiting the created task:

Caution

You need to be a bit careful with this approach. Exceptions raised in asyncio tasks are propagated only when we `await` the task. In this case, we explicitly don’t.

You need to handle all possible exceptions inside the fire-and-forget task. Unhandled exceptions will be silently ignored until the program exits.

    import asyncio
    import aiomqtt

    async def listen():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            async for message in client.messages:
                print(message.payload)

    background_tasks = set()

    async def main():
        loop = asyncio.get_event_loop()
        # Listen for mqtt messages in an (unawaited) asyncio task
        task = loop.create_task(listen())
        # Save a reference to the task so it doesn't get garbage collected
        background_tasks.add(task)
        task.add_done_callback(background_tasks.remove)

        # Infinitely do something else (e.g. handling HTTP requests)
        while True:
            await asyncio.sleep(1)

    asyncio.run(main())

Important

You need to keep a reference to the task so it doesn’t get garbage collected mid-execution. That’s what we’re using the `background_tasks` set in our example for.

## Stop listening

You might want to have a listener task running alongside other code, and then stop it when you’re done. You can use `asyncio.Task.cancel()` for this:

    import asyncio
    import aiomqtt

    async def listen():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            async for message in client.messages:
                print(message.payload)

    async def main():
        loop = asyncio.get_event_loop()
        # Listen for MQTT messages in (unawaited) asyncio task
        task = loop.create_task(listen())
        # Do something else for a while
        await asyncio.sleep(5)
        # Cancel the task
        task.cancel()
        # Wait for the task to be cancelled
        try:
            await task
        except asyncio.CancelledError:
            pass

    asyncio.run(main())

## Stop listening after timeout

For use cases where you only want to listen to messages for a certain amount of time, we can use asyncio’s timeouts, which were introduced in Python `3.11`:

    import asyncio
    import aiomqtt

    async def listen():
        async with aiomqtt.Client("test.mosquitto.org") as client:
            await client.subscribe("temperature/#")
            async for message in client.messages:
                print(message.payload)

    async def main():
        try:
            # Cancel the listener task after 5 seconds
            async with asyncio.timeout(5):
                await listen()
        # Ignore the resulting TimeoutError
        except asyncio.TimeoutError:
            pass

    asyncio.run(main())