pamoja-loopback

An in-process transport with topic matching, a fault injector, and outages on demand, for testing with no broker. One capability of pamoja, one memory-safe Rust core with bindings for TypeScript, Python, and C#.

Install

pip install pamoja-loopback

from pamoja import loopback

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/loopback.py:

import asyncio

from pamoja.core import PamojaError
from pamoja.loopback import LoopbackBroker

QUIET_MS = 50


async def main() -> None:
    # One broker and two links off it, all in this process. Nothing binds a port and
    # nothing has to be running for the traffic below to flow, which is what makes this
    # the link to develop a node against before it has a real one.
    broker = LoopbackBroker()
    publisher = broker.link()
    subscriber = broker.link()
    await publisher.connect()
    await subscriber.connect()

    # A `+` stands for exactly one level, so this takes the mixer's temperature but not the
    # raw reading a level below it.
    await subscriber.subscribe("line/+/temp")
    await publisher.send("line/mixer/temp/raw", "2150")
    await publisher.send("line/mixer/temp", "21.5")

    message = await subscriber.recv()
    print(f"line/+/temp took {message.text} from {message.topic}")

    # The raw reading went out first and never arrived, which a test proves by waiting a
    # set time for anything more rather than forever. Giving up loses nothing: a message
    # that came later would wait for the next receive.
    try:
        await asyncio.wait_for(subscriber.recv(), QUIET_MS / 1000)
        print("line/+/temp took a second reading, which should never happen")
    except asyncio.TimeoutError:
        print(f"line/+/temp heard nothing more within {QUIET_MS} ms")

    # A `#` covers every level that remains, so a second link takes the whole subtree,
    # including the reading the single-level filter passed over.
    watcher = broker.link()
    await watcher.connect()
    await watcher.subscribe("line/#")
    await publisher.send("line/mixer/temp/raw", "2150")

    deep = await watcher.recv()
    print(f"line/#     took {deep.text} from {deep.topic}")

    # A link that has been disconnected reports the failure instead of dropping the
    # reading, which is the case a test wants to reach without unplugging anything.
    await publisher.disconnect()
    try:
        await publisher.send("line/mixer/temp", "21.6")
        print("a disconnected link took a reading, which should never happen")
    except PamojaError as error:
        print(f"disconnected refused the reading: {error}")

    return message, deep


message, deep = asyncio.run(main())

The same capability in every language

Documentation

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

License

MIT