Loopback#

use std::time::Duration;

use pamoja_core::{Receive, Transport};
use pamoja_loopback::{LoopbackBroker, LoopbackTransport};

// 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.
let broker = LoopbackBroker::new();
let mut publisher = LoopbackTransport::new(broker.clone());
let mut subscriber = LoopbackTransport::new(broker.clone());
publisher.connect().await?;
subscriber.connect().await?;

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

let message = subscriber.recv().await?.expect("a message");
let reading = message.text().expect("text");
println!("line/+/temp took {reading} 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.
let quiet = Duration::from_millis(50);
match tokio::time::timeout(quiet, subscriber.recv()).await {
    Ok(_) => println!("line/+/temp took a second reading, which should never happen"),
    Err(_) => println!(
        "line/+/temp heard nothing more within {} ms",
        quiet.as_millis()
    ),
}

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

let deep = watcher.recv().await?.expect("a message");
let raw = deep.text().expect("text");
println!("line/#     took {raw} 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.
publisher.disconnect();
match publisher.send_text("line/mixer/temp", "21.6").await {
    Ok(_) => println!("a disconnected link took a reading, which should never happen"),
    Err(error) => println!("disconnected refused the reading: {error}"),
}

import { LoopbackBroker, type TransportMessage } from '@pamoja/loopback'

const QUIET_MS = 50

async function main() {
  // 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.
  const broker = new LoopbackBroker()
  const publisher = broker.link()
  const 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')

  const message = (await subscriber.recv())!
  console.log(`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 subscriber.recv(QUIET_MS)
    console.log('line/+/temp took a second reading, which should never happen')
  } catch {
    console.log(`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.
  const watcher = broker.link()
  await watcher.connect()
  await watcher.subscribe('line/#')
  await publisher.send('line/mixer/temp/raw', '2150')

  const deep = (await watcher.recv())!
  console.log(`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')
    console.log('a disconnected link took a reading, which should never happen')
  } catch (error) {
    console.log(`disconnected refused the reading: ${(error as Error).message}`)
  }

  return { message, deep }
}

main()

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())

// 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.
using var broker = new LoopbackBroker();
using LoopbackTransport publisher = broker.Link();
using LoopbackTransport subscriber = broker.Link();
await publisher.ConnectAsync();
await subscriber.ConnectAsync();

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

TransportMessage message = (await subscriber.ReceiveAsync())!;
Console.WriteLine(
    $"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.
TimeSpan quiet = TimeSpan.FromMilliseconds(50);
try
{
    await subscriber.ReceiveAsync(quiet);
    Console.WriteLine("line/+/temp took a second reading, which should never happen");
}
catch (TimeoutException)
{
    Console.WriteLine($"line/+/temp heard nothing more within {quiet.TotalMilliseconds} ms");
}

// A `#` covers every level that remains, so a second link takes the whole subtree,
// including the reading the single-level filter passed over.
using LoopbackTransport watcher = broker.Link();
await watcher.ConnectAsync();
await watcher.SubscribeAsync("line/#");
await publisher.SendAsync("line/mixer/temp/raw", "2150");

TransportMessage deep = (await watcher.ReceiveAsync())!;
Console.WriteLine(
    $"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.DisconnectAsync();
try
{
    await publisher.SendAsync("line/mixer/temp", "21.6");
    Console.WriteLine("a disconnected link took a reading, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"disconnected refused the reading: {error.Message}");
}