MQTT read task

Subscribe to MQTT topics and record the values in their payloads in channels.

An MQTT read task subscribes to topics on a broker and records the values in their payloads in channels. One task reads any number of topics, and it can also read Sparkplug B tags. Before you start, connect an MQTT broker.

Entries and fields

An entry is one topic. A field is one value in its payload, found by a JSON pointer such as /temperature, and recorded in one channel.

Bare values -> an empty pointer, shown as Whole payload, reads a topic that carries a bare number or string, such as 42.7.
Text labels -> the Enum mapping of a field maps text, such as ON, to a number.
All or nothing -> the task writes all fields of a payload or none. A payload that lacks a field, or holds a value that the channel cannot store, is skipped.

A topic can be in a task once, and it cannot hold the wildcards + or #. All fields of one topic share one index channel, and two topics cannot write to the same channel.

Create an MQTT read task

In the devices toolbar, right-click the broker and select Create read task. You can also run the “Create MQTT read task” command.
Click Add data. The dialog listens to the broker for a few seconds and lists the topics that it saw.
Select a topic to see the last value and the type of each field, then select the fields to record. For a topic that publishes rarely, click Add a topic by name.
Click Add N channels. Each field becomes a channel.
Select a field to change its Pointer, Data type, or Channel name.
Click (Start).

The Console creates a channel for each field and an index channel for each topic.

Python does not create channels. Create one index channel for each topic first.

import synnax as sy
from synnax import mqtt

broker = client.devices.retrieve(name="Plant broker")
time = client.channels.create(
    name="line1_time", is_index=True, data_type=sy.DataType.TIMESTAMP
)
temperature = client.channels.create(
    name="line1_temperature", index=time.key, data_type=sy.DataType.FLOAT64
)
pressure = client.channels.create(
    name="line1_pressure", index=time.key, data_type=sy.DataType.FLOAT64
)

# The payload of plant/line1/sensors is {"temperature": 23.5, "pressure": 101.3}
task = mqtt.ReadTask(
    name="Line 1 sensors",
    device=broker.key,
    entries=[
        mqtt.PlainReadEntry(
            topic="plant/line1/sensors",
            fields=[
                mqtt.ReadField(pointer="/temperature", channel=temperature.key),
                mqtt.ReadField(pointer="/pressure", channel=pressure.key),
            ],
        )
    ],
)
client.tasks.configure(task)
task.start()

Timestamps

The Source of an entry sets the timestamps of its samples:

Arrival time -> the time at which the payload arrives at the Core. The default.
Payload value -> a time in the payload, at a Pointer, in a Format: ISO 8601, or Unix time in seconds, milliseconds, microseconds, or nanoseconds. The value goes to the index channel of the topic.

Timestamps in an index channel must increase. The task skips a payload whose timestamp is not after the last one it wrote. A broker can send such a payload when it delivers a message twice at quality of service 1.

In code, a payload timestamp is a field on the index channel with a time_format. Put the key of that field in index on the entry:

# The payload is {"value": 0.42, "ts": 1705312200000}
stamp = mqtt.ReadField(
    pointer="/ts",
    channel=sample_time.key,
    data_type="timestamp",
    time_format="unix_ms",
)
entry = mqtt.PlainReadEntry(
    topic="plant/line1/sample",
    index=stamp.key,
    fields=[stamp, mqtt.ReadField(pointer="/value", channel=value.key)],
)

Retained messages

When the task subscribes, the broker sends the retained message of each topic, if it has one. The task records it like any other message, but skips it with no warning when its timestamp is not after the last one written. Turn on Ignore retained to always skip it.

Task options

Setting	Default	Notes
Data saving `data_saving_disabled`	On	See data saving.
Auto start `auto_start`	Off	Starts the task each time the Core starts.

The task has no sample rate. It writes each payload when it arrives.

Entry and field reference

An entry has these settings:

Setting	Default	Notes
Topic `topic`		The topic to subscribe to, such as `plant/line1/sensors`.
Quality of service `qos`	At most once	At most once (0), at least once (1), or exactly once (2).
Ignore retained `retained_ignored`	Off	Skips the retained message of the topic.
Source `index`	Arrival time	The key of the timestamp field. Empty is arrival time.

In code, an entry is PlainReadEntry, of type "plain". The qos values are "at_most_once", "at_least_once", and "exactly_once".

A field has these settings:

Setting	Default	Notes
Pointer `pointer`	Empty	Empty reads the whole payload.
Data type `data_type`	`float64`	The data type of the channel.
Format `time_format`	None	Required for a `timestamp` channel.
Channel `channel`		The channel that records the value.
Enum mapping `enum_values`	None	Maps text labels in the payload to numbers.

An enum mapping in code is a list of mqtt.EnumEntry(label="ON", value=1).

Troubleshooting

Skipped ... of ... messages
Cause
The task read some messages that it cannot record. The description gives the topic and the reason for the last one, such as field ...: no value at /temperature, or a timestamp that is not after the last one written. The warning clears 5 seconds after the last skip.
Fix
Check the pointers and data types against a payload from the topic. Pointers start with / and are case sensitive.
messages arrive faster than the task can write them, and the oldest are dropped
Cause
The task holds up to 4096 messages that wait to be written. More messages arrived, so the task dropped the oldest ones.
Fix
Lower the rate at which the devices publish to the topics of the task.
the broker refused the subscription to ...
Cause
The broker does not let the user subscribe to the topic, for example because of an access rule. At start, the task does not start. After the connection returns, the task keeps running and shows this warning.
Fix
Give the user access to the topic in the broker, then start the task again.
broker is not connected
Cause
The connection to the broker is down. The Core connects again on its own, and the warning clears when the connection returns.
entries: topic ... appears more than once entries: ... and ... write to the same channel ...
Cause
Two entries read the same topic, or two fields write to the same channel.
Fix
Move the fields of a topic into one entry, and give each field its own channel.
entries: the task has no enabled entries
Cause
The task has no entries, or each entry is disabled.
Fix
Add an entry, or enable one.
device: a broker is required
Cause
The task has no broker. In code, device is empty.
Fix
Set device to the key of the broker.

For problems with the connection, see Connect an MQTT broker.

Connect to a broker

Write task