-
Notifications
You must be signed in to change notification settings - Fork 93
/
udp_server.py
40 lines (33 loc) · 1.28 KB
/
udp_server.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import trio
from .context import Context
from ..config import Config
from ..events import Event, RawData
from ..typing import ASGIFramework
from ..utils import parse_socket_addr
MAX_RECV = 2 ** 16
class UDPServer:
def __init__(
self,
app: ASGIFramework,
config: Config,
socket: trio.socket.socket,
nursery: trio._core._run.Nursery,
) -> None:
from ..protocol.quic import QuicProtocol # h3/Quic is an optional part of Hypercorn
self.app = app
self.config = config
self.nursery = nursery
self.socket = trio.socket.from_stdlib_socket(socket)
context = Context(nursery)
server = parse_socket_addr(socket.family, socket.getsockname())
self.protocol = QuicProtocol(self.app, self.config, context, server, self.protocol_send)
async def run(
self, task_status: trio._core._run._TaskStatus = trio.TASK_STATUS_IGNORED
) -> None:
task_status.started()
while True:
data, address = await self.socket.recvfrom(MAX_RECV)
await self.protocol.handle(RawData(data=data, address=address))
async def protocol_send(self, event: Event) -> None:
if isinstance(event, RawData):
await self.socket.sendto(event.data, event.address)