[Python] How to await messages and not affect the main event loop?

I am writing a simple websocket script which registers & unregisters clients and then broadcasts random messages to them in a interval of 5s. This is the code:

import asyncio, websockets, random, string
import websockets.asyncio.server

class web_socket(websockets.asyncio.server.ServerConnection):
pass

connections: set[web_socket] = set()

connections_lock = asyncio.Lock()

async def register(websocket: web_socket):
async with connections_lock:
connections.add(websocket)

async def unregister(websocket: web_socket):
async with connections_lock:
connections.discard(websocket)

async def cleanup_connections():
while True:
async with connections_lock:
closed_clients = [client for client in connections if client.close_code == 1000]
for client in closed_clients:
connections.discard(client)
await asyncio.sleep(1)

async def handler(websocket:web_socket):
await register(websocket)
try:
await websocket.wait_closed()
except Exception as e:
print(f"Error: {e}")

async def random_messages():
while True:
message = '-'.join(random.choices(string.ascii_letters + string.digits, k=10))
async with connections_lock:
send_tasks = [client.send(message) for client in connections if client.close_code != 1000]
if send_tasks:
await asyncio.gather(*send_tasks)
print(f"Broadcasted message: {message}")
await asyncio.sleep(5)

async def respond_to_messages():
while True:
async with connections_lock:
if connections:
for client in connections:
message = await client.recv()
if message:
print(message)
else: await asyncio.sleep(1)

async def main():
async with websockets.serve(handler, "0.0.0.0", 8765):
asyncio.create_task(random_messages())
asyncio.create_task(cleanup_connections())
asyncio.create_task(respond_to_messages())
await asyncio.Future()

asyncio.run(main())


As soon as someone connects, the broadcasting stops and the loop gets stuck at the message await. How do I make sure that listening to messages sent by clients do not affect the main loop?

I tried this:

message = await asyncio.wait(client.recv())


Now the client can connect without stopping the main loop but it throws an error:


Task exception was never retrieved
future: <Task finished name='Task-5' coro=<respond_to_messages() done, defined at /home/j/socket-test/server.py:44> exception=TypeError('expect a list of futures, not coroutine')>
Traceback (most recent call last):
File "/home/j/socket-test/server.py", line 49, in respond_to_messages
message = await asyncio.wait(client.recv())
File "/usr/lib/python3.10/asyncio/tasks.py", line 366, in wait
raise TypeError(f"expect a list of futures, not {type(fs).__name__}")
TypeError: expect a list of futures, not coroutine
/usr/lib/python3.10/asyncio/base_events.py:1910: RuntimeWarning: coroutine 'Connection.recv' was never awaited
handle = None # Needed to break cycles when an exception occurs.
RuntimeWarning: Enable tracemalloc to get the object allocation traceback


It throws this error and keeps going. The client gets all the broadcast messages but whatever the client sends maybe isn't received but definitely doesn't get printed. I know that asycnio.wait() expects a list of futures, so I tried explicitly passing it like this: asyncio.wait([client.recv()]). Running this gave a deprecation warning: DeprecationWarning: The explicit passing of coroutine objects to asyncio.wait() is deprecated since Python 3.8, and scheduled for removal in Python 3.11. message = await asyncio.wait([client.recv()]) and the client must send a message every time to receive a single broadcast message.

If I do this: asyncio.wait([await client.recv()]), then I get raise TypeError('An asyncio.Future, a coroutine or an awaitable ' TypeError: An asyncio.Future, a coroutine or an awaitable is required error and the client must send a message to start receiving the broadcast and then it keeps coming.

I tried GPT but it just keeps telling me to use message = await client.recv() which was the original problem.

I also tried to run the original code with coroutine threadsafe:

async def main():
async with websockets.serve(handler, "0.0.0.0", 8765):
asyncio.create_task(random_messages())
asyncio.create_task(cleanup_connections())
loop = asyncio.get_running_loop()
asyncio.run_coroutine_threadsafe(respond_to_messages(), loop=loop)
await asyncio.Future()


While this did not produce any errors, the client still needs to send the first message to start receiving broadcast. If this is something that is the way to go, that would be good to know. Another thing to note: Only the first client has to send a message, other ones do not. They start receiving broadcast upon connecting without issues.

Any help is appreciated. If you can point me to any article or something, that's fine too. I am a beginner so please keep that in mind.

Continue reading...