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:

Derive a class from AsyncDatagramRequestHandler and redefine its handle() method; this method will process incoming requests.
Instantiate the AsyncUDPNetworkServer class passing it the server’s address, the protocol object and the request handler instance.
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.api_async.server 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:
        await client.send_packet(InternalError())
    else:
        await client.send_packet(Response())

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

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: AsyncDatagramClient[Response],
) -> AsyncGenerator[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

    ...

    await client.send_packet(Response())

    try:
        async with asyncio.timeout(30):
            # The client has 30 seconds to send the 2nd request to the server.
            another_request: Request = yield
    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")

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


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