Hook after subscriber initialisation

[pbonneaudiabolocom]

Hi,

I have a test that initialize my app, and when initialized(after startup hook called) push a message to redis stream in order to validate the system is working as expected.

My issue is that after_startup mean that the broker started, but nothing says the subscribers are all registred and usable.

Would it be possible to add a hook after subscribers are initialized ?

Regards,
Pierre

[Lancetnik]

@pbonneaudiabolocom broker.start() is a method that waits for every registered subscriber to be started. So, the after_startup hook already guarantees you what you want.

[pbonneaudiabolocom]

Thanks.

I'll have to check because my test fails on a regular basis as soon as the stream mode is used with group.

@self.broker.subscriber(stream=StreamSub(stream=self.stream_name, group="call_activity_managers", consumer="1", last_id=">"))

And the test:

main_orchestrator = MainOrchestrator()
            await main_orchestrator.start_listening()
            # add an event to redis
            event = CallEvent(
                event_type=EVENT_NEW_CALL,
                id=str(uuid.uuid4()),
                timestamp=1,
                data={
                    "interaction_id": "1",
                    "diabolocom_account_uuid": "1",
                    "account_id": "1",
                    "configuration_id": "1",
                    "allowed_users_ids": [1, 1],
                }
            )
            await redis_broker.publish(event, stream=MainOrchestrator.stream_name)
            await asyncio.sleep(1)
            mock_start_call.assert_awaited_once()

Subscriber is started by MainOrchestrator init, and start listening is as such:

    async def start_listening(self):
        await self.app.start()
        await self.startup_complete.wait() # that asyncio evet is set in after_startup hook
        logger.info("Listenning to redis started")

With a local redis docker.

[pbonneaudiabolocom]

It was an issue with my redis group being corrupted.

I added a reseting line in my test and it made the job.