asyncio socket client / server example

README.md

A basic example of using the higher level asyncio constructs such as streams in a server and client where both pass messages with regular cadence.

client.py

import asyncio

# # functional format
# async def read_loop(reader):
#     async for message in reader:
#         print(message.decode())


# async def write_loop(writer):
#     c = 0
#     while True:
#         msg = f'client sending {c}\r\n'
#         print(msg)
#         c += 1
#         writer.write(msg.encode())
#         await writer.drain()
#         await asyncio.sleep(1)


# async def client():
#     reader, writer = await asyncio.open_connection('localhost', 8888)
#     await asyncio.gather(read_loop(reader), write_loop(writer))


class Client:
    """A client connecting to the async server via a basic socket."""

    def __init__(self, host: str='localhost', port: int=8888):
        self.host = host
        self.port = port

    async def connect(self):
        reader, writer = await asyncio.open_connection(self.host, self.port)
        self.reader = reader
        self.writer = writer

    async def write(self, message: str | bytes):
        if isinstance(message, str):
            message = message.encode()
        self.writer.write(message + b'\r\n')
        await self.writer.drain()

    async def write_loop(self):
        """Simulate sending messages to the server at a regular cadence."""
        c = 0
        while True:
            msg = f'client sending {c}'
            print(msg)
            c += 1
            await self.write(msg)
            await asyncio.sleep(1)

    async def read_loop(self):
        """Read each message (terminated by a newline)"""
        async for message in self.reader:
            print(message.decode())


    async def run(self):
        await self.connect()
        await asyncio.gather(self.read_loop(), self.write_loop())

if __name__ == "__main__":
    client = Client()
    asyncio.run(client.run())

server.py

import asyncio

# # functional format
# async def client_connected(reader, writer):

#     addr = writer.get_extra_info('peername')
#     writer.write(f'you connected {addr}\r\n'.encode())
#     await writer.drain()

#     async def write_loop():
#         """Continuously ping client connection."""
#         c = 0
#         while True:
#             msg = f'server sending {c}\r\n'
#             print(msg)
#             c += 1
#             writer.write(msg.encode())
#             await writer.drain()
#             await asyncio.sleep(1)

#     async def read_loop():
#         async for message in reader:
#             print(message.decode())

#     asyncio.gather(write_loop(), read_loop())

class ClientConnectionHandler:
    """A class representing a client connection to the server."""

    async def read_loop(self):
        """Read messages up to \n and print."""
        async for msg in self.reader:
            print(f'< {msg}'.encode())

    async def write_loop(self):
        """Continuously write a basic increment to the client."""
        cnt = 0
        while True:
            msg = f'server sending {cnt}'
            print(f'> {msg}')
            cnt += 1
            self.writer.write(msg.encode() + b'\r\n')
            await self.writer.drain()
            await asyncio.sleep(1)

    async def callback(self, reader, writer):
        """The underlying connection handler callback used when a new
        client connects to the server.
        """
        self.reader = reader
        self.writer = writer

        addr = self.writer.get_extra_info('peername')
        self.writer.write(f'you connected {addr}\r\n'.encode())

        await writer.drain()
        await asyncio.gather(self.read_loop(), self.write_loop())


class Server:

    def __init__(self, host: str='localhost', port: int=8888):
        self.host = host
        self.port = port

    async def run(self):
        server = await asyncio.start_server(
            client_connected_cb=ClientConnectionHandler().callback,
            host=self.host,
            port=self.port,
        )
        addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
        print(f'Serving on {addrs}')
        await server.serve_forever()

if __name__ == '__main__':
    server = Server()
    asyncio.run(server.run())