How-to — TCP Client Endpoints

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 call to wait_connected() is required to actually initialize the client, since we cannot perform asynchronous operations at object creation. This is what the client does when it enters the the async with context.

Once completed, wait_connected() is a no-op.

Using An Already Connected Socket 

If you have your own way to obtain a connected socket.socket instance, you can pass it to the client.

If the socket is not connected, an OSError is raised.

Important

It must be a SOCK_STREAM socket with AF_INET or AF_INET6 family.

Warning

The resource ownership is given to the client. You must close the client to close the socket.

from __future__ import annotations

import socket

from easynetwork.api_sync.client import TCPNetworkClient
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer

def obtain_a_connected_socket() -> socket.socket:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    ...

    return sock

def main() -> None:
    protocol = StreamProtocol(JSONSerializer())
    sock = obtain_a_connected_socket()

    with TCPNetworkClient(sock, protocol) as client:
        print(f"Remote address: {client.get_remote_address()}")

        ...

if __name__ == "__main__":
    main()

from __future__ import annotations

import asyncio
import socket

from easynetwork.api_async.client import AsyncTCPNetworkClient
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer


async def obtain_a_connected_socket() -> socket.socket:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    ...

    return sock


async def main() -> None:
    protocol = StreamProtocol(JSONSerializer())
    sock = await obtain_a_connected_socket()

    async with AsyncTCPNetworkClient(sock, protocol) as client:
        print(f"Remote address: {client.get_remote_address()}")

        ...


if __name__ == "__main__":
    asyncio.run(main())

Note

Even with a ready-to-use socket, the call to wait_connected() is still required.

Basic Usage 

Sending Packets 

There’s not much to say, except that objects passed as arguments are automatically converted to bytes to send to the remote host thanks to the protocol object.

client.send_packet({"data": 42})

await client.send_packet({"data": 42})

Receiving Packets 

You get the next available packet, already parsed. Extraneous data is kept for the next call.

packet = client.recv_packet()
print(f"Received packet: {packet!r}")

You can control the receive timeout with the timeout parameter:

try:
    packet = client.recv_packet(timeout=30)
except TimeoutError:
    print("Timed out")
else:
    print(f"Received packet: {packet!r}")

packet = await client.recv_packet()
print(f"Received packet: {packet!r}")

You can control the receive timeout by adding a timeout scope using the asynchronous framework:

try:
    async with asyncio.timeout(30):
        packet = await client.recv_packet()
except TimeoutError:
    print("Timed out")
else:
    print(f"Received packet: {packet!r}")

Tip

Remember to catch invalid data parsing errors.

try:
    packet = client.recv_packet(timeout=30)
except StreamProtocolParseError:
    print("Received something, but was not valid")
except TimeoutError:
    print("Timed out")
else:
    print(f"Received packet: {packet!r}")

try:
    async with asyncio.timeout(30):
        packet = await client.recv_packet()
except StreamProtocolParseError:
    print("Received something, but was not valid")
except TimeoutError:
    print("Timed out")
else:
    print(f"Received packet: {packet!r}")

Receiving Multiple Packets At Once 

You can use iter_received_packets() to get all the received packets in a sequence or a set.

all_packets = [p for p in client.iter_received_packets()]

all_packets = [p async for p in client.iter_received_packets()]

The timeout parameter defaults to zero to get only the data already in the buffer, but you can change it.

all_packets = [p for p in client.iter_received_packets(timeout=1)]

Advanced Usage 

Note

This section is for people who know what they’re doing and are looking for something specific.

Close The Write-End Stream 

If you are sure you will never reuse send_packet(), you can call send_eof() to shut down the write stream.

client.send_eof()

await client.send_eof()

Note

send_eof() will block until all unsent data has been flushed before closing the stream.

Low-Level Socket Operations 

For low-level operations such as setsockopt(), the client object exposes the socket through a SocketProxy:

client.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, False)

client.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, False)

Warning

Make sure that wait_connected() has been called before.

`socket.recv()` Buffer Size 

By default, the client uses a reasonable buffer size when calling recv_packet(). You can control this value by setting the max_recv_size parameter:

with TCPNetworkClient(address, protocol, max_recv_size=1024) as client:
    # Only do socket.recv(1024) calls
    packet = client.recv_packet()

async with AsyncTCPNetworkClient(address, protocol, max_recv_size=1024) as client:
    # Only do socket.recv(1024) calls
    packet = await client.recv_packet()

Note

max_recv_size is also used as a size hint for buffered serializers. See this section for details.

SSL/TLS Connection 

If you want to use SSL to communicate with the remote host, the easiest way is to pass ssl=True:

with TCPNetworkClient(address, protocol, ssl=True) as client:
    client.send_packet({"data": 42})

    packet = client.recv_packet()

async with AsyncTCPNetworkClient(address, protocol, ssl=True) as client:
    await client.send_packet({"data": 42})

    packet = await client.recv_packet()

Note

Since this is a client object, ssl.SSLContext.wrap_socket() and ssl.SSLContext.wrap_bio() are always called with server_side=False.

Danger

You can pass an SSLContext instead, but at this point I expect you to really know what you are doing.

Concurrency And Multithreading 

All client methods are thread-safe. Synchronization follows these rules:

send_packet() and recv_packet() do not share the same threading.Lock instance.
close() will not wait for recv_packet().
The client.socket methods are also thread-safe. This means that you cannot access the underlying socket methods (e.g. getsockopt()) during a write operation.

This allows you to do something like this:

from __future__ import annotations

import contextlib
import threading
import traceback
from concurrent.futures import ThreadPoolExecutor
from typing import Any, TypeAlias

from easynetwork.api_sync.client import TCPNetworkClient
from easynetwork.exceptions import StreamProtocolParseError
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer

RequestType: TypeAlias = dict[str, Any]
ResponseType: TypeAlias = dict[str, Any]


def consumer(response: ResponseType) -> None:
    # Do some stuff...

    print(response)


def receiver_worker(
    client: TCPNetworkClient[RequestType, ResponseType],
    executor: ThreadPoolExecutor,
) -> None:
    while True:
        try:
            # Pass timeout=None to get an infinite iterator.
            # It will be terminated when the client.close() method has been called.
            for response in client.iter_received_packets(timeout=None):
                executor.submit(consumer, response)
        except StreamProtocolParseError:
            print("Parsing error")
            traceback.print_exc()
            continue


def do_main_stuff(client: TCPNetworkClient[RequestType, ResponseType]) -> None:
    while True:
        # Do some stuff...
        request = {"data": 42}

        client.send_packet(request)


def main() -> None:
    remote_address = ("localhost", 9000)
    protocol = StreamProtocol(JSONSerializer())

    with contextlib.ExitStack() as exit_stack:
        # thread pool executor setup
        executor = exit_stack.enter_context(ThreadPoolExecutor())

        # connect to remote
        client = TCPNetworkClient(remote_address, protocol)

        # receiver_worker thread setup
        receiver_worker_thread = threading.Thread(
            target=receiver_worker,
            args=(client, executor),
        )
        receiver_worker_thread.start()
        exit_stack.callback(receiver_worker_thread.join)

        # add the client close, so it will be closed
        # before joining the thread
        exit_stack.enter_context(client)

        # Setup done, let's go
        do_main_stuff(client)


if __name__ == "__main__":
    main()

All client methods do not require external task synchronization. Synchronization follows these rules:

send_packet() and recv_packet() do not share the same lock instance.
close() will not wait for recv_packet().

This allows you to do something like this:

from __future__ import annotations

import asyncio
import contextlib
import traceback
from typing import Any, TypeAlias

from easynetwork.api_async.client import AsyncTCPNetworkClient
from easynetwork.exceptions import StreamProtocolParseError
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer

RequestType: TypeAlias = dict[str, Any]
ResponseType: TypeAlias = dict[str, Any]


async def consumer(response: ResponseType) -> None:
    # Do some stuff...

    print(response)


async def receiver_worker(
    client: AsyncTCPNetworkClient[RequestType, ResponseType],
    task_group: asyncio.TaskGroup,
) -> None:
    while True:
        try:
            # Pass timeout=None to get an infinite iterator.
            # It will be terminated when the client.close() method has been called.
            async for response in client.iter_received_packets(timeout=None):
                _ = task_group.create_task(consumer(response))
        except StreamProtocolParseError:
            print("Parsing error")
            traceback.print_exc()
            continue


async def do_main_stuff(
    client: AsyncTCPNetworkClient[RequestType, ResponseType],
) -> None:
    while True:
        # Do some stuff...
        request = {"data": 42}

        await client.send_packet(request)


async def main() -> None:
    remote_address = ("localhost", 9000)
    protocol = StreamProtocol(JSONSerializer())

    async with contextlib.AsyncExitStack() as exit_stack:
        # task group setup
        task_group = await exit_stack.enter_async_context(asyncio.TaskGroup())

        # connect to remote
        client = AsyncTCPNetworkClient(remote_address, protocol)
        await exit_stack.enter_async_context(client)

        # receiver_worker task setup
        _ = task_group.create_task(receiver_worker(client, task_group))

        # Setup done, let's go
        await do_main_stuff(client)


if __name__ == "__main__":
    asyncio.run(main())

SSL/TLS Considerations 

For safety, concurrent calls to send_packet() and recv_packet() are “disabled” by default when using SSL. In fact, they share the same synchronization lock.

If you need this feature after all, you can pass ssl_shared_lock=False at object creation.

Danger

But you don’t need it, do you?

client = TCPNetworkClient(
    remote_address,
    protocol,
    ssl=True,
    ssl_shared_lock=False,
)

client = AsyncTCPNetworkClient(
    remote_address,
    protocol,
    ssl=True,
    ssl_shared_lock=False,
)

How-to — TCP Client Endpoints

The Basics 

The Protocol Object 

Connecting To The Remote Host 

Using An Already Connected Socket 

Basic Usage 

Sending Packets 

Receiving Packets 

Receiving Multiple Packets At Once 

Advanced Usage 

Close The Write-End Stream 

Low-Level Socket Operations 

`socket.recv()` Buffer Size 

SSL/TLS Connection 

Concurrency And Multithreading 

SSL/TLS Considerations 