How-to — UDP 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

Creating a UDP server requires several steps:

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

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

3. Call serve_forever() to process requests.

Request Handler Objects

Note

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

Here is a simple example:

from __future__ import annotations

from collections.abc import AsyncGenerator

from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler


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

    ...


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

    ...


class MyRequestHandler(AsyncDatagramRequestHandler[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: AsyncDatagramClient[Response],
    ) -> AsyncGenerator[None, Request]:
        # "client" a placeholder to have a stream-like API.
        # All the datagrams sent by this client are sent
        # through the "yield" statement.
        request: Request = yield

        # Do some stuff
        ...

        response = Response()

        # The corresponding call is server_socket.sendto(data, remote_address)
        await client.send_packet(response)

Using handle() Generator

Important

There will always be only one active generator per client. All the pending datagrams received while the generator is running are queued.

This behavior is designed to act like a stream request handler.

Minimum Requirements

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    ### Before 'yield'
    # Initializes the generator.
    # This is the setup part before receiving a request.
    # Unlike the stream request handler, the generator is started
    # when the datagram is received (but is not parsed yet).
    ##################

    request: Request = yield

    ### After 'yield'
    # The received datagram is parsed.
    # 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.
    # The server creates a new generator when a new datagram is received.
    #################
    return

Refuse datagrams

Your UDP socket can receive datagrams from anywhere. You may want to control who can send you information.

async def handle(
    self,
    client: AsyncDatagramClient[Response],
) -> AsyncGenerator[None, Request]:
    if not self.should_handle(client):
        # By returning before the "yield" statement, you ask the server to discard
        # the received datagram.
        return

    request: Request = yield

Error Handling

async def handle(
    self,
    client: AsyncDatagramClient[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 DatagramProtocolParseError:
        await client.send_packet(BadRequest())
    except Exception:
        # Most likely a bug in EasyNetwork code. Log the error.
        traceback.print_exc()

        await client.send_packet(InternalError())
    else:
        await client.send_packet(Response())

Warning

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

except Exception:
    # Most likely a bug in EasyNetwork code. Log the error.
    traceback.print_exc()

    await client.send_packet(InternalError())

Having Multiple yield Statements

async def handle(
    self,
    client: AsyncDatagramClient[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())

Warning

Even if this feature is supported, it is not recommended to have more than one (unless you know what you are doing) for the following reasons:

• UDP does not guarantee ordered delivery. Packets are typically “sent” in order, but they may be received out of order. In large networks, it is reasonably common for some packets to arrive out of sequence (or not at all).

• The server has no way of knowing if this client has stopped sending you requests forever.

If you plan to use multiple yields in your request handler, you should always have a timeout applied. (See the section below.)

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: AsyncDatagramClient[Response],
) -> AsyncGenerator[float | None, Request]:
    # It is *never* useful to have a timeout for the 1st datagram
    # because the datagram is already in the queue.
    request: Request = yield None

    ...

    await client.send_packet(Response())

    try:
        # The client has 30 seconds to send the 2nd request to the server.
        another_request: Request = yield 30
    except TimeoutError:
        await client.send_packet(TimedOut())
    else:
        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: AsyncUDPNetworkServer[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")

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(AsyncDatagramRequestHandler[Request, Response]):
    client_addr_var: ClassVar[contextvars.ContextVar[SocketAddress]]
    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 handle(
        self,
        client: AsyncDatagramClient[Response],
    ) -> AsyncGenerator[None, Request]:
        address = client.extra(INETClientAttribute.remote_address)
        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()'.

        request: Request = yield

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

        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.

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 DatagramProtocol
from easynetwork.servers import AsyncUDPNetworkServer
from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler


class Request: ...


class Response: ...


class MyRequestHandler(AsyncDatagramRequestHandler[Request, Response]):
    async def handle(
        self,
        client: AsyncDatagramClient[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(DatagramProtocol[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 AsyncUDPNetworkServer(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())

See also

An Echo Client/Server Over UDP

A working example of the server implementation.

Run Server In Background

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator
from typing import Any

from easynetwork.clients import AsyncUDPNetworkClient
from easynetwork.protocol import DatagramProtocol
from easynetwork.serializers import JSONSerializer
from easynetwork.servers import AsyncUDPNetworkServer
from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler


class JSONProtocol(DatagramProtocol[dict[str, Any], dict[str, Any]]):
    def __init__(self) -> None:
        super().__init__(JSONSerializer())


class MyRequestHandler(AsyncDatagramRequestHandler[dict[str, Any], dict[str, Any]]):
    async def handle(
        self,
        client: AsyncDatagramClient[dict[str, Any]],
    ) -> AsyncGenerator[None, dict[str, Any]]:
        request: dict[str, Any] = yield

        current_task = asyncio.current_task()
        assert current_task is not None

        await client.send_packet({"task": current_task.get_name(), "request": request})


async def client(host: str, port: int, message: str) -> None:
    async with AsyncUDPNetworkClient((host, port), JSONProtocol()) as client:
        await client.send_packet({"message": message})
        response = await client.recv_packet()
        print(f"From server: {response}")


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

    server = AsyncUDPNetworkServer(host, port, protocol, handler)

    async with server:
        is_up_event = asyncio.Event()
        server_task = asyncio.create_task(server.serve_forever(is_up_event=is_up_event))
        await is_up_event.wait()

        print(f"Server loop running in task: {server_task.get_name()}")

        await client(host, port, "Hello world 1")
        await client(host, port, "Hello world 2")
        await client(host, port, "Hello world 3")

        await server.shutdown()


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

The output of the example should look something like this:

$ python background_server.py
Server loop running in task: Task-2
From server: {'task': 'Task-6', 'request': {'message': 'Hello world 1'}}
From server: {'task': 'Task-7', 'request': {'message': 'Hello world 2'}}
From server: {'task': 'Task-8', 'request': {'message': 'Hello world 3'}}