how to limit the concurrency of @router.event('test')

[willnufe]

Thank you very much for making this project. I have some questions to ask you.

Considering the performance of downstream services, the number of files downloaded by message_comsumer function cannot be too large. Otherwise, downstream services will not be able to handle it, ultimately resulting in the disk being full. So how to limit the throughput of the message_comsumer function?

from propan.fastapi import RabbitRouter
from pydantic import BaseModel

import os
import uuid
import time
from typing import List

import uvicorn
from fastapi import FastAPI, Request, BackgroundTasks, Depends
from contextlib import asynccontextmanager

import aiohttp
import asyncio

from logger import log
from download_file_async import Audio_file_download


rabbit_url = 'amqp://guest:[email protected]/'
FILE_ENDPOINT = os.getenv("FILE_ENDPOINT")
TOKEN_ENDPOINT = os.getenv("TOKEN_ENDPOINT")
SESSION_POOL_D = int(os.getenv('SESSION_POOL_D', '10'))

RAW_AUDIO_DIR = os.getenv("RAW_AUDIO_DIR", '/archive/admin/raw_audio')
WAV_AUDIO_DIR = os.getenv("WAV_AUDIO_DIR", '/archive/admin/wav_audio')

PROCESS_NUM = int(os.getenv('PROCESS_NUM'))
LIMIT_CONCURRENCY = int(os.getenv('LIMIT_CONCURRENCY'))



class AppContext:
    def __init__(self):

        self.download = None


@asynccontextmanager
async def lifespan(app: FastAPI):

    app_ctx = AppContext()

    connector_d = aiohttp.TCPConnector(limit=20, verify_ssl=False)
    session_d = aiohttp.ClientSession(connector=connector_d)
    app_ctx.download = Audio_file_download(TOKEN_ENDPOINT, FILE_ENDPOINT, session_d)

    app.state.app_ctx = app_ctx
    print("worker waiting engine ready")


    yield


    app.state.app_ctx = None
    print("worker exited")



router = RabbitRouter(rabbit_url, lifespan=lifespan, log_level=3)
app = FastAPI()


class Message(BaseModel):
    file_id: str
    session_id: str
    url: str



class Message2(BaseModel):
    file_id: str
    session_id: str
    url: str
    error_code: int
    file_name: str



@router.event('test')
async def message_consumer(m: Message):


    try:
        file_id = m.file_id
        session_id = m.session_id
        url = m.url

        app_ctx = app.state.app_ctx

        ret = {
            'file_id': file_id,
            'session_id': session_id,
            'url': url,
            'error_code': 0,
            'file_name': ''
        }

        # s1--download file
        start_time_1 = int(round(time.time() * 1000))

        rand_uuid = str(uuid.uuid1())
        file_name = f"{file_id}_{rand_uuid}"
        ret['file_name'] = file_name

        raw_audio_file_path = f"{RAW_AUDIO_DIR}/{file_name}"
        download_res = await app_ctx.download.download_file_by_uuid(file_id, raw_audio_file_path)

        end_time_1 = int(round(time.time() * 1000))
        log.logger.info(
            f"[step1: download] {end_time_1 - start_time_1}ms || {file_id} || {session_id}"
        )

        if download_res is None:

            delete_files(file_path_list=[raw_audio_file_path])

            ret['error_code'] = 1001

            await router.broker.publish(Message2(**ret), "test_queue")
            return  

		await router.broker.publish(Message2(**ret), "test_queue")
        log.logger.info(f"[totalTimeS1] {end_time_3 - start_time_1}ms || {file_id} || {session_id}")  

    except Exception as e:
        log.logger.error(f"[messageConsumerInner]: {e}")





@router.post("/behavior/asr/")
async def generate(raw_request: Request):

    request_dict = await raw_request.json()

    file_id = request_dict['source']['image_uuid']
    session_id = request_dict['session']['session_id']
    url = request_dict['response_route']['url']


    ret = {
        'file_id': file_id,
        'session_id': session_id,
        'url': url
    }

    await router.broker.publish(Message(**ret), "test")

    return 




app.include_router(router)

if __name__ == "__main__":


    uvicorn.run(
        "__main__:app",
        host='0.0.0.0',
        port=8080,
        workers=4,
        limit_concurrency=20,
        reload=False
    )

[Lancetnik]

Hi! Use RabbitRouter(max_consumers=10) option. It sets RMQ qos channel parameter and limits maximum number of messages you can consume in the same time

[willnufe]

Hi! Use RabbitRouter(max_consumers=10) option. It sets RMQ qos channel parameter and limits maximum number of messages you can consume in the same time

Thank you very much for your reply.

I used router = RabbitRouter(rabbit_url, lifespan=lifespan, log_level=3, max_consumers=1) option based on the above code, but it doesn't seem to have any effect. My relevant configurations are as follows:

propan==0.1.5.26
fastapi==0.112.2

In addition, I checked the prop/fastapi/rabbit/outer.pyi file and found that it only has the consumers parameter, not the max_comsumers parameter.

I don't know if it's a problem with my version or if the file location I checked is incorrect.

[Lancetnik]

Seems like consumers is an option you need. Btw, why are you using Propan? I deprecated it almost a year ago in prior to FastStream

[willnufe]

Seems like consumers is an option you need. Btw, why are you using Propan? I deprecated it almost a year ago in prior to FastStream

Thank you very much for your reply.

I wrote a service that needs to use FastAPI and RabbitMQ. However, I'm not quite sure how to write it, so I looked up related issues and learned about Propan.

• https://github.com/mosquito/aio-pika/issues/558