Connecting an MQTT device

A sensor, a controller or a plant gateway connects to the gateway in MQTT 5 or 3.1.1, with any MQTT client, with no code change. Available since 1.7.0.

Product side: Devices and IoT; operations side: MQTT devices: the listener in proxy.

Connecting

Property	Value
Address	mqtts://<host>:8883, or wss://<host>:<port>/mqtt over WebSocket - given by the operator
Versions	MQTT 5 (recommended: precise refusal codes) or 3.1.1
Identity	a device certificate; or a password: an API key (its id as the user name) or a token of your identity provider (any user name)
Client id	free: the gateway writes its own at the broker (<tenant>/<identity>); a second device of the same identity takes the first one's session
Topics	the contract's, under your tenant and your identity (plant/acme/press-07/telemetry)
Keep-alive	MQTT 5: the CONNACK's Server Keep Alive wins; 3.1.1: between 1 s and the operator's max_keepalive (1,200 s by default) - 0 is refused
Connection lifetime	at most max_session_lifetime (24 h by default): then DISCONNECT 0xA0, reconnect
Topic aliases	up to the value the CONNACK announces (MQTT 5)
Size, QoS, in flight	the CONNACK gives Maximum Packet Size, Maximum QoS and Receive Maximum: follow them

# MQTT 5, an API key: its id as the user name, the key as the password
mosquitto_pub -V mqttv5 -h mqtt.plant.example.com -p 8883 --cafile ca.crt \
  -i press-07 -u press-07 -P "$DEVICE_KEY" \
  -t plant/acme/press-07/telemetry -q 1 -m '{"v":21.5}'

# A device certificate instead (no password)
mosquitto_sub -V mqttv5 -h mqtt.plant.example.com -p 8883 --cafile ca.crt \
  --cert device.crt --key device.key -t 'plant/acme/press-07/cmd' -q 1

Go (paho.golang, MQTT 5, automatic reconnection):

u, _ := url.Parse("tls://mqtt.plant.example.com:8883")
cm, err := autopaho.NewConnection(ctx, autopaho.ClientConfig{
	ServerUrls:      []*url.URL{u},
	TlsCfg:          &tls.Config{RootCAs: pool},
	KeepAlive:       30,
	ConnectUsername: "press-07",
	ConnectPassword: []byte(deviceKey), // or a token of your identity provider
	WillMessage:     &paho.WillMessage{Topic: "plant/acme/press-07/state", QoS: 1, Payload: []byte("offline")},
	ClientConfig:    paho.ClientConfig{ClientID: "press-07"},
})
if err != nil {
	return err
}
pr, err := cm.Publish(ctx, &paho.Publish{Topic: "plant/acme/press-07/telemetry", QoS: 1, Payload: []byte(`{"v":21.5}`)})
if pr != nil && pr.ReasonCode >= 0x80 {
	// refused by the gateway: 0x87 not authorized, 0x99 payload invalid, 0x97 quota exceeded
}

Python (paho-mqtt 2):

import ssl
import paho.mqtt.client as mqtt

c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="press-07", protocol=mqtt.MQTTv5)
c.tls_set(ca_certs="ca.crt")                       # add certfile=, keyfile= for a device certificate
c.username_pw_set("press-07", device_key)          # or a token as the password
c.will_set("plant/acme/press-07/state", "offline", qos=1)

def on_publish(client, userdata, mid, reason_code, properties):
    if reason_code.is_failure:                     # 0x87, 0x99, 0x97...
        print("refused:", reason_code)

c.on_publish = on_publish
c.connect("mqtt.plant.example.com", 8883, keepalive=30)
c.loop_start()
c.publish("plant/acme/press-07/telemetry", '{"v": 21.5}', qos=1)

Publishing, subscribing

• An admitted PUBLISH gets its PUBACK after the broker's: the gateway holds nothing.
• A filter may carry a wildcard inside your channel (plant/acme/fleet/#), never beyond it (fleet/#); $share/<group>/<filter> only where the channel admits it - your group never meets another tenant's.
• The will is judged at CONNECT like a publication: on a topic you may publish to.
• A Response Topic must be a topic you may subscribe to.

Rates: you are slowed, never refused

Past your rate (messages or bytes per second) or your tenant's, the gateway stops reading your connection for as long as needed: your PUBLISH go out more slowly, none is refused, the connection is not cut and your will is not published. A client that waits for its PUBACK (QoS 1) paces itself. Only your tenant's month or day quota refuses (0x97).

Shutdowns and reconnections

When a replica shuts down, MQTT 5 receives DISCONNECT 0x9C and a Server Reference; 3.1.1 is just closed. Reconnect with the same identity within 30 seconds (the Server Reference, or the same address): your session resumes at the broker. So that your will is not published at the takeover either, declare in MQTT 5 a Will Delay Interval longer than your reconnection and a Session Expiry Interval that covers it; without a delay (and always in 3.1.1), the broker publishes it when your new connection takes the session over. Checked end to end with paho.golang (autopaho) and paho.mqtt.golang with automatic reconnection, which come back through the same address.

Sparkplug B and OPC UA PubSub JSON

A Sparkplug B edge node (Eclipse Tahu, Ignition Edge, a plant gateway) connects like any device, with the topics of the spBv1.0 namespace:

Property	Edge node (edge_node)	Host application (host, state)
Group	Your tenant's name (spBv1.0/acme/...)	Same
Id	The edge node id = your identity: the key's id, the token's subject or the certificate field the operator bound	-
Publishes	spBv1.0/acme/NBIRTH/<node>, NDATA, NDEATH, and its devices' DBIRTH, DDATA, DDEATH	The group's NCMD, DCMD; its STATE on spBv1.0/STATE/acme - the host id is the tenant's name
Receives	Its NCMD, DCMD; the STATE of its tenant's host	The whole group
Will	Its NDEATH (with its NBIRTH's bdSeq), nothing else	Its STATE {"online": false, ...}, nothing else
Payload	Sparkplug B Protobuf (sparkplug_b.proto), checked if the operator chose so	STATE in JSON {"online": true, "timestamp": ...}

• The gateway never requests a rebirth and emits no Sparkplug message: that is your host application's job.
• A seq or bdSeq gap is counted, never refused.
• After a DISCONNECT 0x9C, reconnect with the same id: the session resumes, the node stays online.

What you receive	Why	What to do
CONNACK 0x87	The will is not the node's NDEATH (or the host's STATE)	Declare the NDEATH as the will
PUBACK 0x87	A type your role does not publish (a birth by the host, a command by the node), or another group or node	Publish under your group and your id
PUBACK 0x99	Payload outside sparkplug_b.proto, or STATE outside its schema	Fix the encoding

An OPC UA PubSub publisher sends its NetworkMessages in JSON (Part 14) on <prefix>/json/data/<PublisherId>/<WriterGroup>: the prefix carries your tenant (opcua/acme/...), the PublisherId is your identity; each message is checked by the contract's schema (0x99 otherwise). The UADP binary encoding is not served.

Refusal codes

What you receive	Why	What to do
CONNACK 0x86 (3.1.1 0x04)	Key or token refused; a user name that is not the key's id	Fix the credential; do not retry in a loop
CONNACK 0x87 (3.1.1 0x05)	The identity opens no topic, or the will targets a topic it may not publish to	Ask the operator for the operations
CONNACK 0x88, 0x89 (3.1.1 0x03)	The broker is unavailable	Retry with backoff
CONNACK 0x9C / DISCONNECT 0x9C + Server Reference	The replica shuts down or is full	Reconnect to the Server Reference, or to the same address
CONNACK 0x9F	Too many connections per second for this identity or tenant	Wait a second at least
CONNACK 0x97	The replica, your tenant or your identity (per_client_id) holds its maximum of connections	Close a connection, or retry with backoff
3.1.1: return code 0x05	A keep-alive of 0, or above the operator's maximum	Set a keep-alive (30 to 300 s is common)
PUBACK 0x87	Topic outside your channels, or another device's, or a forbidden Response Topic	Publish under your tenant and identity
PUBACK 0x99	The payload does not follow the contract's schema, or is declared UTF-8 and is not	Fix the message
PUBACK 0x97	Your tenant's month (or day) quota is spent	Wait for the next window
PUBACK or PUBREC 0x9B	QoS above what the channel admits	Publish at a lower QoS
SUBACK 0x87 (3.1.1 0x80)	A filter wider than your channels (plant/#) or another tenant's	Narrow the filter
DISCONNECT 0xA0	Your token (or key, or certificate) expired, or the connection reached its maximum lifetime	Reconnect with a fresh credential
DISCONNECT 0x87	Your key, certificate or operations were withdrawn	Ask the operator; do not retry in a loop
DISCONNECT 0x96	Too many refused packets in a minute (60 by default)	Fix the topics or payloads refused before reconnecting
DISCONNECT 0x8E	Another connection of the same identity took the session	One device per identity, or ask for per_client_id
3.1.1: connection closed after a PUBLISH	MQTT 3.1.1 has no publication refusal code	Read the same table in MQTT 5