How-to — Standalone 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 either asyncio or trio, but you can use a different library thanks to the asynchronous backend engine API.

---

Introduction

Standalone servers are classes that can create ready-to-run servers with no predefined asynchronous context (i.e., no async / await). They use (and block) a thread to accept requests, and their methods are meant to be used by other threads for control (e.g. shutdown).

Server Object

StandaloneTCPNetworkServerStandaloneUDPNetworkServer

from __future__ import annotations

import asyncio
from collections.abc import AsyncGenerator
from contextlib import AsyncExitStack
from typing import Any

from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer
from easynetwork.servers import StandaloneTCPNetworkServer
from easynetwork.servers.handlers import AsyncStreamClient, AsyncStreamRequestHandler


class JSONProtocol(StreamProtocol[dict[str, Any], dict[str, Any]]):
    def __init__(self) -> None:
        super().__init__(JSONSerializer())


class MyRequestHandler(AsyncStreamRequestHandler[dict[str, Any], dict[str, Any]]):
    async def service_init(self, exit_stack: AsyncExitStack, server: Any) -> None:
        # StandaloneTCPNetworkServer wraps an AsyncTCPNetworkServer instance.
        # Therefore, "server" is still asynchronous.

        from easynetwork.servers import AsyncTCPNetworkServer

        assert isinstance(server, AsyncTCPNetworkServer)

    async def handle(
        self,
        client: AsyncStreamClient[dict[str, Any]],
    ) -> AsyncGenerator[None, dict[str, Any]]:
        request: dict[str, Any] = yield

        current_task = asyncio.current_task()
        assert current_task is not None

        response = {"task": current_task.get_name(), "request": request}
        await client.send_packet(response)


def main() -> None:
    host, port = "localhost", 9000
    protocol = JSONProtocol()
    handler = MyRequestHandler()

    # All the parameters are the same as AsyncTCPNetworkServer.
    server = StandaloneTCPNetworkServer(host, port, protocol, handler)

    with server:
        server.serve_forever()


if __name__ == "__main__":
    main()

Use An Other Runner

By default, the runner is "asyncio", but it can be changed during object creation.

See also

new_builtin_backend()

The token is passed to this function.

StandaloneTCPNetworkServerStandaloneUDPNetworkServer

from __future__ import annotations

from collections.abc import AsyncGenerator
from contextlib import AsyncExitStack
from typing import Any

import trio

from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer
from easynetwork.servers import StandaloneTCPNetworkServer
from easynetwork.servers.handlers import AsyncStreamClient, AsyncStreamRequestHandler


class JSONProtocol(StreamProtocol[dict[str, Any], dict[str, Any]]):
    def __init__(self) -> None:
        super().__init__(JSONSerializer())


class MyRequestHandler(AsyncStreamRequestHandler[dict[str, Any], dict[str, Any]]):
    async def service_init(self, exit_stack: AsyncExitStack, server: Any) -> None:
        # StandaloneTCPNetworkServer wraps an AsyncTCPNetworkServer instance.
        # Therefore, "server" is still asynchronous.

        from easynetwork.servers import AsyncTCPNetworkServer

        assert isinstance(server, AsyncTCPNetworkServer)

    async def handle(
        self,
        client: AsyncStreamClient[dict[str, Any]],
    ) -> AsyncGenerator[None, dict[str, Any]]:
        request: dict[str, Any] = yield

        current_task = trio.lowlevel.current_task()

        response = {"task": current_task.name, "request": request}
        await client.send_packet(response)


def main() -> None:
    host, port = "localhost", 9000
    protocol = JSONProtocol()
    handler = MyRequestHandler()

    # All the parameters are the same as AsyncTCPNetworkServer.
    server = StandaloneTCPNetworkServer(host, port, protocol, handler, backend="trio")

    with server:
        server.serve_forever()


if __name__ == "__main__":
    main()

Run Server In Background

Unlike asynchronous tasks, it is mandatory to spawn a separate execution thread. NetworkServerThread is useful for this case:

StandaloneTCPNetworkServerStandaloneUDPNetworkServer

from __future__ import annotations

import asyncio
import threading
from collections.abc import AsyncGenerator
from typing import Any

from easynetwork.clients import TCPNetworkClient
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer
from easynetwork.servers import StandaloneTCPNetworkServer
from easynetwork.servers.handlers import AsyncStreamClient, AsyncStreamRequestHandler
from easynetwork.servers.threads_helper import NetworkServerThread


class JSONProtocol(StreamProtocol[dict[str, Any], dict[str, Any]]):
    def __init__(self) -> None:
        super().__init__(JSONSerializer())


class MyRequestHandler(AsyncStreamRequestHandler[dict[str, Any], dict[str, Any]]):
    async def handle(
        self,
        client: AsyncStreamClient[dict[str, Any]],
    ) -> AsyncGenerator[None, dict[str, Any]]:
        request: dict[str, Any] = yield

        current_task = asyncio.current_task()
        assert current_task is not None

        response = {
            "thread": threading.current_thread().name,
            "task": current_task.get_name(),
            "request": request,
        }
        await client.send_packet(response)


def client(host: str, port: int, message: str) -> None:
    with TCPNetworkClient((host, port), JSONProtocol()) as client:
        client.send_packet({"message": message})
        response = client.recv_packet()
        print(f"From server: {response}")


def main() -> None:
    host, port = "localhost", 9000
    protocol = JSONProtocol()
    handler = MyRequestHandler()

    server = StandaloneTCPNetworkServer(host, port, protocol, handler)

    with server:
        server_thread = NetworkServerThread(server)
        server_thread.start()

        print(f"Server loop running in thread: {server_thread.name}")

        client(host, port, "Hello world 1")
        client(host, port, "Hello world 2")
        client(host, port, "Hello world 3")

        server_thread.join()


if __name__ == "__main__":
    main()

The output of the example should look something like this:

StandaloneTCPNetworkServerStandaloneUDPNetworkServer

$ python background_server.py
Server loop running in thread: Thread-1
From server: {'thread': 'Thread-1', 'task': 'Task-6', 'request': {'message': 'Hello world 1'}}
From server: {'thread': 'Thread-1', 'task': 'Task-8', 'request': {'message': 'Hello world 2'}}
From server: {'thread': 'Thread-1', 'task': 'Task-10', 'request': {'message': 'Hello world 3'}}