Engine surface#

use pamoja_core::{Receive, Transport};
use pamoja_ladder::{Delivery, TransportLadder};
use pamoja_loopback::{Faulty, LoopbackBroker, LoopbackTransport};
use pamoja_sync::MemoryStore;

// Whatever a link is underneath, MQTT, CoAP, or the in-process broker here, it reaches
// the rest of the framework through one trait. Anything that takes a link is generic
// over it, so a node is written once and pointed at whichever link it has.
let broker = LoopbackBroker::new();
let topic = "sensors/1/temperature";
let mut gateway = LoopbackTransport::new(broker.clone());
gateway.connect().await?;
gateway.subscribe(topic).await?;

// The fault injector is itself a transport wrapping a transport, so it composes
// anywhere a link does. This one fails its next send and passes the rest through.
let mut ladder = TransportLadder::new(MemoryStore::new())
    .rung(Faulty::new(LoopbackTransport::new(broker), 1));
ladder.connect().await?;

// The injected failure lands, so the reading is buffered rather than lost.
let first = ladder.send_text(topic, "20.1").await?;
let after_first = ladder.buffered().await.expect("a count");
println!("first reading: {first:?}, {after_first} queued");

// The next reading joins the back of the queue instead of overtaking it, even though
// the link would take it now. Order on the wire is the order the readings were taken.
let second = ladder.send_text(topic, "20.4").await?;
let queued = ladder.buffered().await.expect("a count");
println!("second reading: {second:?}, {queued} queued");

// Flushing forwards the backlog oldest first, and the subscriber sees it in order.
let forwarded = ladder.flush().await?;
let first_out = gateway.recv().await?.expect("a message");
let second_out = gateway.recv().await?.expect("a message");
let earlier = first_out.text().expect("text");
let later = second_out.text().expect("text");
println!("flush forwarded {forwarded}, gateway saw {earlier} then {later}");

import { Transport } from '@pamoja/core'
import { Delivery, Ladder } from '@pamoja/ladder'
import { LoopbackBroker } from '@pamoja/loopback'
import { Store } from '@pamoja/sync'

const TOPIC = 'sensors/1/temperature'

async function main() {
  // Whatever a link is underneath, MQTT, CoAP, or the in-process broker here, it reaches
  // the rest of the framework as a Transport, driven with the same four calls: connect,
  // subscribe, send, and recv. A node is written once and pointed at whichever link it has.
  const broker = new LoopbackBroker()
  const gateway: Transport = broker.rung()
  await gateway.connect()
  await gateway.subscribe(TOPIC)

  // The fault injector is itself a link wrapping a link, so it composes anywhere one does.
  // This one fails its next send and passes the rest through.
  const ladder = new Ladder(Store.memory())
  await ladder.rung(Transport.faulty(broker.rung(), 1))
  await ladder.connect()

  // The injected failure lands, so the reading is buffered rather than lost.
  const first = await ladder.send(TOPIC, '20.1')
  console.log(`first reading: ${first}, ${await ladder.buffered()} queued`)

  // The next reading joins the back of the queue instead of overtaking it, even though the
  // link would take it now. Order on the wire is the order the readings were taken.
  const second = await ladder.send(TOPIC, '20.4')
  const queued = await ladder.buffered()
  console.log(`second reading: ${second}, ${queued} queued`)

  // Flushing forwards the backlog oldest first, and the subscriber sees it in order.
  const forwarded = await ladder.flush()
  const earlier = (await gateway.recv())!.text!
  const later = (await gateway.recv())!.text!
  console.log(`flush forwarded ${forwarded}, gateway saw ${earlier} then ${later}`)

  return { first, second, queued, forwarded, left: await ladder.buffered(), earlier, later }
}

main()

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:
    # Whatever a link is underneath, MQTT, CoAP, or the in-process broker here, it reaches
    # the rest of the framework as a Transport, driven with the same four calls: connect,
    # subscribe, send, and recv. A node is written once and pointed at whichever link it has.
    broker = LoopbackBroker()
    gateway: Transport = broker.rung()
    await gateway.connect()
    await gateway.subscribe(TOPIC)

    # The fault injector is itself a link wrapping a link, so it composes anywhere one
    # does. This one fails its next send and passes the rest through.
    ladder = Ladder(Store.memory())
    await ladder.rung(Transport.faulty(broker.rung(), 1))
    await ladder.connect()

    # The injected failure lands, so the reading is buffered rather than lost.
    first = await ladder.send(TOPIC, "20.1")
    print(f"first reading: {first}, {await ladder.buffered()} queued")

    # The next reading joins the back of the queue instead of overtaking it, even though
    # the link would take it now. Order on the wire is the order the readings were taken.
    second = await ladder.send(TOPIC, "20.4")
    queued = await ladder.buffered()
    print(f"second reading: {second}, {queued} queued")

    # Flushing forwards the backlog oldest first, and the subscriber sees it in order.
    forwarded = await ladder.flush()
    earlier = (await gateway.recv()).text
    later = (await gateway.recv()).text
    print(f"flush forwarded {forwarded}, gateway saw {earlier} then {later}")

    return first, second, queued, forwarded, await ladder.buffered(), earlier, later


first, second, queued, forwarded, left, earlier, later = asyncio.run(main())

const string Topic = "sensors/1/temperature";

// Whatever a link is underneath, MQTT, CoAP, or the in-process broker here, it
// reaches the rest of the framework as a Transport, driven with the same four calls:
// connect, subscribe, send, and receive. A node is written once and pointed at
// whichever link it has.
using var broker = new LoopbackBroker();
using Transport gateway = broker.Rung();
await gateway.ConnectAsync();
await gateway.SubscribeAsync(Topic);

// The fault injector is itself a link wrapping a link, so it composes anywhere one
// does. This one fails its next send and passes the rest through.
using var ladder = new Ladder(Store.Memory());
ladder.Rung(Transport.Faulty(broker.Rung(), 1));
await ladder.ConnectAsync();

// The injected failure lands, so the reading is buffered rather than lost.
Delivery first = await ladder.SendAsync(Topic, "20.1");
Console.WriteLine($"first reading: {first}, {await ladder.BufferedAsync()} queued");

// The next reading joins the back of the queue instead of overtaking it, even
// though the link would take it now. Order on the wire is the order they were
// taken.
Delivery second = await ladder.SendAsync(Topic, "20.4");
int queued = await ladder.BufferedAsync();
Console.WriteLine($"second reading: {second}, {queued} queued");

// Flushing forwards the backlog oldest first, and the subscriber sees it in order.
int forwarded = await ladder.FlushAsync();
TransportMessage earlier = (await gateway.ReceiveAsync())!;
TransportMessage later = (await gateway.ReceiveAsync())!;
Console.WriteLine(
    $"flush forwarded {forwarded}, gateway saw"
    + $" {earlier.Text} then"
    + $" {later.Text}");