Skip to main content

Durable messaging

In this tutorial you give the gateway a stream and two buckets. Then you publish at QoS 1 and 2, retain a message, and keep a session across a gateway restart.

Before you start​

You need:

In each terminal, point the nats CLI at the local server:

export NATS_URL=nats://127.0.0.1:4222

1. Start NATS with JetStream​

nats-server -js -sd ./js-data

The server listens on port 4222 and stores JetStream data in ./js-data. Keep it running.

2. Create the stream and the buckets​

  1. Save this config as gateway.yaml:

    bindings:
      - { subjects: [plant1.*.*.alarms], stream: ALARMS }
    mqtt:
      bindings: [ALARMS]
      retained: { bucket: MQTT_RETAINED }
      sessions: { bucket: MQTT_SESSIONS }

    The binding stores the topics plant1/+/+/alarms in the stream ALARMS. The mqtt block serves MQTT on port 1883.

  2. Create the stream and the buckets that the config names:

    nats stream add ALARMS --subjects 'plant1.*.*.alarms' --deny-delete --deny-purge --defaults
    nats kv add MQTT_RETAINED --history 2
    nats kv add MQTT_SESSIONS --history 1

    The gateway never creates them for you.

  3. Check them:

    ./mq-gateway check --config gateway.yaml --verbose

    Each binding and bucket line ends in ok. The warnings are about size limits. You can ignore them in this tutorial.

3. Start the gateway​

On a licensed build, first set MQGW_LICENSE_FILE to the path of your license file. Then start the gateway:

./mq-gateway --config gateway.yaml

The log ends with msg="account ready". Keep the gateway running.

4. Publish at QoS 1​

  1. Publish an alarm:

    mqttx pub -h 127.0.0.1 -q 1 -t 'plant1/line3/press7/alarms' -m 'pressure low'

    MQTTX prints Message published. The gateway acknowledges the publish only after the stream stores the message.

  2. Read the message from the stream:

    nats stream get ALARMS 1

    The output shows the subject plant1.line3.press7.alarms and the payload pressure low.

  3. Publish at QoS 1 to a topic that no binding stores:

    mqttx pub -h 127.0.0.1 -q 1 -t 'plant1/line3/press7/telemetry' -m '48.5C'

    The gateway refuses the publish. MQTTX prints Publish error: Implementation specific error.

5. Subscribe at QoS 2​

  1. Subscribe:

    mqttx sub -h 127.0.0.1 -q 2 -t 'plant1/+/+/alarms'

  2. In a new terminal, publish at QoS 2:

    mqttx pub -h 127.0.0.1 -q 2 -t 'plant1/line3/press7/alarms' -m 'overtemp'

    The subscriber prints overtemp with qos: 2.

  3. List the consumers of the stream:

    nats consumer ls ALARMS

    The subscriber's session has one consumer.

  4. Stop the subscriber with Ctrl-C.

6. Retain a message​

  1. Publish with the retain flag:

    mqttx pub -h 127.0.0.1 -r -t 'plant1/line3/press7/telemetry' -m '48.5C'

    This topic needs no binding. The retained bucket stores the message:

    nats kv ls MQTT_RETAINED

    The output is plant1.line3.press7.telemetry.

  2. Subscribe after the publish:

    mqttx sub -h 127.0.0.1 -t 'plant1/+/+/telemetry'

    The subscriber prints 48.5C with retain: true.

  3. Stop the subscriber.

7. Keep a session across a restart​

  1. Subscribe as client hmi-1 with a persistent session of one hour:

    mqttx sub -h 127.0.0.1 -q 1 -t 'plant1/+/+/alarms' -i hmi-1 --no-clean -se 3600

  2. Stop the subscriber with Ctrl-C.

  3. Publish an alarm while hmi-1 is away:

    mqttx pub -h 127.0.0.1 -q 1 -t 'plant1/line3/press7/alarms' -m 'while away'

  4. Stop the gateway with Ctrl-C. Then start it again:

    ./mq-gateway --config gateway.yaml

  5. Connect hmi-1 again with the command from item 1.

    The subscriber prints while away. The gateway that held the session stopped, but NATS kept the session in the bucket MQTT_SESSIONS. Its consumer kept the alarm.

  6. Stop all processes with Ctrl-C. To remove the stored data, remove the ./js-data directory.

Next steps​