Get latest message from websockets queue in Python

Question

How do you get the last message received from server even if there are unread messages in queue?

Also, how could I ignore (delete) the rest of the unread messages?

CODE example:

while True:
    msg = await ws_server.recv()
    await do_something_with_latest_message(msg)

I Nead something like:

while True:
    msg = await ws_server.recv_last_msg() # On next loop I should "await" until a newer msg comes, not te receive the previous msg in LIFO order
    await do_something_with_latest_message(msg)

pigrammer · Accepted Answer · 2022-08-26 20:04:49Z

1

There's no way to do this natively, just with the websockets library. However, you could use an asyncio.LifoQueue:

queue = asyncio.LifoQueue()

async def producer(ws_server):
    async for msg in ws_server:
        await queue.put(msg)

async def consumer():
    while True:
        msg = await queue.get()
        # clear queue
        while not queue.empty():
            await queue.get()
        await do_something_with_latest_message(msg)
await asyncio.gather(producer(ws_server), consumer())

answered Aug 26, 2022 at 20:04

pigrammer

3,6592 gold badges15 silver badges30 bronze badges

Sign up to request clarification or add additional context in comments.

4 Comments

Oliver Mohr Bonometti Over a year ago

I can't test it right now, but it seems like a solution

Oliver Mohr Bonometti Over a year ago

Second time the code goes through async for msg in ws_server: in producer function it raises this error

RuntimeError: Task <Task pending name='Task-11' coro=<consumer() running at xxx\main.py:139> cb=[_gather.<locals>._done_callback() at xxx\asyncio\tasks.py:767]> got Future <Future pending> attached to a different loop

. Line 139 referenced in the error is the msg = await queue.get() of the consumer function.

pigrammer Over a year ago

This should solve your issue.

Oliver Mohr Bonometti Over a year ago

Nevermind, it's working. I call the await asyncio.gather(prod, cons) instruction inside an async with websockets.connect context manager, but the asyncio.LifoQueue() outside of it. I put the queue declaration inside the context manager and passed it as parameter to the producer and consumer and it's working now.

Collectives™ on Stack Overflow

Get latest message from websockets queue in Python

1 Answer 1

4 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

4 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related