runtimepy.net.udp package

Submodules

runtimepy.net.udp.connection module

A module implementing a UDP connection interface.

class runtimepy.net.udp.connection.EchoUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: UdpConnection, EchoConnection

An echo connection for UDP.

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.connection.NullUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: UdpConnection, NullConnection

A null UDP connection.

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.connection.UdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: Connection, TransportMixin

A UDP connection interface.

async close() → None

Close this connection.

async classmethod create_connection(connect: bool = True, **kwargs) → T

Create a UDP connection.

async classmethod create_pair() → tuple[T, T]

Create a connection pair.

env: ChannelEnvironment

latest_rx_address: tuple[str, int] | None

logger: LoggerType

abstract async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

remote_address: _Optional[_IpHost]

async restart() → bool

Reset necessary underlying state for this connection to ‘process’ again.

send_binary(data: bytes | bytearray | memoryview) → None

Enqueue a binary message tos end.

send_text(data: str) → None

Enqueue a text message to send.

sendto(data: bytes, addr: IPv4Host | IPv6Host | tuple[str, int] = None) → None

Send to a specific address.

set_remote_address(addr: IPv4Host | IPv6Host) → None

Set a new remote address. Note that this doesn’t interact with or attempt to ‘connect’ the underlying socket. That should be done at creation time.

tx_binary_hwm: int

tx_text_hwm: int

uses_binary_tx_queue = False

uses_text_tx_queue = False

runtimepy.net.udp.create module

A module for instantiating the underlying networking resources for UdpConnection.

async runtimepy.net.udp.create.try_udp_transport_protocol(callback: Callable[[tuple[DatagramTransport, UdpQueueProtocol]], None] = None, **kwargs) → tuple[DatagramTransport, UdpQueueProtocol] | None

Attempt to create a transport and protocol pair.

async runtimepy.net.udp.create.udp_transport_protocol(**kwargs) → tuple[DatagramTransport, UdpQueueProtocol]

Create a transport and protocol pair relevant for this class’s implementation.

async runtimepy.net.udp.create.udp_transport_protocol_backoff(backoff: ExponentialBackoff = None, **kwargs) → tuple[DatagramTransport, UdpQueueProtocol]

Create a transport and protocol pair relevant for this class’s implementation.

runtimepy.net.udp.protocol module

A module implementing a DatagramProtocol for UdpConnection.

class runtimepy.net.udp.protocol.UdpQueueProtocol

Bases: DatagramProtocol

A simple UDP protocol that populates a message queue.

conn: Connection

datagram_received(data: bytes, addr: Tuple[str, int]) → None

Handle incoming data.

error_received(exc: Exception) → None

Log any received errors.

logger: Logger | LoggerAdapter[Any]

runtimepy.net.udp.queue module

A module implementing a simple queue-based UDP interface.

class runtimepy.net.udp.queue.QueueUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: UdpConnection

An echo connection for UDP.

datagrams: Queue[tuple[bytes, tuple[str, int]]]

init() → None

Initialize this instance.

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

Module contents

A module implementing networking utilities relevant to UDP.

class runtimepy.net.udp.EchoUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: UdpConnection, EchoConnection

An echo connection for UDP.

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.NullUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: UdpConnection, NullConnection

A null UDP connection.

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

class runtimepy.net.udp.QueueUdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: UdpConnection

An echo connection for UDP.

datagrams: Queue[tuple[bytes, tuple[str, int]]]

env: ChannelEnvironment

init() → None

Initialize this instance.

latest_rx_address: _Optional[tuple[str, int]]

logger: LoggerType

async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

remote_address: _Optional[_IpHost]

tx_binary_hwm: int

tx_text_hwm: int

class runtimepy.net.udp.UdpConnection(transport: DatagramTransport, protocol: UdpQueueProtocol)

Bases: Connection, TransportMixin

A UDP connection interface.

async close() → None

Close this connection.

async classmethod create_connection(connect: bool = True, **kwargs) → T

Create a UDP connection.

async classmethod create_pair() → tuple[T, T]

Create a connection pair.

env: ChannelEnvironment

latest_rx_address: tuple[str, int] | None

logger: LoggerType

abstract async process_datagram(data: bytes, addr: tuple[str, int]) → bool

Process a datagram.

remote_address: _Optional[_IpHost]

async restart() → bool

Reset necessary underlying state for this connection to ‘process’ again.

send_binary(data: bytes | bytearray | memoryview) → None

Enqueue a binary message tos end.

send_text(data: str) → None

Enqueue a text message to send.

sendto(data: bytes, addr: IPv4Host | IPv6Host | tuple[str, int] = None) → None

Send to a specific address.

set_remote_address(addr: IPv4Host | IPv6Host) → None

Set a new remote address. Note that this doesn’t interact with or attempt to ‘connect’ the underlying socket. That should be done at creation time.

tx_binary_hwm: int

tx_text_hwm: int

uses_binary_tx_queue = False

uses_text_tx_queue = False