Server implementation tools — Miscellaneous

Server implementation tools module.

---

Bridge between low-level and high-level APIs

easynetwork.servers.misc.build_lowlevel_stream_server_handler(initializer: Callable[[ConnectedStreamClient[_T_Response]], AbstractAsyncContextManager[AsyncStreamClient[_T_Response] | None]], request_handler: AsyncStreamRequestHandler[_T_Request, _T_Response], *, logger: Logger | None = None) → Callable[[ConnectedStreamClient[_T_Response]], AsyncGenerator[float | None, _T_Request]]

Creates an asynchronous generator function, usable by AsyncStreamServer.serve(), from an AsyncStreamRequestHandler.

Parameters:

• initializer (Callable[[ConnectedStreamClient[_T_Response]], AbstractAsyncContextManager[AsyncStreamClient[_T_Response] | None]]) – a callback returning an asynchronous context manager to create the final client interface and set up the request handler generator. The yielded value can be None if the initializer failed silently.

• request_handler (AsyncStreamRequestHandler[_T_Request, _T_Response]) – the high-level interface which handles the incoming requests.

• logger (Logger | None) – if given, will be used to log some warnings.

Returns:

an asynchronous generator function.

Return type:

Callable[[ConnectedStreamClient[_T_Response]], AsyncGenerator[float | None, _T_Request]]

easynetwork.servers.misc.build_lowlevel_datagram_server_handler(initializer: Callable[[DatagramClientContext[_T_Response, _T_Address]], AbstractAsyncContextManager[AsyncDatagramClient[_T_Response] | None]], request_handler: AsyncDatagramRequestHandler[_T_Request, _T_Response]) → Callable[[DatagramClientContext[_T_Response, _T_Address]], AsyncGenerator[float | None, _T_Request]]

Creates an asynchronous generator function, usable by AsyncDatagramServer.serve(), from an AsyncDatagramRequestHandler.

Parameters:

• initializer (Callable[[DatagramClientContext[_T_Response, _T_Address]], AbstractAsyncContextManager[AsyncDatagramClient[_T_Response] | None]]) – a callback returning an asynchronous context manager to create the final client interface and set up the request handler generator. The yielded value can be None if the initializer failed silently.

• request_handler (AsyncDatagramRequestHandler[_T_Request, _T_Response]) – the high-level interface which handles the incoming requests.

Returns:

an asynchronous generator function.

Return type:

Callable[[DatagramClientContext[_T_Response, _T_Address]], AsyncGenerator[float | None, _T_Request]]