python – Rabbitmq Stream queues consumers always inactive

I am using aio_pika in python to consume from a stream queue and I want to get the number of consumers that consume from the stream queue but I always get the same result of 0 consumers. I tried: using multiple consumers, using pika instead of aio_pika and changing the consumer messages offset but the result stayed the same.

After using the RMQ management UI I saw that the consumer count is always zero no matter how many consumers are consuming because they are all considered inactive
despite the fact I managed to consume messages. My goal is to somehow make the consumers active to see the right amount of consumers consuming from the stream queue.

I am adding my python code and a screenshot of the stream queue in the RMQ management UI.

My script:

import asyncio
import aio_pika
from aio_pika.abc import AbstractIncomingMessage


async def consume_message(message: AbstractIncomingMessage) -> None:
    async with message.process():
        print(f"Received message: {message.body.decode()}")


async def main():
    rabbitmq_url = "amqp://guest:guest@localhost/"

    connection = await aio_pika.connect_robust(rabbitmq_url)
    async with connection:
        channel = await connection.channel()
        await channel.set_qos(prefetch_count=5)

        queue_name = "stream_queue"
        await channel.declare_queue(
            queue_name,
            durable=True,
            arguments={
                "x-queue-type": "stream"
            }
        )

        print(f"Listening for messages on queue '{queue_name}'...")

        queue = await channel.get_queue(queue_name)
        await queue.consume(consume_message)
        queue_status = await queue.declare()
**        print(f"Consumer count: {queue_status.consumer_count}")  # Result: "Consumer count: 0"**
        await asyncio.Future()


if __name__ == "__main__":
    asyncio.run(main())

Stream queue and an inactive consumer in RMQ UI:

Stream queue and an inactive consumer in RMQ UI

I tried declaring a stream queue with messages and consume from it and publish more messages to it and despite the consumer getting the messages it was still considered inactive despite it receiving messages.