Event bus#

use std::time::Duration;

use pamoja_bus::EventPublisher;
use pamoja_core::EventBus;

// The station's wiring makes one bus and hands each part what it needs: a
// publisher to announce, an endpoint to listen. No part holds a reference to
// another, so any of them can be replaced without touching the rest.
let bus: EventPublisher<String> = EventPublisher::new(2);
let power = bus.publisher();
let sampler = bus.publisher();
let mut heater = bus.subscribe();
let mut logger = bus.subscribe();

// One announcement reaches every part that listens, and each reads its own copy.
let reached = power.publish("battery.low".into());
println!("power     handed battery.low to {reached} parts");
let heater_took = heater.next_event().await?.expect("an event");
println!("heater    took {heater_took}");
let logger_took = logger.next_event().await?.expect("an event");
println!("logger    took {logger_took}");

// Publishing never waits, even while the part's own wait is open, and a part
// hears what it publishes. The wait borrows the endpoint, so in Rust the heater
// announces through a publisher of its own.
let heater_says = heater.publisher();
let (heard, _) = tokio::join!(heater.next_event(), async {
    heater_says.publish("heater.off".into())
});
let heard = heard?.expect("an event");
println!("heater    heard its own {heard}, sent while it waited");

// A part that joins late sees only what is published after it subscribes. There
// is no history to replay.
let mut radio = bus.subscribe();
power.publish("battery.ok".into());
let first = radio.next_event().await?.expect("an event");
println!("radio     joined late, so the first event it sees is {first}");

// Each endpoint buffers two events. The logger, busy writing to flash, falls
// behind while the sampler publishes five readings: it loses the oldest events,
// resumes with the newest, and counts what it lost.
for reading in 0..5 {
    sampler.publish(format!("wind {reading}"));
}
let resumed = logger.next_event().await?.expect("an event");
let missed = logger.missed();
println!("logger    missed {missed} and resumes at {resumed}");
let newest = logger.next_event().await?.expect("an event");
println!("logger    then took {newest}");

// A wait with a limit gives up without taking anything, so a part can do other
// work between events and lose nothing by it.
let quiet = Duration::from_millis(50);
match tokio::time::timeout(quiet, logger.next_event()).await {
    Ok(_) => println!("logger    took an event no one published, which should never happen"),
    Err(_) => println!(
        "logger    heard nothing more within {} ms",
        quiet.as_millis()
    ),
}

import { EventPublisher } from '@pamoja/bus'

const QUIET_MS = 50

async function main() {
  // The station's wiring makes one bus and hands each part what it needs: a publisher
  // to announce, an endpoint to listen. No part holds a reference to another, so any of
  // them can be replaced without touching the rest.
  const bus = new EventPublisher(2)
  const power = bus.publisher()
  const sampler = bus.publisher()
  const heater = bus.subscribe()
  const logger = bus.subscribe()

  // One announcement reaches every part that listens, and each reads its own copy.
  const reached = power.publish('battery.low')
  console.log(`power     handed battery.low to ${reached} parts`)
  const heaterTook = await heater.nextText()
  console.log(`heater    took ${heaterTook}`)
  const loggerTook = await logger.nextText()
  console.log(`logger    took ${loggerTook}`)

  // Publishing never waits, even while the part's own wait is open, and a part hears
  // what it publishes.
  const waiting = heater.nextText()
  heater.publish('heater.off')
  const heard = await waiting
  console.log(`heater    heard its own ${heard}, sent while it waited`)

  // A part that joins late sees only what is published after it subscribes. There is
  // no history to replay.
  const radio = bus.subscribe()
  power.publish('battery.ok')
  const first = await radio.nextText()
  console.log(`radio     joined late, so the first event it sees is ${first}`)

  // Each endpoint buffers two events. The logger, busy writing to flash, falls behind
  // while the sampler publishes five readings: it loses the oldest events, resumes with
  // the newest, and counts what it lost.
  for (let reading = 0; reading < 5; reading += 1) {
    sampler.publish(`wind ${reading}`)
  }
  const resumed = await logger.nextText()
  const missed = logger.missed
  console.log(`logger    missed ${missed} and resumes at ${resumed}`)
  const newest = await logger.nextText()
  console.log(`logger    then took ${newest}`)

  // A wait with a limit gives up without taking anything, so a part can do other work
  // between events and lose nothing by it.
  try {
    await logger.nextText(QUIET_MS)
    console.log('logger    took an event no one published, which should never happen')
  } catch {
    console.log(`logger    heard nothing more within ${QUIET_MS} ms`)
  }

  return { reached, heaterTook, loggerTook, heard, first, missed, resumed, newest }
}

main()

import asyncio

from pamoja.bus import EventPublisher

QUIET_MS = 50


async def main() -> None:
    # The station's wiring makes one bus and hands each part what it needs: a publisher
    # to announce, an endpoint to listen. No part holds a reference to another, so any
    # of them can be replaced without touching the rest.
    bus = EventPublisher(2)
    power = bus.publisher()
    sampler = bus.publisher()
    heater = bus.subscribe()
    logger = bus.subscribe()

    # One announcement reaches every part that listens, and each reads its own copy.
    reached = power.publish("battery.low")
    print(f"power     handed battery.low to {reached} parts")
    heater_took = await heater.next_text()
    print(f"heater    took {heater_took}")
    logger_took = await logger.next_text()
    print(f"logger    took {logger_took}")

    # Publishing never waits, even while the part's own wait is open, and a part hears
    # what it publishes.
    waiting = heater.next_text()
    heater.publish("heater.off")
    heard = await waiting
    print(f"heater    heard its own {heard}, sent while it waited")

    # A part that joins late sees only what is published after it subscribes. There is
    # no history to replay.
    radio = bus.subscribe()
    power.publish("battery.ok")
    first = await radio.next_text()
    print(f"radio     joined late, so the first event it sees is {first}")

    # Each endpoint buffers two events. The logger, busy writing to flash, falls behind
    # while the sampler publishes five readings: it loses the oldest events, resumes
    # with the newest, and counts what it lost.
    for reading in range(5):
        sampler.publish(f"wind {reading}")
    resumed = await logger.next_text()
    missed = logger.missed
    print(f"logger    missed {missed} and resumes at {resumed}")
    newest = await logger.next_text()
    print(f"logger    then took {newest}")

    # A wait with a limit gives up without taking anything, so a part can do other work
    # between events and lose nothing by it.
    try:
        await asyncio.wait_for(logger.next_text(), QUIET_MS / 1000)
        print("logger    took an event no one published, which should never happen")
    except asyncio.TimeoutError:
        print(f"logger    heard nothing more within {QUIET_MS} ms")

    return reached, heater_took, logger_took, heard, first, missed, resumed, newest


seen = asyncio.run(main())

// The station's wiring makes one bus and hands each part what it needs: a
// publisher to announce, an endpoint to listen. No part holds a reference to
// another, so any of them can be replaced without touching the rest.
using var bus = new EventPublisher(2);
using EventPublisher power = bus.Publisher();
using EventPublisher sampler = bus.Publisher();
using EventBus heater = bus.Subscribe();
using EventBus logger = bus.Subscribe();

// One announcement reaches every part that listens, and each reads its own copy.
int reached = power.Publish("battery.low");
Console.WriteLine($"power     handed battery.low to {reached} parts");
string heaterTook = await heater.NextTextAsync();
Console.WriteLine($"heater    took {heaterTook}");
string loggerTook = await logger.NextTextAsync();
Console.WriteLine($"logger    took {loggerTook}");

// Publishing never waits, even while the part's own wait is open, and a part
// hears what it publishes.
Task<string> waiting = heater.NextTextAsync();
heater.Publish("heater.off");
string heard = await waiting;
Console.WriteLine($"heater    heard its own {heard}, sent while it waited");

// A part that joins late sees only what is published after it subscribes.
// There is no history to replay.
using EventBus radio = bus.Subscribe();
power.Publish("battery.ok");
string first = await radio.NextTextAsync();
Console.WriteLine($"radio     joined late, so the first event it sees is {first}");

// Each endpoint buffers two events. The logger, busy writing to flash, falls
// behind while the sampler publishes five readings: it loses the oldest
// events, resumes with the newest, and counts what it lost.
for (int reading = 0; reading < 5; reading++)
{
    sampler.Publish($"wind {reading}");
}

string resumed = await logger.NextTextAsync();
long missed = logger.Missed;
Console.WriteLine($"logger    missed {missed} and resumes at {resumed}");
string newest = await logger.NextTextAsync();
Console.WriteLine($"logger    then took {newest}");

// A wait with a limit gives up without taking anything, so a part can do
// other work between events and lose nothing by it.
TimeSpan quiet = TimeSpan.FromMilliseconds(50);
try
{
    await logger.NextTextAsync(quiet);
    Console.WriteLine("logger    took an event no one published, which should never happen");
}
catch (TimeoutException)
{
    Console.WriteLine($"logger    heard nothing more within {quiet.TotalMilliseconds} ms");
}