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 asyncio
,
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.
1from __future__ import annotations
2
3from typing import Any, TypeAlias
4
5from easynetwork.protocol import DatagramProtocol
6from easynetwork.serializers import JSONSerializer
7
8# Use of type aliases in order not to see two Any types without real meaning
9# In our case, any serializable object will be sent/received
10SentDataType: TypeAlias = Any
11ReceivedDataType: TypeAlias = Any
12
13
14class JSONDatagramProtocol(DatagramProtocol[SentDataType, ReceivedDataType]):
15 def __init__(self) -> None:
16 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.
1from __future__ import annotations
2
3from collections.abc import AsyncGenerator
4from typing import Any, TypeAlias
5
6from easynetwork.exceptions import DatagramProtocolParseError
7from easynetwork.servers.handlers import AsyncDatagramClient, AsyncDatagramRequestHandler, INETClientAttribute
8
9RequestType: TypeAlias = Any
10ResponseType: TypeAlias = Any
11
12
13class EchoRequestHandler(AsyncDatagramRequestHandler[RequestType, ResponseType]):
14 async def handle(
15 self,
16 client: AsyncDatagramClient[ResponseType],
17 ) -> AsyncGenerator[None, RequestType]:
18 try:
19 request: RequestType = yield
20 except DatagramProtocolParseError:
21 await client.send_packet({"error": "Invalid JSON", "code": "parse_error"})
22 return
23
24 client_address = client.extra(INETClientAttribute.remote_address)
25 print(f"{client_address.host} sent {request}")
26
27 response: ResponseType = request
28 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.
1from __future__ import annotations
2
3from easynetwork.servers import StandaloneUDPNetworkServer
4
5from echo_request_handler import EchoRequestHandler
6from json_protocol import JSONDatagramProtocol
7
8
9def main() -> None:
10 host = None
11 port = 9000
12 protocol = JSONDatagramProtocol()
13 handler = EchoRequestHandler()
14
15 with StandaloneUDPNetworkServer(host, port, protocol, handler) as server:
16 try:
17 server.serve_forever()
18 except KeyboardInterrupt:
19 pass
20
21
22if __name__ == "__main__":
23 main()
1from __future__ import annotations
2
3import asyncio
4
5from easynetwork.servers import AsyncUDPNetworkServer
6
7from echo_request_handler import EchoRequestHandler
8from json_protocol import JSONDatagramProtocol
9
10
11async def main() -> None:
12 host = None
13 port = 9000
14 protocol = JSONDatagramProtocol()
15 handler = EchoRequestHandler()
16
17 async with AsyncUDPNetworkServer(host, port, protocol, handler) as server:
18 try:
19 await server.serve_forever()
20 except asyncio.CancelledError:
21 pass
22
23
24if __name__ == "__main__":
25 asyncio.run(main())
The Client
This is the client side:
1from __future__ import annotations
2
3import sys
4
5from easynetwork.clients import UDPNetworkClient
6
7from json_protocol import JSONDatagramProtocol
8
9
10def main() -> None:
11 host = "localhost"
12 port = 9000
13
14 # Connect to server
15 with UDPNetworkClient((host, port), JSONDatagramProtocol()) as client:
16 # Send data
17 request = {"command-line arguments": sys.argv[1:]}
18 client.send_packet(request)
19
20 # Receive data from the server and shut down
21 response = client.recv_packet()
22
23 print(f"Sent: {request}")
24 print(f"Received: {response}")
25
26
27if __name__ == "__main__":
28 main()
1from __future__ import annotations
2
3import asyncio
4import sys
5
6from easynetwork.clients import AsyncUDPNetworkClient
7
8from json_protocol import JSONDatagramProtocol
9
10
11async def main() -> None:
12 host = "localhost"
13 port = 9000
14
15 # Connect to server
16 async with AsyncUDPNetworkClient((host, port), JSONDatagramProtocol()) as client:
17 # Send data
18 request = {"command-line arguments": sys.argv[1:]}
19 await client.send_packet(request)
20
21 # Receive data from the server and shut down
22 response = await client.recv_packet()
23
24 print(f"Sent: {request}")
25 print(f"Received: {response}")
26
27
28if __name__ == "__main__":
29 asyncio.run(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:
(.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']}
(.venv) $ python server.py
::1 sent {'command-line arguments': ['Hello', 'world!']}
::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']}