RabbitMQ Routing Key Binding & Queue Declaration

[bnugent2]

Hi all, I'm evaluating this library for a tech spike (I'm really impressed with the usability so far) and I just wanted to confirm if some behaviour is intended with the RabbitMQ client or whether I'm calling something incorrectly.

When declaring a subscriber like:

@broker.subscriber("test", "test")
async def handler():
    logger.info("received message")

The client seems to automatically create the specified exchange and queue. However doing the same with a publisher like:

@broker.publisher("test", "test")
async def handler():
    logger.info("send message")

No exchange and queues are automatically created. I've also noticed that even when manually declaring a queue/exchange with declare_exchange/declare_queue no routing key is bound, even when specified.

Happy to create an issue if this is indeed a bug but just thought I'd confirm first. Thanks

[Lancetnik]

Hi! You can setup routing key as a field of RabbitQueue/Exchange object, but it has no sense for 'declare' method calling
Declare methods returns you a raw aio-pika objects, so you can bind them manually
Publisher should declare objects automatically too, I just missed this thing, sorry. I'll upload this fix evening

[Lancetnik]

@bnugent2 So, you can manually declare any queue/exchange in your on_startup event and bind them each other

from faststream import FastStream
from faststream.rabbit import RabbitBroker, RabbitExchange, RabbitQueue

broker = RabbitBroker()
app = FastStream(broker)

@app.after_startup
async def declare():
    queue = await broker.declare_queue(RabbitQueue("test"))
    exchange = await broker.declare_exchange(RabbitExchange("test"))

    await queue.bind(
        exchange,
        routing_key="custom-key",
    )

Also, with the 0.2.9 release, publishers will automatically register exchanges (but not queue) too

[davorrunje]

The 0.2.8 version has just been released, please check it out.

[betaboon]

i just stumbled upon this and i find it quite confusing that @publisher differs in behavior from @subscriber.
i would have expected the following:

• @subscriber("queue-name") -> automatically declare the queue
• @subscriber(RabbitQueue("queue-name", passive=True))-> do not automatically declare the queue
• • @publisher("queue-name") -> automatically declare the queue
• @publisher(RabbitQueue("queue-name", passive=True))-> do not automatically declare the queue

but if i get this right @publisher does not declare the queue?
is that expected? and if so, is there a good reason?

[Lancetnik]

Yes, this is planned behavior. Subscriber requires exchange and queue to work with both. But Publisher uses just an exchange (queue argument just a shortcut to routing_key argument), so it doesn't require a real queue object. So, to minimize side effects number all entities are declare only required to their work RMQ objects.

Subscriber creates queue and exchange both. Publisher creates exchange only.

[betaboon]

the maybe this line in the documentation should be expanded upon: https://github.com/airtai/faststream/blob/main/docs/docs/en/rabbit/declare.md?plain=1#L13

FastStream declares and validates all exchanges and queues using publishers and subscribers RabbitMQ objects.
This sentence seems to imply the behavior i described above.

[Lancetnik]

Sorry for that. Of course, we should edit this line!