MQTT 3.1.1

Zeus.Protocols.Mqtt targets message-oriented devices, edge gateways, and industrial IoT brokers. It layers MQTT sessions on top of Zeus channels and supports publish/subscribe, QoS, retained messages, wills, keep-alive, reconnects, and direct topic-to-point-table mapping.

Install

dotnet add package Zeus.Communications
dotnet add package Zeus.Protocols.Mqtt

Hardware-Free Testing

Use the virtual broker to verify connection, subscription, publishing, and point write-back locally:

using Zeus;

var memory = new MqttBrokerMemory();
memory.SetText("factory/temperature", "25.3");
memory.SetText("factory/running", "true");

await using var app = ZeusHost.Create(builder =>
{
    builder.AddAcquisition(TimeSpan.FromMilliseconds(500));
    builder.AddVirtualChannel("mqtt-link", new MqttBrokerResponder(memory));
    builder.AddMqtt(
        "gateway",
        "mqtt-link",
        new MqttOptions { ClientId = "zeus-mqtt-demo" },
        points: map => map
            .Double("temperature", "factory/temperature")
            .Boolean("running", "factory/running")
            .Double("setpoint", "factory/setpoint")
            .Writable("setpoint"));
});

await app.StartAsync();
await Task.Delay(600);
Console.WriteLine(app.Points.Get<double>("temperature"));

await app.Points.WriteAsync("setpoint", 18.6);

MqttBrokerMemory stores retained messages for the virtual broker. Publishing an empty payload with retain: true deletes the retained message for that topic.

Connect to a Real Broker

MQTT commonly uses TCP port 1883. Replace only the channel; the device and point map can remain the same:

await using var app = ZeusHost.Create(builder =>
{
    builder.AddTcpClient("mqtt-link", "192.168.1.20", 1883);
    builder.AddMqtt(
        "gateway",
        "mqtt-link",
        new MqttOptions
        {
            ClientId = "zeus-gateway-01",
            Username = "zeus",
            Password = "secret",
            KeepAliveSeconds = 60,
            AutomaticKeepAlive = true,
            AutomaticReconnect = true
        },
        points: map => map
            .Double("temperature", "factory/temperature")
            .Writable("temperature"));
});

await app.StartAsync();

QoS, Retained Messages, and Wills

Client publishing defaults to QoS 0. Select QoS 1 or QoS 2 explicitly when needed:

var gateway = app.Devices.Get<MqttDevice>("gateway");

await gateway.PublishTextAsync(
    "factory/status",
    "online",
    MqttQualityOfService.AtLeastOnce,
    retain: true);

await gateway.Client.SubscribeAsync(
    "factory/#",
    MqttQualityOfService.ExactlyOnce);

var message = await gateway.Client.WaitForMessageAsync("factory/status");

QoS 1 uses PUBACK; QoS 2 uses PUBREC / PUBREL / PUBCOMP. Subscription filters support + and #; publish topics and point topics cannot contain wildcards.

Configure a will with both a topic and a payload:

var options = new MqttOptions
{
    ClientId = "zeus-gateway-01",
    WillTopic = "factory/status",
    WillPayload = "offline"u8.ToArray(),
    WillQualityOfService = MqttQualityOfService.AtLeastOnce,
    WillRetain = true
};

Point Mapping

MqttPointMap supports common message payload types:

API	Payload	Result
Text	UTF-8 text	string
Boolean	true / false or 1 / 0	bool
Int32 / Int64	Decimal text	Integer
Double	Invariant-culture number text	double
Bytes	Raw bytes	byte[]

Numeric points can define alarm limits. Writable points publish using the point's QoS and retain settings:

points: map => map
    .Double("temperature", "factory/temperature",
        new PointAlarmLimits(0, 80))
    .WithQualityOfService("temperature", MqttQualityOfService.AtLeastOnce)
    .Retained("temperature")
    .Writable("temperature")

JSON Configuration

{
  "acquisition": { "intervalMilliseconds": 500, "pollImmediately": true },
  "channels": [
    { "name": "mqtt-link", "type": "virtual", "responder": "mqtt" }
  ],
  "devices": [
    {
      "name": "gateway",
      "channel": "mqtt-link",
      "type": "mqtt",
      "mqttClientId": "zeus-json-gateway",
      "mqttKeepAliveSeconds": 60,
      "mqttAutomaticKeepAlive": true,
      "mqttAutomaticReconnect": true,
      "points": [
        {
          "name": "temperature",
          "topic": "factory/temperature",
          "dataType": "double",
          "mqttQos": "1",
          "mqttRetain": true,
          "lowAlarmLimit": 0,
          "highAlarmLimit": 80
        },
        {
          "name": "running",
          "topic": "factory/running",
          "dataType": "boolean"
        },
        {
          "name": "setpoint",
          "topic": "factory/setpoint",
          "dataType": "double",
          "writable": true
        }
      ]
    }
  ]
}

The device type can be mqtt, mqtt311, or mqtt-3-1-1. Use responder: "mqtt" for a virtual channel. For a real broker, change the channel to tcp and provide host and port.

Troubleshooting

Symptom	First check
CONNECT timeout	TCP address/port, firewall, and whether the broker is listening
CONNACK rejected	ClientId, credentials, and MQTT 3.1.1 support on the broker
No messages received	Topic filter, broker ACL, subscription QoS, and $ system-topic rules
Point has no value	Payload format for the point's dataType and whether the first poll completed
Reconnect does not restore data	AutomaticReconnect, channel returning to Open, and reconnect backoff settings

For field diagnostics, enable channel tracing and inspect CONNECT, SUBSCRIBE, PUBLISH, and acknowledgment packets.

Next: JSON Configuration, Communication Tracing, or Virtual Channels.