pamoja-ladder

Cheapest reachable link first, buffering to a store when every link is down. One capability of pamoja, one memory-safe Rust core with bindings for TypeScript, Python, and C#.

Install

pip install pamoja-ladder

from pamoja import ladder

This pulls in pamoja-native, the compiled engine. pip install pamoja is the whole framework in one package.

Example

The script the test suite runs, spliced here as it ran.

From bindings/python/guides/ladder.py:

import asyncio

from pamoja.core import Transport
from pamoja.ladder import Delivery, Ladder
from pamoja.loopback import LoopbackBroker
from pamoja.sync import Store

TOPIC = "sensors/1/temperature"


async def main() -> None:
    # Two links off the same node: a near mesh hop and a metered backhaul. Each has its
    # own broker, so which rung carried a reading is visible from its subscriber.
    mesh = LoopbackBroker()
    backhaul = LoopbackBroker()
    gateway = backhaul.link()
    await gateway.connect()
    await gateway.subscribe(TOPIC)

    # Rungs are tried in the order they are added, cheapest first. The mesh hop loses
    # every packet here; the backhaul carries one send, then drops the next two.
    ladder = Ladder(Store.memory())
    await ladder.rung(Transport.degraded(mesh.rung(), drop_every=1))
    await ladder.rung(Transport.degraded(backhaul.rung(), up=1, down=2))
    await ladder.connect()

    # The mesh hop refuses, so the reading goes out over the backhaul and arrives on the
    # broker only that rung publishes to.
    first = await ladder.send(TOPIC, "21.5")
    arrived = await gateway.recv()
    print(f"first reading: {first}, gateway got {arrived.text}")

    # Now nothing will take a send, so the next reading is buffered rather than lost.
    second = await ladder.send(TOPIC, "21.6")
    waiting = await ladder.buffered()
    print(f"second reading: {second}, {waiting} waiting in the queue")

    # A flush while the links are still down forwards nothing and leaves the backlog
    # intact, because a record is removed only once a rung has accepted it.
    while_down = await ladder.flush()
    print(f"flush while down forwarded {while_down}, queue still {await ladder.buffered()}")

    # The backhaul is reachable again, so the buffered reading goes out exactly once.
    when_up = await ladder.flush()
    late = await gateway.recv()
    print(f"flush when up forwarded {when_up}, gateway got {late.text}")

    # The ladder is a link both ways. A subscription placed on it goes onto every rung
    # that listens, and a receive takes whichever rung delivers, so a command reaches
    # the node over whatever link is up. This one comes back over the backhaul.
    await ladder.subscribe("actuators/1/valve")
    await gateway.send("actuators/1/valve", "open")
    command = await ladder.recv()
    print(f"command back over the ladder: {command.text}")

    left = await ladder.buffered()
    return first, second, waiting, while_down, when_up, left, late, command


first, second, waiting, while_down, when_up, left, late, command = asyncio.run(main())

The same capability in every language

Documentation

• pamoja.ladder reference, every class and function in this module.
• The Transport ladder guide, with the same example in Rust, TypeScript, and C#.
• Every capability, and the install page.

License

MIT