# Connecting an MQTT device

> Connect a sensor, a controller or a Sparkplug B node to the gateway over MQTT: identity, topics, refusal codes, backpressure.

Concept - Since version 1.7.0 - Flexway v1.7.1 - https://flexway.io/en/docs/developer-mqtt-devices/

> 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](https://flexway.io/en/docs/product-iot-devices/); operations side: [MQTT devices: the listener in proxy](https://flexway.io/en/docs/operations-mqtt-devices/).

## 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 |

```bash
# 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):

```go
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):

```python
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 |