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.

[DC1991Lau]

@radiachkik did it work? I'm thaving the same issue.

Thank you

[stronk7]

@radiachkik did it work? I'm thaving the same issue.

Thank you

I imagine that it will be something like this:

#@broker.subscriber(qu, ex)
async def programmatic_handler(msg, logger: Logger):
    logger.info("programmatic_handler with message: %s" % msg)

broker.subscriber(qu, ex)(programmatic_handler)

So, instead of the decorator (that I've left, but commented out), the method is used. Not sure how it works with multiple or mixed cases but, for the simple case above, it's working here.

Ciao :-)

PS: Yeah, I faced the same, and took some good time to get it working (python and faststream newbie here, heh).