MQTT#

use pamoja_core::{Receive, Transport};
use pamoja_mqtt::{MqttConfig, MqttTransport, PublishOptions, QualityOfService, Will};

let connection = |connected: bool| {
    if connected {
        "still connected"
    } else {
        "not connected"
    }
};

// The gateway takes every temperature on the site. A `+` stands for exactly one level,
// so this matches every node's temperature and nothing deeper.
let gateway_config = MqttConfig::new("site-gateway", "127.0.0.1", port)
    .keep_alive(Duration::from_secs(5))
    .qos(QualityOfService::AtLeastOnce);
let mut gateway = connect(gateway_config).await;
gateway
    .subscribe("sensors/+/temperature")
    .await
    .expect("the broker accepts the subscription");
println!("gateway   subscribed to sensors/+/temperature");

// A node publishes under that pattern. At least once has the broker acknowledge each
// message, where at most once would send it and forget it. It also leaves a will: should
// it drop off the network without saying goodbye, the broker publishes offline on its
// status topic for it.
let will = Will::new("sites/node-1/status", "offline")
    .qos(QualityOfService::AtLeastOnce)
    .retained();
let node_config = MqttConfig::new("node-1", "127.0.0.1", port)
    .keep_alive(Duration::from_secs(5))
    .qos(QualityOfService::AtLeastOnce)
    .last_will(will);
let mut node = connect(node_config).await;
node.send_text("sensors/1/temperature", "21.5")
    .await
    .expect("the broker takes the reading");
println!("node      published 21.5 to sensors/1/temperature");

// The gateway receives it with the topic attached, which is how it knows which node
// sent the reading without the payload having to repeat it.
let received = gateway
    .recv()
    .await
    .expect("the link is up")
    .expect("a message arrives");
let reading = received.text().expect("text");
let topic = &received.topic;
println!("gateway   got {reading} on {topic}");

// The node says it is up, retained, so a dashboard that opens later sees it at once.
// `publish` takes options for this one message and hands back a delivery, which settles
// when the broker acknowledges the message.
node.publish(
    "sites/node-1/status",
    b"online",
    PublishOptions::new().retained(),
)
.await
.expect("the link is up")
.confirmed()
.await
.expect("the broker acknowledges the status");
println!("node      the broker holds online on sites/node-1/status");

// A dashboard that subscribes afterwards still gets it, because the broker keeps the
// last retained message on each topic for whoever subscribes next.
let dashboard_config =
    MqttConfig::new("site-dashboard", "127.0.0.1", port).keep_alive(Duration::from_secs(5));
let mut dashboard = connect(dashboard_config).await;
dashboard
    .subscribe("sites/node-1/status")
    .await
    .expect("the broker accepts the subscription");
let status = dashboard
    .recv()
    .await
    .expect("the link is up")
    .expect("the retained status arrives");
let state = status.text().expect("text");
println!("dashboard {} is {state}", status.topic);

// Two days of readings saved at one a minute, sent as one message, make a packet over
// the connection's 10 KiB limit. The send is refused before anything leaves and the
// connection stays up; a node that must send it raises the limit on every client that
// shares the topic, or splits it.
let backlog = vec!["21.5"; 2 * 24 * 60].join(",");
match node.send_text("sensors/1/backlog", &backlog).await {
    Ok(()) => println!("node      sent an oversized backlog, which should never happen"),
    Err(error) => println!("node      backlog refused: {error}"),
}
let after_refusal = node.is_connected();
println!("node      {}", connection(after_refusal));

// Before it leaves, the node says so itself. A clean disconnect discards the will, which
// is only for a node that drops off without this goodbye.
node.publish(
    "sites/node-1/status",
    b"offline",
    PublishOptions::new().retained(),
)
.await
.expect("the link is up")
.confirmed()
.await
.expect("the broker acknowledges the goodbye");
let goodbye = dashboard
    .recv()
    .await
    .expect("the link is up")
    .expect("the goodbye arrives");
let farewell = goodbye.text().expect("text");
println!("dashboard {} is {farewell}", goodbye.topic);

// Disconnecting leaves the transport reusable, so a node that loses its link can
// reconnect the same object when the broker comes back.
node.disconnect().await.expect("a clean disconnect");
let after_disconnect = node.is_connected();
println!(
    "node      {} after disconnecting",
    connection(after_disconnect)
);

// A broker that is not there is reported rather than leaving a client that looks
// connected, so a retry loop has something to test. Nothing listens on port 1.
let mut nowhere = MqttTransport::new(
    MqttConfig::new("node-2", "127.0.0.1", 1).qos(QualityOfService::ExactlyOnce),
);
match nowhere.connect().await {
    Ok(()) => {
        println!("an unreachable broker accepted a connection, which should never happen")
    }
    Err(error) => println!("unreachable broker refused: {error}"),
}

import { MqttClient, Qos } from '@pamoja/mqtt'

// The broker on the site. The guide's CI runs one on localhost; point these at yours and
// nothing else changes.
const BROKER = '127.0.0.1'
const PORT = 1883

const connection = (connected: boolean) => (connected ? 'still connected' : 'not connected')

async function main() {
  // The gateway takes every temperature on the site. A `+` stands for exactly one level,
  // so this matches every node's temperature and nothing deeper.
  const gateway = new MqttClient({
    clientId: 'site-gateway',
    host: BROKER,
    port: PORT,
    qos: Qos.AtLeastOnce,
  })
  await gateway.connect()
  await gateway.subscribe('sensors/+/temperature')
  console.log('gateway   subscribed to sensors/+/temperature')

  // A node publishes under that pattern. At least once has the broker acknowledge each
  // message, where at most once would send it and forget it. It also leaves a will: should
  // it drop off the network without saying goodbye, the broker publishes offline on its
  // status topic for it.
  const node = new MqttClient({
    clientId: 'node-1',
    host: BROKER,
    port: PORT,
    qos: Qos.AtLeastOnce,
    will: { topic: 'sites/node-1/status', payload: 'offline', qos: Qos.AtLeastOnce, retain: true },
  })
  await node.connect()
  await node.publish('sensors/1/temperature', '21.5')
  console.log('node      published 21.5 to sensors/1/temperature')

  // The gateway receives it with the topic attached, which is how it knows which node
  // sent the reading without the payload having to repeat it.
  const received = (await gateway.recv())!
  console.log(`gateway   got ${received.text!} on ${received.topic}`)

  // The node says it is up, retained, so a dashboard that opens later sees it at once.
  // `publishConfirmed` resolves once the broker acknowledges the message.
  await node.publishConfirmed('sites/node-1/status', 'online', { retain: true })
  console.log('node      the broker holds online on sites/node-1/status')

  // A dashboard that subscribes afterwards still gets it, because the broker keeps the
  // last retained message on each topic for whoever subscribes next.
  const dashboard = new MqttClient({ clientId: 'site-dashboard', host: BROKER, port: PORT })
  await dashboard.connect()
  await dashboard.subscribe('sites/node-1/status')
  const status = (await dashboard.recv())!
  console.log(`dashboard ${status.topic} is ${status.text!}`)

  // Two days of readings saved at one a minute, sent as one message, make a packet over
  // the connection's 10 KiB limit. The send is refused before anything leaves and the
  // connection stays up; a node that must send it raises the limit on every client that
  // shares the topic, or splits it.
  const backlog = Array(2 * 24 * 60).fill('21.5').join(',')
  try {
    await node.publish('sensors/1/backlog', backlog)
    console.log('node      sent an oversized backlog, which should never happen')
  } catch (error) {
    console.log(`node      backlog refused: ${(error as Error).message}`)
  }
  const afterRefusal = await node.isConnected()
  console.log(`node      ${connection(afterRefusal)}`)

  // Before it leaves, the node says so itself. A clean disconnect discards the will, which
  // is only for a node that drops off without this goodbye.
  await node.publishConfirmed('sites/node-1/status', 'offline', { retain: true })
  const goodbye = (await dashboard.recv())!
  console.log(`dashboard ${goodbye.topic} is ${goodbye.text!}`)

  // Disconnecting leaves the client reusable, so a node that loses its link can reconnect
  // the same object when the broker comes back.
  await node.disconnect()
  const afterDisconnect = await node.isConnected()
  console.log(`node      ${connection(afterDisconnect)} after disconnecting`)
  await gateway.disconnect()
  await dashboard.disconnect()

  // A broker that is not there is reported rather than leaving a client that looks
  // connected, so a retry loop has something to test.
  const nowhere = new MqttClient({ clientId: 'node-2', host: BROKER, port: 1, keepAliveSecs: 1 })
  try {
    await nowhere.connect()
    console.log('an unreachable broker accepted a connection, which should never happen')
  } catch (error) {
    console.log(`unreachable broker refused: ${(error as Error).message}`)
  }

  return { received, status, goodbye, afterRefusal, afterDisconnect }
}

main()

import asyncio

from pamoja.core import PamojaError
from pamoja.mqtt import MqttClient, MqttWill, Qos

# The broker on the site. The guide's CI runs one on localhost; point these at yours and
# nothing else changes.
BROKER = "127.0.0.1"
PORT = 1883


def connection(connected: bool) -> str:
    return "still connected" if connected else "not connected"


async def main() -> None:
    # The gateway takes every temperature on the site. A `+` stands for exactly one level,
    # so this matches every node's temperature and nothing deeper.
    gateway = MqttClient(
        client_id="site-gateway", host=BROKER, port=PORT, qos=Qos.AT_LEAST_ONCE
    )
    await gateway.connect()
    await gateway.subscribe("sensors/+/temperature")
    print("gateway   subscribed to sensors/+/temperature")

    # A node publishes under that pattern. At least once has the broker acknowledge each
    # message, where at most once would send it and forget it. It also leaves a will: should
    # it drop off the network without saying goodbye, the broker publishes offline on its
    # status topic for it.
    will = MqttWill("sites/node-1/status", "offline", qos=Qos.AT_LEAST_ONCE, retain=True)
    node = MqttClient(
        client_id="node-1", host=BROKER, port=PORT, qos=Qos.AT_LEAST_ONCE, will=will
    )
    await node.connect()
    await node.publish("sensors/1/temperature", "21.5")
    print("node      published 21.5 to sensors/1/temperature")

    # The gateway receives it with the topic attached, which is how it knows which node
    # sent the reading without the payload having to repeat it.
    received = await gateway.recv()
    print(f"gateway   got {received.text} on {received.topic}")

    # The node says it is up, retained, so a dashboard that opens later sees it at once.
    # `publish_confirmed` returns once the broker acknowledges the message.
    await node.publish_confirmed("sites/node-1/status", "online", retain=True)
    print("node      the broker holds online on sites/node-1/status")

    # A dashboard that subscribes afterwards still gets it, because the broker keeps the
    # last retained message on each topic for whoever subscribes next.
    dashboard = MqttClient(client_id="site-dashboard", host=BROKER, port=PORT)
    await dashboard.connect()
    await dashboard.subscribe("sites/node-1/status")
    status = await dashboard.recv()
    print(f"dashboard {status.topic} is {status.text}")

    # Two days of readings saved at one a minute, sent as one message, make a packet over
    # the connection's 10 KiB limit. The send is refused before anything leaves and the
    # connection stays up; a node that must send it raises the limit on every client that
    # shares the topic, or splits it.
    backlog = ",".join(["21.5"] * (2 * 24 * 60))
    try:
        await node.publish("sensors/1/backlog", backlog)
        print("node      sent an oversized backlog, which should never happen")
    except PamojaError as error:
        print(f"node      backlog refused: {error}")
    after_refusal = await node.is_connected()
    print(f"node      {connection(after_refusal)}")

    # Before it leaves, the node says so itself. A clean disconnect discards the will, which
    # is only for a node that drops off without this goodbye.
    await node.publish_confirmed("sites/node-1/status", "offline", retain=True)
    goodbye = await dashboard.recv()
    print(f"dashboard {goodbye.topic} is {goodbye.text}")

    # Disconnecting leaves the client reusable, so a node that loses its link can
    # reconnect the same object when the broker comes back.
    await node.disconnect()
    after_disconnect = await node.is_connected()
    print(f"node      {connection(after_disconnect)} after disconnecting")
    await gateway.disconnect()
    await dashboard.disconnect()

    # A broker that is not there is reported rather than leaving a client that looks
    # connected, so a retry loop has something to test.
    nowhere = MqttClient(client_id="node-2", host=BROKER, port=1, keep_alive_secs=1)
    try:
        await nowhere.connect()
        print("an unreachable broker accepted a connection, which should never happen")
    except PamojaError as error:
        print(f"unreachable broker refused: {error}")

    return received, status, goodbye, after_refusal, after_disconnect


received, status, goodbye, after_refusal, after_disconnect = asyncio.run(main())

// The broker on the site. The guide's CI runs one on localhost; point these at
// yours and nothing else changes.
const string Broker = "127.0.0.1";
const ushort Port = 1883;

static string Connection(bool connected) =>
    connected ? "still connected" : "not connected";

// The gateway takes every temperature on the site. A `+` stands for exactly one
// level, so this matches every node's temperature and nothing deeper.
await using var gateway = new MqttClient(new MqttClientOptions
{
    ClientId = "site-gateway",
    Host = Broker,
    Port = Port,
    Qos = Qos.AtLeastOnce,
});
await gateway.ConnectAsync();
await gateway.SubscribeAsync("sensors/+/temperature");
Console.WriteLine("gateway   subscribed to sensors/+/temperature");

// A node publishes under that pattern. At least once has the broker acknowledge
// each message, where at most once would send it and forget it. It also leaves a
// will: should it drop off the network without saying goodbye, the broker
// publishes offline on its status topic for it.
await using var node = new MqttClient(new MqttClientOptions
{
    ClientId = "node-1",
    Host = Broker,
    Port = Port,
    Qos = Qos.AtLeastOnce,
    Will = new MqttWill("sites/node-1/status", "offline") { Qos = Qos.AtLeastOnce, Retain = true },
});
await node.ConnectAsync();
await node.PublishAsync("sensors/1/temperature", "21.5");
Console.WriteLine("node      published 21.5 to sensors/1/temperature");

// The gateway receives it with the topic attached, which is how it knows which
// node sent the reading without the payload having to repeat it.
MqttMessage received = (await gateway.RecvAsync())!;
Console.WriteLine(
    $"gateway   got {received.Text}"
    + $" on {received.Topic}");

// The node says it is up, retained, so a dashboard that opens later sees it at
// once. PublishConfirmedAsync completes once the broker acknowledges the message.
await node.PublishConfirmedAsync("sites/node-1/status", "online", new MqttPublishOptions(Retain: true));
Console.WriteLine("node      the broker holds online on sites/node-1/status");

// A dashboard that subscribes afterwards still gets it, because the broker keeps
// the last retained message on each topic for whoever subscribes next.
await using var dashboard = new MqttClient(new MqttClientOptions
{
    ClientId = "site-dashboard",
    Host = Broker,
    Port = Port,
});
await dashboard.ConnectAsync();
await dashboard.SubscribeAsync("sites/node-1/status");
MqttMessage status = (await dashboard.RecvAsync())!;
Console.WriteLine($"dashboard {status.Topic} is {status.Text}");

// Two days of readings saved at one a minute, sent as one message, make a packet
// over the connection's 10 KiB limit. The send is refused before anything leaves
// and the connection stays up; a node that must send it raises the limit on every
// client that shares the topic, or splits it.
string backlog = string.Join(",", Enumerable.Repeat("21.5", 2 * 24 * 60));
try
{
    await node.PublishAsync("sensors/1/backlog", backlog);
    Console.WriteLine("node      sent an oversized backlog, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"node      backlog refused: {error.Message}");
}

bool afterRefusal = await node.IsConnectedAsync();
Console.WriteLine($"node      {Connection(afterRefusal)}");

// Before it leaves, the node says so itself. A clean disconnect discards the will,
// which is only for a node that drops off without this goodbye.
await node.PublishConfirmedAsync("sites/node-1/status", "offline", new MqttPublishOptions(Retain: true));
MqttMessage goodbye = (await dashboard.RecvAsync())!;
Console.WriteLine($"dashboard {goodbye.Topic} is {goodbye.Text}");

// Disconnecting leaves the client reusable, so a node that loses its link can
// reconnect the same object when the broker comes back.
await node.DisconnectAsync();
bool afterDisconnect = await node.IsConnectedAsync();
Console.WriteLine($"node      {Connection(afterDisconnect)} after disconnecting");

// A broker that is not there is reported rather than leaving a client that looks
// connected, so a retry loop has something to test.
await using var nowhere = new MqttClient(new MqttClientOptions
{
    ClientId = "node-2",
    Host = Broker,
    Port = 1,
    KeepAliveSecs = 1,
});
try
{
    await nowhere.ConnectAsync();
    Console.WriteLine("an unreachable broker accepted a connection, which should never happen");
}
catch (PamojaException error)
{
    Console.WriteLine($"unreachable broker refused: {error.Message}");
}