pamoja.ladder

Idiomatic transport-ladder facade.

A ladder is the answer to a node with more than one way to reach the network and no single one that always works: rungs are tried in the order they were added, cheapest first, and a message no rung accepts goes into a buffer rather than being lost.

 1"""Idiomatic transport-ladder facade.
 2
 3A ladder is the answer to a node with more than one way to reach the network and
 4no single one that always works: rungs are tried in the order they were added,
 5cheapest first, and a message no rung accepts goes into a buffer rather than
 6being lost.
 7"""
 8
 9from __future__ import annotations
10
11import enum
12
13from pamoja._native import Ladder
14
15__all__ = ["Delivery", "Ladder"]
16
17
18class Delivery(str, enum.Enum):
19    """What became of a message handed to a ladder."""
20
21    #: A rung took the message and it is on its way.
22    SENT = "Sent"
23    #: No rung would take it, so it is in the buffer awaiting a flush.
24    BUFFERED = "Buffered"
class Delivery(builtins.str, enum.Enum): 
19class Delivery(str, enum.Enum):
20    """What became of a message handed to a ladder."""
21
22    #: A rung took the message and it is on its way.
23    SENT = "Sent"
24    #: No rung would take it, so it is in the buffer awaiting a flush.
25    BUFFERED = "Buffered"

What became of a message handed to a ladder.

SENT = <Delivery.SENT: 'Sent'>
BUFFERED = <Delivery.BUFFERED: 'Buffered'>
class Ladder:

An ordered set of transports backed by an offline buffer.

def rung(self, /, transport):

Adds a rung, which is tried after the rungs already added.

Add the cheapest, most-preferred link first and the costliest fallback last, because a send takes the first rung that accepts it. The transport is consumed. A transport that delivers is subscribed and listened on; a host transport without recv is an uplink the ladder never listens on.

def connect(self, /):

Connects every rung, so a send can be tried against each in turn.

A rung that will not connect is left in the ladder: it may come back, and a send simply falls through it until it does.

def send(self, /, topic, payload):

Sends a payload, falling through the rungs and buffering if none take it.

Returns "Sent" or "Buffered". Buffering is a success, not a failure: it is what the ladder exists to do.

def flush(self, /):

Replays the buffer over the rungs, oldest message first, and reports how many went out.

def buffered(self, /):

How many messages are waiting in the buffer.

def subscribe(self, /, topic):

Subscribes every rung that listens to a topic.

The filter is kept for the life of the ladder: a rung that is down when it is placed receives it when it next connects, so subscribing before connect is fine and never fails for want of a link.

def recv(self, /):

Waits for the next message from any rung that listens, whichever delivers first.

Raises if no connected rung listens: none was added, the ladder is not connected, or every listening link has ended. The ladder does one thing at a time, so a send waits behind a receive in progress; a node that listens and reports waits with asyncio.wait_for and sends between waits.