How to wrap broker in a class?

[gaby]

I'm trying to figure out how to encapsulate the broker into a class. I noticed most of the examples are written in a way that assumes you will be using the faststream cli.

Is something like this supported?

from faststream import FastStream
from faststream.annotations import Logger
from faststream.rabbit import RabbitBroker

class Consumer:
	def __init__(self):
		self.broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
		self.app = FastStream(broker)
	
		@self.broker.subscriber("test-queue")
		async def handle(msg, logger: Logger):
		    logger.info(msg)

	async consume(self):
		await self.app.run()

Is there a way to pass a consumer callback function to the broker without using a decorator? Because of the decorator I can't use @self.broker as a regular function. I'm not sure/can't find in the docs what's the correct way starting/doing something like this?

Do I need to call connect() / run() on the broker, or does the app.run() do it for me?

[Lancetnik]

You can declare subscribers anywhere with router and include them to broker lately
https://faststream.airt.ai/latest/getting-started/routers/

[gaby]

That solves my problem, Thank you!

TLDR: https://faststream.airt.ai/latest/getting-started/routers/#delay-handler-registration

[flipyap]

@gaby do you have a minimal example of what it looks like to work out the router config for each queue/class method. I started doing this in my init and it got out of hand quickly.

[gaby]

@flipyap I only have one queue.

My actual problem was that the TestBroker if you call publish without a queue param, it doesnt fail or throws a warning.

@Lancetnik Not sure if thats a bug or feature. But the queue param in TestBrokers shouldnt be optional

[Lancetnik]

@flipyap I only have one queue.

My actual problem was that the TestBroker if you call publish without a queue param, it doesnt fail or throws a warning.

@Lancetnik Not sure if thats a bug or feature. But the queue param in TestBrokers shouldnt be optional

I should test, but in 0.6.0 version TestBroker throws an error if there no suitable subscriber - it should solve the problem

[gaby]

Gotcha, I'm using v0.5.34