Your own link#

/// What the modem vendor's SDK holds: what it was given to send, the filters it
/// listens on, what it has to hand over, and whether it has signal.
#[derive(Default)]
struct Sdk {
    sent: Vec<Message>,
    filters: Vec<String>,
    inbox: VecDeque<Result<Message>>,
    no_signal: bool,
}

/// A cellular modem reached through its vendor's SDK. Nothing in it names a broker
/// or a protocol: it needs only the operations the contract asks for, and
/// `Receive` is what makes it a link that delivers.
struct Modem {
    sdk: Arc<Mutex<Sdk>>,
    arrived: Arc<Notify>,
    connected: bool,
    ended: bool,
}

impl Transport for Modem {
    async fn connect(&mut self) -> Result<()> {
        self.connected = true;
        self.ended = false;
        Ok(())
    }

    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
        if !self.connected {
            return Err(Error::Closed);
        }
        let mut sdk = self.sdk.lock().expect("the SDK");
        if sdk.no_signal {
            return Err(Error::Transport("no signal".to_owned()));
        }
        sdk.sent.push(Message::new(topic, payload));
        Ok(())
    }

    async fn subscribe(&mut self, topic: &str) -> Result<()> {
        if !self.connected {
            return Err(Error::Closed);
        }
        self.sdk
            .lock()
            .expect("the SDK")
            .filters
            .push(topic.to_owned());
        Ok(())
    }
}

impl Receive for Modem {
    async fn recv(&mut self) -> Result<Option<Message>> {
        if !self.connected {
            return Err(Error::Closed);
        }
        if self.ended {
            return Ok(None);
        }
        loop {
            let next = self.sdk.lock().expect("the SDK").inbox.pop_front();
            if let Some(next) = next {
                self.ended = next.is_err();
                return next.map(Some);
            }
            self.arrived.notified().await;
        }
    }
}

/// A satellite messenger: it sends and never receives, so it implements
/// `Transport` alone and goes on a ladder as an uplink.
struct Satellite {
    sent: Arc<Mutex<Vec<Message>>>,
}

impl Transport for Satellite {
    async fn connect(&mut self) -> Result<()> {
        Ok(())
    }

    async fn send(&mut self, topic: &str, payload: &[u8]) -> Result<()> {
        self.sent
            .lock()
            .expect("the messenger")
            .push(Message::new(topic, payload));
        Ok(())
    }

    async fn subscribe(&mut self, _topic: &str) -> Result<()> {
        Err(Error::Transport(
            "a satellite messenger only sends".to_owned(),
        ))
    }
}

use pamoja_ladder::TransportLadder;
use pamoja_sync::MemoryStore;

// The SDK's state is shared with the example, which plays the vendor's side: it
// hands the modem a command, takes its signal away, and drops its session.
let sdk = Arc::new(Mutex::new(Sdk::default()));
let arrived = Arc::new(Notify::new());
let beamed = Arc::new(Mutex::new(Vec::new()));
let hand_over = |next: Result<Message>| {
    sdk.lock().expect("the SDK").inbox.push_back(next);
    arrived.notify_one();
};
let modem = Modem {
    sdk: Arc::clone(&sdk),
    arrived: Arc::clone(&arrived),
    connected: false,
    ended: false,
};
let satellite = Satellite {
    sent: Arc::clone(&beamed),
};

// The modem is the cheaper link and goes first. The messenger only sends, so it
// goes on as an uplink, which a ladder never listens on or subscribes.
let mut ladder = TransportLadder::new(MemoryStore::new())
    .rung(modem)
    .uplink(satellite);
ladder.connect().await?;

// A reading goes out over the first link that takes it.
ladder.send_text("waves/height", "1.8").await?;
let carried = sdk.lock().expect("the SDK").sent[0].clone();
println!("modem     carried {} {}", carried.topic, carried.text()?);

// A subscription reaches every link that listens, and only those: the messenger,
// whose subscribe would refuse, is never asked.
ladder.subscribe("commands/#").await?;
let filter = sdk.lock().expect("the SDK").filters[0].clone();
println!("modem     listens on {filter}, and the satellite was never asked");

// What the modem hands over comes back through the ladder.
hand_over(Ok(Message::new("commands/interval", b"600")));
let command = ladder.recv().await?.expect("a command");
println!("buoy      took {} {}", command.topic, command.text()?);

// A link that refuses a send passes the reading down to the next link.
sdk.lock().expect("the SDK").no_signal = true;
ladder.send_text("waves/height", "2.4").await?;
let relayed = beamed.lock().expect("the messenger")[0].clone();
println!(
    "satellite carried {} {} while the modem had no signal",
    relayed.topic,
    relayed.text()?
);

// A link that fails while listening says why, once, and has ended after that.
// With no other link listening, the ladder then has nothing to wait on.
hand_over(Err(Error::Transport(
    "the modem lost its session".to_owned(),
)));
let lost = ladder.recv().await.expect_err("the session was lost");
println!("buoy      lost the modem: {lost}");
let idle = ladder.recv().await.expect_err("no link listens");
println!("buoy      has no link left to listen on: {idle}");

// Connecting again brings the modem back, and the ladder places its filter on it
// again, so a link's subscribe runs once for every connect.
ladder.connect().await?;
let filters = sdk.lock().expect("the SDK").filters.clone();
println!(
    "modem     reconnected, and the ladder placed {} on it again",
    filters[1]
);
hand_over(Ok(Message::new("commands/interval", b"900")));
let later = ladder.recv().await?.expect("a command");
println!("buoy      took {} {}", later.topic, later.text()?);

import { Transport, type DeliveredMessage, type TransportHandlers } from '@pamoja/core'
import { Ladder } from '@pamoja/ladder'
import { Store } from '@pamoja/sync'

// A cellular modem reached through its vendor's SDK, which this class stands in for.
// Nothing in it names a broker or a protocol: it needs only the operations the
// contract asks for, and `recv` is what makes it a link that delivers.
class Modem implements TransportHandlers {
  sent: { topic: string; text: string }[] = []
  filters: string[] = []
  noSignal = false
  private inbox: (DeliveredMessage | Error)[] = []
  private waiting: ((next: DeliveredMessage | Error) => void)[] = []

  connect(): void {}

  send(topic: string, payload: Buffer): void {
    if (this.noSignal) throw new Error('no signal')
    this.sent.push({ topic, text: payload.toString() })
  }

  subscribe(topic: string): void {
    this.filters.push(topic)
  }

  async recv(): Promise<DeliveredMessage> {
    const next =
      this.inbox.shift() ?? (await new Promise<DeliveredMessage | Error>((hand) => this.waiting.push(hand)))
    if (next instanceof Error) throw next
    return next
  }

  // The vendor's side: a message from the network, or the loss of the session,
  // handed to the wait that is open or kept for the next one.
  handOver(next: DeliveredMessage | Error): void {
    const waiter = this.waiting.shift()
    if (waiter) waiter(next)
    else this.inbox.push(next)
  }
}

// A satellite messenger sends and never receives. With no `recv` it is an uplink,
// which a ladder never listens on or subscribes.
class Satellite implements TransportHandlers {
  sent: { topic: string; text: string }[] = []

  connect(): void {}

  send(topic: string, payload: Buffer): void {
    this.sent.push({ topic, text: payload.toString() })
  }

  subscribe(): void {
    throw new Error('a satellite messenger only sends')
  }
}

async function main() {
  // The modem is the cheaper link and goes first. The messenger has no `recv`, so it
  // goes on as an uplink.
  const modem = new Modem()
  const satellite = new Satellite()
  const ladder = new Ladder(Store.memory())
  await ladder.rung(Transport.fromHandlers(modem))
  await ladder.rung(Transport.fromHandlers(satellite))
  await ladder.connect()

  // A reading goes out over the first link that takes it.
  await ladder.send('waves/height', '1.8')
  const carried = modem.sent[0]
  console.log(`modem     carried ${carried.topic} ${carried.text}`)

  // A subscription reaches every link that listens, and only those: the messenger,
  // whose subscribe would throw, is never asked.
  await ladder.subscribe('commands/#')
  const filter = modem.filters[0]
  console.log(`modem     listens on ${filter}, and the satellite was never asked`)

  // What the modem hands over comes back through the ladder.
  modem.handOver({ topic: 'commands/interval', payload: '600' })
  const command = (await ladder.recv())!
  console.log(`buoy      took ${command.topic} ${command.text!}`)

  // A link that refuses a send passes the reading down to the next link.
  modem.noSignal = true
  await ladder.send('waves/height', '2.4')
  const relayed = satellite.sent[0]
  console.log(`satellite carried ${relayed.topic} ${relayed.text} while the modem had no signal`)

  // A link that fails while listening says why, once, and has ended after that. With
  // no other link listening, the ladder then has nothing to wait on.
  modem.handOver(new Error('the modem lost its session'))
  let lost = ''
  try {
    await ladder.recv()
  } catch (error) {
    lost = (error as Error).message
  }
  console.log(`buoy      lost the modem: ${lost}`)
  let idle = ''
  try {
    await ladder.recv()
  } catch (error) {
    idle = (error as Error).message
  }
  console.log(`buoy      has no link left to listen on: ${idle}`)

  // Connecting again brings the modem back, and the ladder places its filter on it
  // again, so a link's subscribe runs once for every connect.
  await ladder.connect()
  console.log(`modem     reconnected, and the ladder placed ${modem.filters[1]} on it again`)
  modem.handOver({ topic: 'commands/interval', payload: '900' })
  const later = (await ladder.recv())!
  console.log(`buoy      took ${later.topic} ${later.text!}`)

  return { carried, filters: modem.filters, command, relayed, lost, idle, later }
}

main()

import asyncio
from typing import Union

from pamoja.core import Message, PamojaError, Transport
from pamoja.ladder import Ladder
from pamoja.sync import Store


class Modem:
    """A cellular modem reached through its vendor's SDK, which this class stands in for.

    Nothing in it names a broker or a protocol: it needs only the operations the
    contract asks for, and ``recv`` is what makes it a link that delivers.
    """

    def __init__(self) -> None:
        self.sent: list[Message] = []
        self.filters: list[str] = []
        self.no_signal = False
        self.inbox: asyncio.Queue[Union[Message, Exception]] = asyncio.Queue()

    def connect(self) -> None:
        pass

    def send(self, topic: str, payload: bytes) -> None:
        if self.no_signal:
            raise ConnectionError("no signal")
        self.sent.append(Message(topic, payload))

    def subscribe(self, topic: str) -> None:
        self.filters.append(topic)

    async def recv(self) -> Message:
        arrived = await self.inbox.get()
        if isinstance(arrived, Exception):
            raise arrived
        return arrived

    def hand_over(self, arrived: Union[Message, Exception]) -> None:
        """The vendor's side: a message from the network, or the loss of the session."""
        self.inbox.put_nowait(arrived)


class Satellite:
    """A satellite messenger sends and never receives.

    With no ``recv`` it is an uplink, which a ladder never listens on or subscribes.
    """

    def __init__(self) -> None:
        self.sent: list[Message] = []

    def connect(self) -> None:
        pass

    def send(self, topic: str, payload: bytes) -> None:
        self.sent.append(Message(topic, payload))

    def subscribe(self, topic: str) -> None:
        raise ConnectionError("a satellite messenger only sends")


async def main() -> None:
    # The modem is the cheaper link and goes first. The messenger has no `recv`, so it
    # goes on as an uplink.
    modem = Modem()
    satellite = Satellite()
    ladder = Ladder(Store.memory())
    await ladder.rung(Transport.from_handlers(modem))
    await ladder.rung(Transport.from_handlers(satellite))
    await ladder.connect()

    # A reading goes out over the first link that takes it.
    await ladder.send("waves/height", "1.8")
    carried = modem.sent[0]
    print(f"modem     carried {carried.topic} {carried.text}")

    # A subscription reaches every link that listens, and only those: the messenger,
    # whose subscribe would raise, is never asked.
    await ladder.subscribe("commands/#")
    placed = modem.filters[0]
    print(f"modem     listens on {placed}, and the satellite was never asked")

    # What the modem hands over comes back through the ladder.
    modem.hand_over(Message("commands/interval", "600"))
    command = await ladder.recv()
    print(f"buoy      took {command.topic} {command.text}")

    # A link that refuses a send passes the reading down to the next link.
    modem.no_signal = True
    await ladder.send("waves/height", "2.4")
    relayed = satellite.sent[0]
    print(f"satellite carried {relayed.topic} {relayed.text} while the modem had no signal")

    # A link that fails while listening says why, once, and has ended after that. With
    # no other link listening, the ladder then has nothing to wait on.
    modem.hand_over(ConnectionResetError("the modem lost its session"))
    try:
        await ladder.recv()
    except PamojaError as error:
        lost = str(error)
    print(f"buoy      lost the modem: {lost}")
    try:
        await ladder.recv()
    except PamojaError as error:
        idle = str(error)
    print(f"buoy      has no link left to listen on: {idle}")

    # Connecting again brings the modem back, and the ladder places its filter on it
    # again, so a link's subscribe runs once for every connect.
    await ladder.connect()
    print(f"modem     reconnected, and the ladder placed {modem.filters[1]} on it again")
    modem.hand_over(Message("commands/interval", "900"))
    later = await ladder.recv()
    print(f"buoy      took {later.topic} {later.text}")

    return carried, modem.filters, command, relayed, lost, idle, later


seen = asyncio.run(main())

/// <summary>
/// A cellular modem reached through its vendor's SDK, which this class stands in
/// for. Nothing in it names a broker or a protocol: it needs only the operations
/// the contract asks for, and <see cref="ReceiveAsync"/> is what makes it a link
/// that delivers.
/// </summary>
private sealed class Modem : IReceivingTransportHandlers
{
    private readonly Channel<object> _inbox = Channel.CreateUnbounded<object>();

    public List<TransportMessage> Sent { get; } = new();

    public List<string> Filters { get; } = new();

    public bool NoSignal { get; set; }

    public Task ConnectAsync() => Task.CompletedTask;

    public Task SendAsync(string topic, ReadOnlyMemory<byte> payload)
    {
        if (NoSignal)
        {
            return Task.FromException(new IOException("no signal"));
        }

        Sent.Add(new TransportMessage(topic, payload.ToArray()));
        return Task.CompletedTask;
    }

    public Task SubscribeAsync(string topic)
    {
        Filters.Add(topic);
        return Task.CompletedTask;
    }

    public async Task<TransportMessage?> ReceiveAsync() =>
        await _inbox.Reader.ReadAsync() switch
        {
            Exception lost => throw lost,
            var arrived => (TransportMessage)arrived,
        };

    /// <summary>
    /// The vendor's side: a message from the network, or the loss of the session.
    /// </summary>
    /// <param name="arrived">What the SDK hands over.</param>
    public void HandOver(object arrived) => _inbox.Writer.TryWrite(arrived);
}

/// <summary>
/// A satellite messenger sends and never receives. Without
/// <see cref="IReceivingTransportHandlers"/> it is an uplink, which a ladder never
/// listens on or subscribes.
/// </summary>
private sealed class Satellite : ITransportHandlers
{
    public List<TransportMessage> Sent { get; } = new();

    public Task ConnectAsync() => Task.CompletedTask;

    public Task SendAsync(string topic, ReadOnlyMemory<byte> payload)
    {
        Sent.Add(new TransportMessage(topic, payload.ToArray()));
        return Task.CompletedTask;
    }

    public Task SubscribeAsync(string topic) =>
        Task.FromException(new NotSupportedException("a satellite messenger only sends"));
}

// The modem is the cheaper link and goes first. The messenger has no
// ReceiveAsync, so it goes on as an uplink.
var modem = new Modem();
var satellite = new Satellite();
using var ladder = new Ladder(Store.Memory());
ladder.Rung(Transport.FromHandlers(modem));
ladder.Rung(Transport.FromHandlers(satellite));
await ladder.ConnectAsync();

// A reading goes out over the first link that takes it.
await ladder.SendAsync("waves/height", "1.8");
TransportMessage carried = modem.Sent[0];
Console.WriteLine($"modem     carried {carried.Topic} {carried.Text}");

// A subscription reaches every link that listens, and only those: the
// messenger, whose subscribe would fail, is never asked.
await ladder.SubscribeAsync("commands/#");
string placed = modem.Filters[0];
Console.WriteLine($"modem     listens on {placed}, and the satellite was never asked");

// What the modem hands over comes back through the ladder.
modem.HandOver(new TransportMessage("commands/interval", "600"));
TransportMessage command = (await ladder.ReceiveAsync())!;
Console.WriteLine($"buoy      took {command.Topic} {command.Text}");

// A link that refuses a send passes the reading down to the next link.
modem.NoSignal = true;
await ladder.SendAsync("waves/height", "2.4");
TransportMessage relayed = satellite.Sent[0];
Console.WriteLine(
    $"satellite carried {relayed.Topic} {relayed.Text} while the modem had no signal");

// A link that fails while listening says why, once, and has ended after that.
// With no other link listening, the ladder then has nothing to wait on.
modem.HandOver(new IOException("the modem lost its session"));
string lost = string.Empty;
try
{
    await ladder.ReceiveAsync();
}
catch (PamojaException error)
{
    lost = error.Message;
}

Console.WriteLine($"buoy      lost the modem: {lost}");
string idle = string.Empty;
try
{
    await ladder.ReceiveAsync();
}
catch (PamojaException error)
{
    idle = error.Message;
}

Console.WriteLine($"buoy      has no link left to listen on: {idle}");

// Connecting again brings the modem back, and the ladder places its filter on
// it again, so a link's subscribe runs once for every connect.
await ladder.ConnectAsync();
Console.WriteLine(
    $"modem     reconnected, and the ladder placed {modem.Filters[1]} on it again");
modem.HandOver(new TransportMessage("commands/interval", "900"));
TransportMessage later = (await ladder.ReceiveAsync())!;
Console.WriteLine($"buoy      took {later.Topic} {later.Text}");