Alternative — Unix Datagram 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 datagram clients expect a DatagramProtocol 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_DGRAM 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
import tempfile

from easynetwork.clients.unix_datagram import UnixDatagramClient
from easynetwork.protocol import DatagramProtocol
from easynetwork.serializers import JSONSerializer


def obtain_a_connected_socket() -> socket.socket:
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    sock.bind(f"{tempfile.mkdtemp()}/local.sock")

    ...

    return sock


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

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

        ...


if __name__ == "__main__":
    main()

from __future__ import annotations

import asyncio
import socket
import tempfile

from easynetwork.clients.async_unix_datagram import AsyncUnixDatagramClient
from easynetwork.protocol import DatagramProtocol
from easynetwork.serializers import JSONSerializer


async def obtain_a_connected_socket() -> socket.socket:
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    sock.bind(f"{tempfile.mkdtemp()}/local.sock")

    ...

    return sock


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

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

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

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_DEBUG, True)

client.socket.setsockopt(socket.SOL_SOCKET, socket.SO_DEBUG, True)

Warning

Make sure that wait_connected() has been called before.

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().