Your own link#

/// The vendor's side of a link: what it was given to send, what it was told to
/// listen for, and what it has to deliver.
#[derive(Default)]
struct Vendor {
    sent: Vec<Message>,
    filters: Vec<String>,
    inbox: VecDeque<Message>,
}

/// A link over the vendor's queues, standing in for a radio or cloud SDK. Nothing
/// about it names a broker: it needs only the operations the contract asks for.
struct QueueLink {
    vendor: Arc<Mutex<Vendor>>,
    delivered: Arc<Notify>,
    connected: bool,
}

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

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

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

impl Receive for QueueLink {
    async fn recv(&mut self) -> Result<Option<Message>> {
        if !self.connected {
            return Err(Error::Closed);
        }
        loop {
            if let Some(message) = self.vendor.lock().expect("the vendor").inbox.pop_front() {
                return Ok(Some(message));
            }
            self.delivered.notified().await;
        }
    }
}

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

// The vendor side is shared with the link so the example can watch it, the way a
// real SDK exposes its own handles.
let vendor = Arc::new(Mutex::new(Vendor::default()));
let delivered = Arc::new(Notify::new());
let link = QueueLink {
    vendor: Arc::clone(&vendor),
    delivered: Arc::clone(&delivered),
    connected: false,
};

// The link is a rung like any shipped transport, and the ladder is the link a
// node is written against.
let mut ladder = TransportLadder::new(MemoryStore::new()).rung(link);
ladder.connect().await.expect("the ladder connects");

// A reading out through the ladder lands in the link, topic and bytes intact.
ladder.send("sensors/1", b"21.5").await.expect("a delivery");
let carried = vendor.lock().expect("the vendor").sent[0].clone();
let reading = String::from_utf8_lossy(&carried.payload);
println!("link carried: {} {reading}", carried.topic);

// A subscription placed on the ladder reaches the link.
ladder.subscribe("commands/#").await.expect("subscribe");
let filter = vendor.lock().expect("the vendor").filters[0].clone();
println!("link subscribed to: {filter}");

// What the link delivers comes back through the ladder.
vendor
    .lock()
    .expect("the vendor")
    .inbox
    .push_back(Message::new("commands/1", b"open"));
delivered.notify_one();
let command = ladder.recv().await.expect("recv").expect("a command");
let order = String::from_utf8_lossy(&command.payload);
println!("command over the ladder: {} {order}", command.topic);

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

// A link over two queues, standing in for a radio or cloud SDK. Nothing about it
// names a broker: it needs only the operations the contract asks for, and `recv` is
// what makes it a link that delivers rather than an uplink.
class QueueLink implements TransportHandlers {
  sent: TransportMessage[] = []
  filters: string[] = []
  private inbox: TransportMessage[] = []
  private waiting: ((message: TransportMessage) => void)[] = []

  async connect(): Promise<void> {}

  async send(topic: string, payload: Buffer): Promise<void> {
    this.sent.push({ topic, payload })
  }

  async subscribe(topic: string): Promise<void> {
    this.filters.push(topic)
  }

  recv(): Promise<TransportMessage> {
    const next = this.inbox.shift()
    return next ? Promise.resolve(next) : new Promise((resolve) => this.waiting.push(resolve))
  }

  // The vendor side: a message arriving from the radio.
  deliver(message: TransportMessage): void {
    const waiter = this.waiting.shift()
    if (waiter) waiter(message)
    else this.inbox.push(message)
  }
}

async function main() {
  // The link is a rung like any shipped transport, and the ladder is the link a node
  // is written against.
  const link = new QueueLink()
  const ladder = new Ladder(Store.memory())
  await ladder.rung(Transport.fromHandlers(link))
  await ladder.connect()

  // A reading out through the ladder lands in the link, topic and bytes intact.
  await ladder.send('sensors/1', Buffer.from('21.5'))
  const carried = link.sent[0]
  console.log(`link carried: ${carried.topic} ${carried.payload.toString()}`)

  // A subscription placed on the ladder reaches the link.
  await ladder.subscribe('commands/#')
  const filter = link.filters[0]
  console.log(`link subscribed to: ${filter}`)

  // What the link delivers comes back through the ladder.
  link.deliver({ topic: 'commands/1', payload: Buffer.from('open') })
  const command = (await ladder.recv())!
  console.log(`command over the ladder: ${command.topic} ${command.payload.toString()}`)

  return { carried, filter, command }
}

main()

import asyncio

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


class QueueLink:
    """A link over two queues, standing in for a radio or cloud SDK.

    Nothing about it names a broker: it needs only the operations the contract asks
    for, and ``recv`` is what makes it a link that delivers rather than an uplink.
    """

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

    async def connect(self) -> None:
        pass

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

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

    async def recv(self) -> Message:
        return await self.inbox.get()

    def deliver(self, message: Message) -> None:
        """The vendor side: a message arriving from the radio."""
        self.inbox.put_nowait(message)


async def main() -> None:
    # The link is a rung like any shipped transport, and the ladder is the link a
    # node is written against.
    link = QueueLink()
    ladder = Ladder(Store.memory())
    await ladder.rung(Transport.from_handlers(link))
    await ladder.connect()

    # A reading out through the ladder lands in the link, topic and bytes intact.
    await ladder.send("sensors/1", b"21.5")
    carried = link.sent[0]
    print(f"link carried: {carried.topic} {carried.payload.decode()}")

    # A subscription placed on the ladder reaches the link.
    await ladder.subscribe("commands/#")
    filter = link.filters[0]
    print(f"link subscribed to: {filter}")

    # What the link delivers comes back through the ladder.
    link.deliver(Message("commands/1", b"open"))
    command = await ladder.recv()
    print(f"command over the ladder: {command.topic} {command.payload.decode()}")

    return carried, filter, command


carried, filter, command = asyncio.run(main())

/// <summary>
/// A link over two queues, standing in for a radio or cloud SDK. Nothing about it
/// names a broker: it needs only the operations the contract asks for, and
/// <see cref="ReceiveAsync"/> is what makes it a link that delivers rather than an
/// uplink.
/// </summary>
private sealed class QueueLink : IReceivingTransportHandlers
{
    private readonly Channel<TransportMessage?> _inbox =
        Channel.CreateUnbounded<TransportMessage?>();

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

    public List<string> Filters { 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)
    {
        Filters.Add(topic);
        return Task.CompletedTask;
    }

    public async Task<TransportMessage?> ReceiveAsync() => await _inbox.Reader.ReadAsync();

    /// <summary>The vendor side: a message arriving from the radio.</summary>
    public void Deliver(TransportMessage message) => _inbox.Writer.TryWrite(message);
}

// The link is a rung like any shipped transport, and the ladder is the link a
// node is written against.
var link = new QueueLink();
using var ladder = new Ladder(Store.Memory());
ladder.Rung(Transport.FromHandlers(link));
await ladder.ConnectAsync();

// A reading out through the ladder lands in the link, topic and bytes intact.
await ladder.SendAsync("sensors/1", "21.5"u8.ToArray());
TransportMessage carried = link.Sent[0];
Console.WriteLine(
    $"link carried: {carried.Topic} {System.Text.Encoding.UTF8.GetString(carried.Payload)}");

// A subscription placed on the ladder reaches the link.
await ladder.SubscribeAsync("commands/#");
string filter = link.Filters[0];
Console.WriteLine($"link subscribed to: {filter}");

// What the link delivers comes back through the ladder.
link.Deliver(new TransportMessage("commands/1", "open"u8.ToArray()));
TransportMessage command = (await ladder.ReceiveAsync())!;
Console.WriteLine(
    $"command over the ladder: {command.Topic} {System.Text.Encoding.UTF8.GetString(command.Payload)}");