Pamoja.Mqtt 0.2.0

Pamoja.Mqtt

An MQTT client with the topic and wildcard rules, acknowledged delivery, retained messages, a last will, and TLS, as the core transport. One capability of pamoja, one memory-safe Rust core with bindings for TypeScript, Python, and C#.

Install

dotnet add package Pamoja.Mqtt

using Pamoja.Mqtt;

This pulls in Pamoja.Native, the compiled engine, and Pamoja.Core. dotnet add package Pamoja is the whole framework in one package.

Example

The guide project's example, spliced here as it ran in CI.

From bindings/dotnet/samples/Pamoja.Guides/MqttGuide.cs:

// 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}");
}

The same capability in every language

Documentation

• Pamoja.Mqtt reference, every type in this namespace.
• The MQTT guide, with the same example in Rust, TypeScript, and Python.
• Every capability, and the install page.

License

MIT