Died workers

[bgervan]

If a worker dies, the subscribers in it can be restored (Redis streams)? (Dynamic subscribes)

[Lancetnik]

You should use consumer groups for it: https://faststream.airt.ai/latest/redis/streams/groups/

[theobouwman]

is using f'worker-{uuid.uuid4()}' as the consumer= parameter correct for when I want to have multiple instances of FastStream distributing the workload ?

BROKER_CONSUMER_NAME = f'worker-{uuid.uuid4()}'


@broker.subscriber(
    stream=StreamSub(
        "scheduled.bimonthly.organisation.email.summary.send", 
        group="scheduled.bimonthly.organisation.email.summary.send", 
        consumer=BROKER_CONSUMER_NAME
    )
)
@inject
async def handler(organisation_service: Annotated[OrganizationService, FromDishka()],):
    organisations = await organisation_service.get_organizations()
...

[Lancetnik]

is using f'worker-{uuid.uuid4()}' as the consumer= parameter correct for when I want to have multiple instances of FastStream distributing the workload ?

BROKER_CONSUMER_NAME = f'worker-{uuid.uuid4()}'


@broker.subscriber(
    stream=StreamSub(
        "scheduled.bimonthly.organisation.email.summary.send", 
        group="scheduled.bimonthly.organisation.email.summary.send", 
        consumer=BROKER_CONSUMER_NAME
    )
)
@inject
async def handler(organisation_service: Annotated[OrganizationService, FromDishka()],):
    organisations = await organisation_service.get_organizations()
...

You should use the same group for concurrent message distribution. If you are using different group names - Redis recognize them as different consumers and send them message duplicates.

[theobouwman]

So to handle message concurrently use the same group name, and a unique consumer name for each instance?

[Lancetnik]

So to handle message concurrently use the same group name, and a unique consumer name for each instance?

Yes