Add and remove subscribers at runtime

[pbonneaudiabolocom]

Hi,

I have an application that consumes differents redis sources in order to manage calls.

1 redis stream is receiving start_call and stop_call events

For each call, I need to start a new subscriber, as each call events will be pushed to a specific redis list.
At some point, that call will end, and I can close it fully, as it will never be up again(the customer hang up)

Today, I buit something working:
1 faststream app for my call_activity_manager, with 1 subscriber of stream type.
For each new call, I start a new fastsstream app, with a subscriber pointing to the right redis list.
When the call end, I can close faststream and normally, everything should be cleaned up.

It works.

But, I'm under the impression that it's not optimal. I would prefer to have a single faststream app, with one main subscriber available from start, and subscribers per call added on demand, and removed when the call finishes.

Questions:

• Is it the right direction ?
• I will have my app running forever, so as I don't see a way to remove / stop a subscriber, how can I prevent a memory leak ?
• In case it's not a good idea, and my current way is better, how can we stop an app, without closing it's broker ? It's not convenient when that broker is a singleton. I overided stop(), but the logic will change a bit in 0.6, so I'm not sure how it could be handled.

Thank you,
Pierre

[Lancetnik]

@pbonneaudiabolocom hi! Did you check dynamic subscribers documentation section? It looks like a feature for your case

[pbonneaudiabolocom]

Hi,

Thank you for your answer.
Sorry for my late reply, I was pushed on other piority.

I tried that approach, but it does't seem present on 5.0 yet.

On RC4, I ended up having plenty of issues with the event parsing to get my pydantic object as before.

I'll wait until the situation is stabilized with the Message management before moving to that new solution.

Regards,
Pierre

[Lancetnik]

@pbonneaudiabolocom sorry, but we don't plan to make a new RC before stable release. 0.6 is totally ready and scheduled to publish this Friday already. So, please, feel free to open any Issues about your problems with 0.6 migration

[pbonneaudiabolocom]

Hi,

I'll do that then.

Thank you

[pbonneaudiabolocom]

OK, after investigating the bug, I found out that it's because I'm using JSONFormat, which is not default comportment anymore.

Question would be... how are we supposed to transform subscriber message to pydantic one ?

I trued something like that... but no luck.

            async for msg in self.subscriber:
                msg:RedisListMessage = msg
                event = CallEvent(**self.broker.decode(msg.body)) # decode doesn't exist, _decoder is None
                self.state.new_event_received()
                logger.info('Received event "%s"', event.event_type)

[Lancetnik]

@pbonneaudiabolocom seems like it's the same bug with #2554
I hope to fix it and release tomorrow (with JsonFormat fallback)

[pbonneaudiabolocom]

in the meantime, I did that:

msg:RedisListMessage = msg
                data = json.loads(msg.body.decode())
                event = CallEvent(**data)

[pbonneaudiabolocom]

Hi,

I'm trying to use these subscribers now that 0.6 is live.

I don't know if the documentation is up to date, but it doesn't seem to work for me. But I don't really understand from the example how you are supposed to make it work.

The documentation says, in a nutshell.

1. declare the subscriber
2. start it
3. iterate over it to get messages
4. stop it.

But in the end, I don't see how it could integrate with a queueing mechanism.

in my case I:
initialize the subscriber
start it
keep going with my code.

In parralel, I'm expecting the subscriber to execute some code for each message until I stop it.

self.subscriber = self.broker.subscriber(list=f"{self.call_id}:events", persistent=False)

async def consume_events(self):
    logger.info("Started event consumer")
    async for msg in self.subscriber:
        msg:RedisListMessage = msg
        data = json.loads(msg.body.decode())
        event = CallEvent(**data)
        self.state.new_event_received()
        if event.event_type == EVENT_STOP_CALL:
            self.stop_event.set()
        else:
            self.state.append_event(event)
        logger.info('Received event "%s"', event.event_type)
        await msg.ack()
    logger.info("Stopping event consumer")

self.subscriber(self.consume_events)
await self.subscriber.start()
# do some other stuff

Somehow, I'm trying to reproduce the handler mechanism that I have with a classic fast stream app where you declare it and it will manage itself properly.

        @self.broker.subscriber(list=f"{self.call_id}:events")
        async def handle_batch(event: CallEvent):
            self.state.new_event_received()
            if event.event_type == EVENT_STOP_CALL:
                self.stop_event.set()
            else:
                self.state.append_event(event)
            logger.info('Received event "%s"', event.event_type)

Doesn't make much sence for me from the doc how I'm supposed to arrive somewhere with dynamic subscriber.

Any hint ?

Regards,
Pierre

[pbonneaudiabolocom]

Ok, found it:

        async def consume_events(msg:CallEvent):
            logger.info("Started event consumer")

        try:

            subscriber = self.broker.subscriber(list=f"{self.call_id}:events")
            subscriber(consume_events)
            await subscriber.start()
            logger.info("Started call manager")
            

It's working as expected.