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

The Protocol Object 

The UNIX stream clients expect a StreamProtocol instance to communicate with the peer endpoint.

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_UNIX 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.clients.unix_stream import UnixStreamClient
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer

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

    ...

    return sock

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

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

        ...

if __name__ == "__main__":
    main()

from __future__ import annotations

import asyncio
import socket

from easynetwork.clients.async_unix_stream import AsyncUnixStreamClient
from easynetwork.protocol import StreamProtocol
from easynetwork.serializers import JSONSerializer


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

    ...

    return sock


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

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

        ...


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

Note

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

Sending Packets 

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})

Sending Packets with socket control messages 

By using SocketAncillary, you can send SCM data. See the Unix manual page sendmsg(2) for details.

from easynetwork.lowlevel.socket import SocketAncillary

ancillary = SocketAncillary()
ancillary.add_fds([4])
client.send_packet({"data": 42}, ancillary_data=ancillary)

from easynetwork.lowlevel.socket import SocketAncillary

ancillary = SocketAncillary()
ancillary.add_fds([4])
await client.send_packet({"data": 42}, ancillary_data=ancillary)

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}")

try:
    with trio.fail_after(30):
        packet = await client.recv_packet()
except trio.TooSlowError:
    print("Timed out")
else:
    print(f"Received packet: {packet!r}")

try:
    with client.backend().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}")

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

try:
    with client.backend().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 Packets with socket control messages 

By using SocketAncillary, you can receive SCM data. See the Unix manual page recvmsg(2) for details.

from easynetwork.lowlevel.socket import SCMCredentials, SCMRights, SocketAncillary

ancillary = SocketAncillary()
packet = client.recv_packet(ancillary_data=ancillary)
print(f"Received packet: {packet!r}")
for message in ancillary.messages():
    match message:
        case SCMRights(fds):
            for fd in fds:
                print(f"Received file descriptor: {fd}")
        case SCMCredentials(credentials):
            for ucred in credentials:
                print(f"Received unix credential: {ucred}")

from easynetwork.lowlevel.socket import SCMCredentials, SCMRights, SocketAncillary

ancillary = SocketAncillary()
packet = await client.recv_packet(ancillary_data=ancillary)
print(f"Received packet: {packet!r}")
for message in ancillary.messages():
    match message:
        case SCMRights(fds):
            for fd in fds:
                print(f"Received file descriptor: {fd}")
        case SCMCredentials(credentials):
            for ucred in credentials:
                print(f"Received unix credential: {ucred}")

Tip

The default buffer size for this operation is approximately 8 KiB. However, you can customize this behavior.

from socket import CMSG_LEN

from easynetwork.lowlevel.socket import SocketAncillary

max_fds = 128
ancillary = SocketAncillary()
packet = client.recv_packet(ancillary_data=ancillary, ancillary_bufsize=CMSG_LEN(max_fds * 4))
print(f"Received packet: {packet!r}")
for fd in ancillary.iter_fds():
    print(f"Received file descriptor: {fd}")

from socket import CMSG_LEN

from easynetwork.lowlevel.socket import SocketAncillary

max_fds = 128
ancillary = SocketAncillary()
packet = await client.recv_packet(ancillary_data=ancillary, ancillary_bufsize=CMSG_LEN(max_fds * 4))
print(f"Received packet: {packet!r}")
for fd in ancillary.iter_fds():
    print(f"Received file descriptor: {fd}")

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)]

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.

Peer Credentials 

Get the peer’s process information with get_peer_credentials():

peer_creds = client.get_peer_credentials()
print(f"pid={peer_creds.pid}, user_id={peer_creds.uid}, group_id={peer_creds.gid}")

peer_creds = client.get_peer_credentials()
print(f"pid={peer_creds.pid}, user_id={peer_creds.uid}, group_id={peer_creds.gid}")

Warning

Make sure that wait_connected() has been called before.

Low-Level Socket Operations 

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

client.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, False)

client.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 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 UnixStreamClient(address, protocol, max_recv_size=1024) as client:
    # Only do socket.recv(1024) calls
    packet = client.recv_packet()

async with AsyncUnixStreamClient(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.

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.

All client methods do not require external task synchronization (such as asyncio.Lock). Synchronization follows these rules:

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