Is it possible to create subscribers (subscribe and unsubscribe) dynamically at runtime?

[radiachkik]

I want to get messages messages for a specific user that connected to a FastAPI websocket. Upon connecting to the websocket, I'd like to create a non-durable queue and an associated subscriber.

I already have all the required functionality completed with aio_pika, but for testing purposes I wanted to give FastStream a try.

The regular subscribers seem to work like charm, but I can't get my head around dynamically creating subscribers with FastStream.

I can imagine this concept also conflicting with the generation of the AsyncAPI documentation. Nevertheless, I tried a few things, but failed at "runtimeWarning: You are trying to register handler with already running broker. It has no effect until broker restarting.".

Any comments on that would be highly appreciated, as I'd really love to make use of the abstractions provided by FastStream.

[Lancetnik]

You can still create subscriber dynamically and use broker.start() to start all new subscribers, but I am working on explicit subscriber.start() method.

[radiachkik]

Thank you very much for the quick response! I will try it with broker.start() and wait for the subscriber.start() method. 👍

[radiachkik]

Is there a feature request or some issue I can watch to get notified about the subscriber,start() method? Because as it seems right now, using broker.start() will not allow me to dynamically unregister the subscribers. Or am I missing something?

[Lancetnik]

Indeed, you can't unregister subscribers, but u can raise StopConsume exception inside

[radiachkik]

Okay thank you. I will have a look at it and see whether it will be sufficient for my use case.