(
    ingest_workers: dict[str, Task[NoReturn]], queue: RedisWebhookQueue
)
Listen on Redis Pubsub and start queue worker if we don't have one already.
Listen on Redis Pubsub and start queue worker if we don't have one already.
Listen on Redis Pubsub and start queue worker if we don t have one already .
async def ingest_queue_starter(
    ingest_workers: dict[str, Task[NoReturn]], queue: RedisWebhookQueue
) -> None:
    """
    Listen on Redis Pubsub and start queue worker if we don't have one already.
    """
    redis = await redis_client.create_connection()
    subscriber = await redis.start_subscribe()
    await subscriber.subscribe([QUEUE_PUBSUB_INGEST])
    log = logger.bind(task="ingest_queue_starter")
    log.info("start watch for ingest_queues")
    while True:
        reply = await subscriber.next_published()
        installation_id = PubsubIngestQueueSchema.parse_raw(reply.value).installation_id
        queue_name = get_ingest_queue(installation_id)
        if queue_name not in ingest_workers:
            ingest_workers[queue_name] = asyncio.create_task(
                work_ingest_queue(queue, queue_name=queue_name)
            )
            log.info("started new task")
async def ingest_queue_starter ( ingest_workers : dict [ str , Task [ NoReturn ] ] , queue : RedisWebhookQueue ) -> None : redis = await redis_client . create_connection ( ) subscriber = await redis . start_subscribe ( ) await subscriber . subscribe ( [ QUEUE_PUBSUB_INGEST ] ) log = logger . bind ( task = "ingest_queue_starter" ) log . info ( "start watch for ingest_queues" ) while True : reply = await subscriber . next_published ( ) installation_id = PubsubIngestQueueSchema . parse_raw ( reply . value ) . installation_id queue_name = get_ingest_queue ( installation_id ) if queue_name not in ingest_workers : ingest_workers [ queue_name ] = asyncio . create_task ( work_ingest_queue ( queue , queue_name = queue_name ) ) log . info ( "started new task" )