Alternative — Unix Stream 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 either asyncio or trio, but you can use a different library thanks to the asynchronous backend engine API.

---

Introduction

Creating a Unix server requires several steps:

1. Derive a class from AsyncStreamRequestHandler and redefine its handle() method; this method will process incoming requests.

2. Instantiate the AsyncUnixStreamServer class passing it the server’s address, the protocol object and the request handler instance.

3. Call serve_forever() to process requests.

Writing coroutine functions is mandatory to use this server.

See also

PEP 492 — Coroutines with async and await syntax

The proposal to introduce native coroutines in Python with async and await syntax.

asyncio — Asynchronous I/O

If you are not familiar with async/await syntax, you can use the standard library to get started with coroutines.

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.servers.handlers 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

See also

PEP 525 — Asynchronous Generators

The proposal that expanded on PEP 492 by adding generator capabilities to coroutine functions.

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:
        # Runtime error. Log the error.
        traceback.print_exc()

        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.

Warning

You should always log or re-raise a bare Exception thrown in your generator.

except Exception:
    # Runtime error. Log the error.
    traceback.print_exc()

    await client.send_packet(InternalError())

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

Using yield (Recommended)Using with

It is possible to send the timeout delay to the parent task:

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

Deprecated since version 1.2: Yielding a timeout without using RecvParams. Will be removed in 2.0.

try:
    # The client has 30 seconds to send the request to the server.
    request: Request = yield 30.0
except TimeoutError:
    await client.send_packet(TimedOut())
else:
    await client.send_packet(Response())

Sending Packets with socket control messages

By using SocketAncillary, you can send SCM data. See the Unix manual page sendmsg(2) for details.

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

    ancillary = SocketAncillary()
    ancillary.add_fds([4])
    await client.send_packet_with_ancillary(Response(), ancillary)

Receiving Packets with socket control messages

By using RecvAncillaryDataParams and SocketAncillary, you can receive SCM data. See the Unix manual page recvmsg(2) for details.

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[RecvParams | None, Request]:
    ancillary = SocketAncillary()
    request: Request = yield RecvParams(recv_with_ancillary=RecvAncillaryDataParams(ancillary.update_from_raw))

    for message in ancillary.messages():
        match message:
            case SCMRights(fds):
                for fd in fds:
                    print(f"Received file descriptor: {fd}")
            case SCMCredentials(credentials):
                for ucred in credentials:
                    print(f"Received unix credential: {ucred}")

Tip

The default buffer size for this operation is approximately 8 KiB. However, you can customize this behavior.

from socket import CMSG_LEN

max_fds = 128
server = AsyncUnixStreamServer(
    "/var/run/app.sock",
    StreamProtocol(JSONSerializer()),
    SCMRecvRequestHandler(),
    ancillary_bufsize=CMSG_LEN(max_fds * 4),
)

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())

Client Metadata

The client’s metadata are available via UNIXClientAttribute:

async def handle(
    self,
    client: AsyncStreamClient[Response],
) -> AsyncGenerator[None, Request]:
    client_address = client.extra(UNIXClientAttribute.peer_name)

    request: Request = yield

    print(f"{client_address} sent {request}")

    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:

Using asyncioUsing trioUsing the AsyncBackend API

async def service_init(
    self,
    exit_stack: contextlib.AsyncExitStack,
    server: AsyncUnixStreamServer[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")

Low-Level Socket Operations

For low-level operations such as setsockopt(), the client and server objects expose the sockets through a SocketProxy:

async def service_init(
    self,
    exit_stack: contextlib.AsyncExitStack,
    server: AsyncUnixStreamServer[Request, Response],
) -> None:
    for sock in server.get_sockets():
        assert sock.getsockopt(SOL_SOCKET, SO_ACCEPTCONN) != 0

async def on_connection(self, client: AsyncStreamClient[Response]) -> None:
    # Enable SO_PASSCRED in order to use SCMCredentials
    client.extra(UNIXClientAttribute.socket).setsockopt(SOL_SOCKET, SO_PASSCRED, 1)

Per-client variables (contextvars integration)

If your asynchronous framework supports per-task context variables, you can use this feature in your request handler:

class ClientContextRequestHandler(AsyncStreamRequestHandler[Request, Response]):
    client_addr_var: ClassVar[contextvars.ContextVar[UnixSocketAddress]]
    client_addr_var = contextvars.ContextVar("client_addr")

    @classmethod
    def client_log(cls, message: str) -> None:
        # The address of the currently handled client can be accessed
        # without passing it explicitly to this function.

        logger = logging.getLogger(cls.__name__)

        client_address = cls.client_addr_var.get()

        logger.info("From %s: %s", client_address, message)

    async def on_connection(
        self,
        client: AsyncStreamClient[Response],
    ) -> None:
        address = client.extra(UNIXClientAttribute.peer_name)
        self.client_addr_var.set(address)

        # In any code that we call within "handle()" is now possible to get
        # client's address by calling 'client_addr_var.get()'.

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

        self.client_log(f"Received request: {request!r}")

        await client.send_packet(Response())
        await client.send_packet(Response())

Tip

It is possible to initialize the context to be copied in service_init().

This means that the contextvars.ContextVar.set() calls made in service_init() will be applied to subsequent client tasks.

Warning

There is no context isolation between handle() calls.

This means that the contextvars.ContextVar.set() calls made in handle() are applied to whole client task.

Server Object

A basic example of how to run the server:

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator

from easynetwork.protocol import StreamProtocol
from easynetwork.servers.async_unix_stream import AsyncUnixStreamServer
from easynetwork.servers.handlers import AsyncStreamClient, AsyncStreamRequestHandler


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:
    path = "/var/run/app/app.sock"
    protocol = ServerProtocol()
    handler = MyRequestHandler()

    # Create the server, binding to /var/run/app/app.sock
    async with AsyncUnixStreamServer(path, 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())