Consuming messages from a asyncio.Queue inside a WebSocket

Questions and Help
#1

Hello!

I’m trying to notify a client the progress of a synchronous process through WebSocket, but I’m having issues with it, it looks like all the messages are sent only after the HTTP request is completed. This is a simplified snippet

from sanic import Sanic
from asyncio import Queue
from sanic import Request, Websocket
from sanic.response import json
from sanic.log import logger
app = Sanic(__name__)

@app.listener("after_server_start")
async def after_server_start_listener(app: Sanic):
    app.ctx.queue = Queue(maxsize=10)

# Clients connect to the feed through a Vue application
@app.websocket("/feed")
async def feed(request: Request, ws: Websocket):
    while True:
        data = await request.app.ctx.queue.get()
        await ws.send(data)
        logger.info("Data sent")

@app.route("/process")
async def compute_process(request: Request):

    def socket_cb(message):
        request.app.ctx.queue.put_nowait(message)

    # this foo method must be synchronous, it belongs to a 3rd party library
    def foo(cb):
        for i in range(9):
            cb('Processing ' + str(i))
            logger.info("Data added to queue")

    foo(socket_cb)

    return json({"msg": "ok"})


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, dev=True, debug=True)

These are the logs produced:

[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data added to queue
[2023-01-12 12:34:52 +0000] - (sanic.access)[INFO][127.0.0.1:52194]: GET http://localhost:8080/process  200 12
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 0' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 1' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 2' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 3' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 4' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 5' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 6' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 7' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:52 +0000] [45903] [DEBUG] > TEXT 'Processing 8' [12 bytes]
[2023-01-12 12:34:52 +0000] [45903] [INFO] Data sent
[2023-01-12 12:34:58 +0000] [45903] [DEBUG] KeepAlive Timeout. Closing connection.

What I expect is something like this:

[INFO] Data added to queue
[DEBUG] > TEXT 'Processing N' [12 bytes]
[INFO] Data sent

Is this possible?

Thanks

#2

Unfortunately you are a bit constrained. Because you need to run that in a sync operation, you are never yielding back to the loop. If you never yield back to the loop, then queue.get will block until the loop’s scheduler tells it to continue.

1 Like
#3

Thanks Adam. I will try a different strategy then :slightly_smiling_face: