An Echo Client/Server Over UDP

It is possible to do the same as the TCP server tutorial with UDP sockets.

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.

---

The Communication Protocol

You will need a protocol object, as for the echo client/server over TCP.

For the tutorial, JSONSerializer will also be used.

For communication via UDP, a DatagramProtocol object must be created this time.

json_protocol.py

from __future__ import annotations

from typing import Any, TypeAlias

from easynetwork.protocol import DatagramProtocol
from easynetwork.serializers import JSONSerializer

# Use of type aliases in order not to see two Any types without real meaning
# In our case, any serializable object will be sent/received
SentDataType: TypeAlias = Any
ReceivedDataType: TypeAlias = Any


class JSONDatagramProtocol(DatagramProtocol[SentDataType, ReceivedDataType]):
    def __init__(self) -> None:
        super().__init__(JSONSerializer())

The Server

Create Your Datagram Request Handler

First, you must create a request handler class by subclassing the AsyncDatagramRequestHandler class and overriding its handle() method; this method will process incoming requests.

echo_request_handler.py

from __future__ import annotations

from collections.abc import AsyncGenerator
from typing import Any, TypeAlias

from easynetwork.exceptions import DatagramProtocolParseError
from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler, INETClientAttribute

RequestType: TypeAlias = Any
ResponseType: TypeAlias = Any


class EchoRequestHandler(AsyncDatagramRequestHandler[RequestType, ResponseType]):
    async def handle(
        self,
        client: AsyncDatagramClient[ResponseType],
    ) -> AsyncGenerator[None, RequestType]:
        try:
            request: RequestType = yield
        except DatagramProtocolParseError:
            await client.send_packet({"error": "Invalid JSON", "code": "parse_error"})
            return

        client_address = client.extra(INETClientAttribute.remote_address)
        print(f"{client_address.host} sent {request}")

        response: ResponseType = request
        await client.send_packet(response)

Note

There is no connection pipe with UDP, so there is no aclose() method. But the client object still has an is_closing() that returns True when the server itself closes.

Start The Server

Second, you must instantiate the UDP server class, passing it the server’s address, the protocol object instance, and the request handler instance.

SynchronousAsynchronous (asyncio)Asynchronous (trio)

server.py

from __future__ import annotations

from easynetwork.servers import StandaloneUDPNetworkServer

from echo_request_handler import EchoRequestHandler
from json_protocol import JSONDatagramProtocol


def main() -> None:
    host = None
    port = 9000
    protocol = JSONDatagramProtocol()
    handler = EchoRequestHandler()

    with StandaloneUDPNetworkServer(host, port, protocol, handler) as server:
        server.serve_forever()


if __name__ == "__main__":
    try:
        main()
    except* KeyboardInterrupt:
        pass

The Client

This is the client side:

SynchronousAsynchronous (asyncio)Asynchronous (trio)

client.py

from __future__ import annotations

import sys

from easynetwork.clients import UDPNetworkClient

from json_protocol import JSONDatagramProtocol


def main() -> None:
    host = "localhost"
    port = 9000

    # Connect to server
    with UDPNetworkClient((host, port), JSONDatagramProtocol()) as client:
        # Send data
        request = {"command-line arguments": sys.argv[1:]}
        client.send_packet(request)

        # Receive data from the server and shut down
        response = client.recv_packet()

    print(f"Sent:     {request}")
    print(f"Received: {response}")


if __name__ == "__main__":
    main()

Note

This is a “spoofed” connection. In fact, the socket is saving the address and will only send data to that endpoint.

Outputs

The output of the example should look something like this:

Server:

IPv4 connectionIPv6 connection

(.venv) $ python server.py
127.0.0.1 sent {'command-line arguments': ['Hello', 'world!']}
127.0.0.1 sent {'command-line arguments': ['Python', 'is', 'nice']}

Client:

(.venv) $ python client.py Hello world!
Sent:     {'command-line arguments': ['Hello', 'world!']}
Received: {'command-line arguments': ['Hello', 'world!']}
(.venv) $ python client.py Python is nice
Sent:     {'command-line arguments': ['Python', 'is', 'nice']}
Received: {'command-line arguments': ['Python', 'is', 'nice']}