How-to — TCP Servers

Note

This page uses two different API variants:

Synchronous API with classic def functions, usable in any context.
Asynchronous API with async def functions, using an asynchronous framework to perform I/O operations.

All asynchronous API examples assume that you are using asyncio, but you can use a different library thanks to the asynchronous backend engine API.

Introduction 

The easynetwork.api_async.server module simplifies the task of writing network servers. The service creation model is inspired by the standard socketserver library, but is an enhanced version with even more abstraction.

Creating a server requires several steps:

Derive a class from AsyncStreamRequestHandler and redefine its handle() method; this method will process incoming requests.
Instantiate the AsyncTCPNetworkServer class passing it the server’s address, the protocol object and the request handler instance.
Call serve_forever() to process requests.

Writing coroutine functions is mandatory to use this server.

Request Handler Objects 

Note

Unlike socketserver.BaseRequestHandler, there is only one AsyncStreamRequestHandler instance for the entire service.

Here is a simple example:

from __future__ import annotations

from collections.abc import AsyncGenerator

from easynetwork.api_async.server import AsyncStreamClient, AsyncStreamRequestHandler


class Request:
    """Object representing the client request."""

    ...


class Response:
    """Object representing the response to send to the client."""

    ...


class MyRequestHandler(AsyncStreamRequestHandler[Request, Response]):
    """
    The request handler class for our server.

    It is instantiated once to the server, and must
    override the handle() method to implement communication to the
    client.
    """

    async def handle(
        self,
        client: AsyncStreamClient[Response],
    ) -> AsyncGenerator[None, Request]:
        # "client" is the write stream of the connection to the remote host.
        # The read stream is covered by the server and the incoming
        # request is sent through the "yield" statement.
        request: Request = yield

        # Do some stuff
        ...

        response = Response()

        await client.send_packet(response)

Using `handle()` Generator 

Minimum Requirements 

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    ### Before 'yield'
    # Initializes the generator.
    # This is the setup part before receiving a request.
    # You don't have much thing to do here.
    ##################

    request: Request = yield

    ### After 'yield'
    # Once the server has sent you the client's request,
    # you can do whatever you want with it and send responses back
    # to the client if necessary.
    await client.send_packet(Response())
    #################

    ### On a 'return'
    # When handle() returns, it means that this request handler is finished.
    # It does not close the connection or anything.
    # The server immediately creates a new generator.
    #################
    return

Closing the connection 

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    request: Request = yield

    await client.send_packet(Response())

    # At this point, the transport is closed and the server
    # will not create a new generator.
    await client.aclose()

Tip

You can use contextlib.aclosing() to close the client at the generator exit.

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    async with contextlib.aclosing(client):
        request: Request = yield

        await client.send_packet(Response())

Important

The connection is forcibly closed under the following conditions:

handle() raises an exception.

handle() returns before the first yield statement.

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    if not self.should_handle(client):
        return

    request: Request = yield

Error Handling 

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    try:
        # *All* exceptions are thrown through the "yield" statement
        # (including BaseException). But you should only catch Exception subclasses.
        request: Request = yield
    except StreamProtocolParseError:
        await client.send_packet(BadRequest())
    except OSError:
        # It is possible that something went wrong with the underlying
        # transport (the socket) at the OS level.
        # You should check if the client is always usable.
        try:
            await client.send_packet(InternalError())
        except OSError:
            await client.aclose()
            raise
    except Exception:
        await client.send_packet(InternalError())
    else:
        await client.send_packet(Response())

Note

handle() will never get a ConnectionError subclass. In case of an unexpected disconnect, the generator is closed, so you should handle GeneratorExit instead.

Having Multiple `yield` Statements 

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    request: Request = yield

    ...

    await client.send_packet(Response())

    if self.need_something_else(request, client):
        additional_data: Request = yield

        ...

        await client.send_packet(Response())

Tip

The number of yield allowed is… infinite!

You can take advantage of this by having an internal main loop inside the generator:

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    # Close the client at the loop break
    async with contextlib.aclosing(client):
        # Ask the user to log in
        initial_user_info: Request = yield

        ...

        # Sucessfully logged in
        await client.send_packet(Response())

        # Start handling requests
        while not client.is_closing():
            request: Request = yield

            ...

            await client.send_packet(Response())

Cancellation And Timeouts 

Since all BaseException subclasses are thrown into the generator, you can apply a timeout to the read stream using the asynchronous framework (the cancellation exception is retrieved in the generator):

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    try:
        async with asyncio.timeout(30):
            # The client has 30 seconds to send the request to the server.
            request: Request = yield
    except TimeoutError:
        await client.send_packet(TimedOut())
    else:
        await client.send_packet(Response())

Connecting/Disconnecting Hooks 

You can override on_connection() and on_disconnection() methods:

on_connection() is called on client task startup.
on_disconnection() is called on client task teardown.

async def on_connection(self, client: AsyncStreamClient[Response]) -> None:
    print(f"{client!r} is connected")

    # Notify the client that the service is ready.
    await client.send_packet(Response())

async def on_disconnection(self, client: AsyncStreamClient[Response]) -> None:
    # Perfom service shutdown clean-up
    ...

    print(f"{client!r} is disconnected")

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    request: Request = yield

    ...

    await client.send_packet(Response())

Wait For Client Data On Connection 

If you need to use the read stream, on_connection() can be an asynchronous generator instead of a coroutine function:

async def on_connection(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    # Ask the user to log in
    initial_user_info: Request = yield

    ...

    # Sucessfully logged in
    await client.send_packet(Response())

async def on_disconnection(self, client: AsyncStreamClient[Response]) -> None:
    # Perfom log out clean-up
    ...

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    request: Request = yield

    ...

    await client.send_packet(Response())

Service Initialization 

The server will call service_init() and pass it an AsyncExitStack at the beginning of the serve_forever() task to set up the global service.

This allows you to do something like this:

async def service_init(
    self,
    exit_stack: contextlib.AsyncExitStack,
    server: AsyncTCPNetworkServer[Request, Response],
) -> None:
    exit_stack.callback(self._service_quit)

    self.background_tasks = await exit_stack.enter_async_context(asyncio.TaskGroup())

    _ = self.background_tasks.create_task(self._service_actions())

async def _service_actions(self) -> None:
    while True:
        await asyncio.sleep(1)

        # Do some stuff each second in background
        ...

def _service_quit(self) -> None:
    print("Service stopped")

Server Object 

A basic example of how to run the server:

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator

from easynetwork.api_async.server import AsyncStreamClient, AsyncStreamRequestHandler, AsyncTCPNetworkServer
from easynetwork.protocol import StreamProtocol


class Request:
    ...


class Response:
    ...


class MyRequestHandler(AsyncStreamRequestHandler[Request, Response]):
    async def handle(
        self,
        client: AsyncStreamClient[Response],
    ) -> AsyncGenerator[None, Request]:
        request: Request = yield

        ...

        await client.send_packet(Response())


# NOTE: The sent packet is "Response" and the received packet is "Request"
class ServerProtocol(StreamProtocol[Response, Request]):
    def __init__(self) -> None:
        ...


async def main() -> None:
    host, port = "localhost", 9000
    protocol = ServerProtocol()
    handler = MyRequestHandler()

    # Create the server, binding to localhost on port 9000
    async with AsyncTCPNetworkServer(host, port, protocol, handler) as server:
        # Activate the server; this will keep running until you
        # interrupt the program with Ctrl-C
        await server.serve_forever()


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