[Python] Implement a server and client communicating via sockets in Python

I need to implement a server and client communicating via sockets in Python. I decided to implement the server with asyncio.start_server. Clients connect to the server with socket connections. The server generates some messages at random moments and immediately pushes those messages to the client. All clients should receive the same messages as soon as possible and print them.

My implementation of server:

import asyncio
import datetime as dt
import struct
import sys

from random import randrange
from zoneinfo import ZoneInfo


async def prepare_weather_data():
while True:
await asyncio.sleep(randrange(0, 7))

now = dt.datetime.now(ZoneInfo("Europe/Kyiv")).replace(microsecond=0).isoformat()
massage = f"{now} The temperature is {str(randrange(20, 30))} degrees Celsius"

message_encoded = massage.encode()
message_length = struct.pack('>I', len(message_encoded))

yield message_length + message_encoded


async def read_massage(reader: asyncio.StreamReader):
# Read the 4-byte length header
length_data = await reader.readexactly(4)
message_length = struct.unpack('>I', length_data)[0]
message_encoded = await reader.readexactly(message_length)
return message_encoded.decode()


async def handle_client(
_: asyncio.StreamReader,
writer: asyncio.StreamWriter,
weather_data: bytes,
):
try:
writer.write(weather_data)
await writer.drain()
except KeyboardInterrupt:
print("Closing the connection")
writer.close()
sys.exit(0)


async def main():
async def client_handler(reader, writer):
message = await read_massage(reader)
print(f"Received the message \"{message}\" that won't be handled any way")

async for weather_data in prepare_weather_data():
await handle_client(reader, writer, weather_data)

server = await asyncio.start_server(client_handler, "localhost", 8000)
async with server:
await server.serve_forever()


if __name__ == "__main__":
asyncio.run(main())


The client:

import sys
import socket
import struct


def recvall(sock: socket.socket, n: int) -> bytes | None:
# Helper function to recv n bytes or return None if EOF is hit
data = bytearray()
while len(data) < n:
packet = sock.recv(n - len(data))
if not packet:
return None
data.extend(packet)
return data


def read_massage(sock: socket.socket) -> str | None:
# Read the 4-byte length
message_length_encoded = recvall(sock, 4)
if not message_length_encoded:
print("Socket closed")
sys.exit(0)

message_length = struct.unpack('>I', message_length_encoded)[0]
return recvall(sock, message_length).decode()


so = socket.socket(
socket.AF_INET,
socket.SOCK_STREAM,
)

so.connect(("localhost", 8000)) # Blocking
print("Connected")

try:
message = "Hello, give me a weather, please".encode()

# Pack the length of the message into 4 bytes (big-endian)
message_length = struct.pack('>I', len(message))
sent = so.send(message_length + message) # Blocking
print(f"Sent {sent} bytes")

while True:
msg = read_massage(so)
print(msg)

except KeyboardInterrupt:
so.close()
print("Socket closed")
sys.exit(0)


This version of the server generally works. The server generates messages and pushes them to the clients. However, all clients receive different messages! How to modify the server to push the same message to all clients? I tried everything I could.

UPD

I see two possible approaches.

1. 
   Store generated data in some container accessible to all clients. Like this:
   
   async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, weather_data):
   _ = await read_message(reader)
   try:
   while True:
   async for data in weather_data:
   writer.write(data)
   await writer.drain()
   except KeyboardInterrupt:
   print("Closing the connection")
   writer.close()
   sys.exit(0)
   
   async def main():
   weather_data = prepare_weather_data()
   
   async def client_handler(reader, writer):
   await handle_client(reader, writer, weather_data)
   
   server = await asyncio.start_server(client_handler, "localhost", 8000)
   async with server:
   await server.serve_forever()

However, it crashes for the second connected client as asynchronous generator is already running.

1. Register all connected clients and push generated data to them. I tried getting all connected clients as asyncio.all_tasks(). However, I didn't get how to use it.

Continue reading...