RabbitMQ sender design

[BrikerMan]

Hi all, we have been run into RabbitMQ Async client issue. So we had to roll back to the RabbitMQ sync sender with pika.

But we need to discuss how to use the pika. Here are several way to use it.

1. Use one shared connect and channel for each FastAPI worker, and reuse them when sending message.
2. Use one shared connect for each FastAPI worker, create new channel for each message and close the channel after message sent.
3. Create one connect and one channel for each message and closing them after message sent.

Personally I prefer the solution 2. Let's discuss which one is more suitable for our use case.

Deadline: 2022-01-11.

[vietor]

Maybe use the HTTP API for send message, it's simple and thread safe.

Doc: https://www.cloudamqp.com/docs/http.html

code sample:

https://github.com/Yodo1Games/eva-core/blob/dev/core/tools/rabbit.py

    async def publish(cls, host: str, user: str, password: str, virtualhost: str, exchange: str, data: Dict, retry: int = 3) -> Tuple[bool, str]:
        target_url = 'https://' + host + '/api/exchanges/' + quote_plus(virtualhost) + '/' + exchange + '/publish'
        target_params = {
            'properties': {},
            'routing_key': '',
            'payload_encoding': 'string',
            'payload': json.dumps(data)
        }

        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30), auth=aiohttp.BasicAuth(user, password)) as session:
            try:

[Dustyposa]

I have two points.

1. direct use rabbit mq async client, if the issue cause by this one, please ignore this line :)
2. Use one shared connect for each FastAPI worker, create by(other way is ok, can make sure close the client, like db):

@app.on_event('startup')
async def startup(app):
    loop = asyncio.get_running_loop()
    app.request.mq_client = create_client(xxxxx)

Two reference:
first
second
Not recommend close channel after publish a message, the reason.

[BrikerMan]

@vietor @Dustyposa Please review #9, implemented HTTP sender.