Sparkplug B

Sparkplug B is a standard for running industrial devices over MQTT: the topic
names, the payload format, and the rules for saying "I am here" and "I have
gone". This library has both ends of it, conformant to Sparkplug 3.0:

Both are in #include <ModBeeMqtt.h>, and neither costs anything in a program
that does not construct one.

Examples: Sparkplug_Edge_Node
(ESP32), sparkplug_node and
sparkplug_host (a host, and a pair that
talk to each other).

What Sparkplug adds to MQTT

Plain MQTT moves bytes on topics anyone can invent. Sparkplug fixes the rest:

• Births. When a node connects, it publishes a birth certificate
  (NBIRTH) listing every metric it has, with its datatype, units and current
  value. A SCADA system such as Ignition builds its tag list from that, so
  nobody types tag names in by hand.
• Deaths. The node registers a death certificate (NDEATH) as its MQTT
  Will. If it drops off the network, the broker publishes it, and every value
  from that node is marked stale at once rather than left looking live.
• Sequence numbers. Every message is numbered. A host that sees a gap
  asks the node to publish its birth again (a rebirth).
• Aliases. After the birth, metrics are sent by number rather than by
  name, which is most of Sparkplug's bandwidth saving. A host keeps the birth
  to turn the numbers back into names; MqttSparkplugHost does that for you.
• Report by exception. Data messages (NDATA) carry only what changed.
• A primary host. A node can be told to stay quiet until a named SCADA
  host says it is online, and to move to another broker if that host goes
  offline.

The topics are spBv1.0/<group>/<message type>/<node>[/<device>], for example
spBv1.0/Plant/NDATA/Tank-01. The payload is Protocol Buffers; this library
encodes and decodes it itself, with no protobuf library.

An edge node

#include <WiFi.h>
#include <ModBeeMqtt.h>

WiFiClient net;
MqttSparkplugNode node(net);   // or EthernetClient, TinyGsmClient, WiFiClientSecure

void setup() {
    WiFi.begin("plant-wifi", "");
    while (WiFi.status() != WL_CONNECTED) delay(250);

    node.epochNow(1767225600000ULL);   // UTC milliseconds - see "The clock"

    node.group("Plant");
    node.node("Tank-01");
    node.server("192.168.1.10");

    node.declare("Tank/Level", MqttSparkplugDataType::Float, "m");
    node.declare("Pump/Running", MqttSparkplugDataType::Boolean);
    node.declareWritable("Tank/Setpoint", MqttSparkplugDataType::Float);

    node.connect();                    // returns at once
}

void loop() {
    node.poll();
    node.set("Tank/Level", mq_spFloat(analogRead(A0) * 0.001f));
    node.set("Pump/Running", mq_spBool(digitalRead(5)));
    node.publishChanged();             // one NDATA, with only what moved
}

On a host there is no network object to pass: MqttSparkplugNode node; uses
the computer's own sockets.

Declare everything first

The birth must list every metric the node will ever publish, so each one
is declared before connect(). Writing to a name that was not declared is
refused, because a SCADA system discards metrics it was never told about.
declare() returns the metric's alias, or 0 if it was refused: a duplicate
name, an unsupported datatype, or a full table.

set() must be given a value of the declared type: mq_spFloat() for a
Float, mq_spDouble() for a Double, and so on. A mismatch is refused
rather than converted, because the birth told the SCADA system what type to
expect. The makers are mq_spBool, mq_spInt8 to mq_spInt64, mq_spUInt8
to mq_spUInt64, mq_spFloat, mq_spDouble, mq_spString, mq_spText,
mq_spDateTime, mq_spBytes and mq_spNull(type).

Sending

• publishChanged() sends one NDATA with the metrics whose value moved
  since they were last sent, and one DDATA for each online device that has
  some. Call it on a timer (once a second in the examples) or every loop;
  with nothing changed it sends nothing. It looks only at the metrics that
  were marked - by set()/setAt() changing a value, a host's write, or
  reportAt() - and a metric stays marked while it differs from what was
  last sent (inside its deadband, or held), so a call costs what moved, not
  what is declared. A text value rewritten in place in the same buffer is
  seen when it is set() again, as the OpenPLC blocks do every scan.

• publishAtPoll() asks for one publishChanged() at the next poll().
  For a program that sets values in many places - the OpenPLC blocks: each
  asks, and the node sends one NDATA (and one DDATA a device) per poll for
  all of them. pollPublishes() counts those made and pollPublishSent()
  says whether the last one went; a caller keeps the count from when it
  asked to know its message has been made.

• A deadband decides what "moved" means for a noisy number:

  node.deadband("Tank/Level", 0.01);          // ignore moves of 10 mm or less
  node.deadband("Line/Pressure", 0, 2.0);     // ignore moves of 2 % or less
  node.deadbandFor("Pump-2", "Speed", 5.0);   // a device's metric
  

  A move is measured from the value last published, not the last one set,
  so a slow drift still adds up and goes. A value a host wrote is always
  reported back, deadband or not - the host watches for that echo to know the
  write took. With both numbers set, a move must
  clear both. Only numeric metrics take one (a Boolean is refused - a
  deadband would freeze the bit), and a birth or publishAll() always
  carries every value as it is.

• publishAll() sends every metric, changed or not.

• rebirth() publishes the birth again. A host asks for this with a
  rebirth command, and the node does it by itself.

The clock

Every Sparkplug message carries a UTC timestamp, and the specification
requires it. On a host the system clock is used and nothing needs doing.

On an ESP32, ESP8266 or RP2040 the board's own system clock is used as soon as
something has set it - so starting SNTP is enough, and the node needs no call:

configTime(0, 0, "pool.ntp.org");                // ESP32 and ESP8266

A modem's network time, or an RTC read at start-up, sets it the same way:
mq_setEpochMillis(ms) (or settimeofday()). Until the clock reads a date
after 2024 it is treated as not set.

On any other board, give the node the time once, in milliseconds since 1970:

node.epochNow(rtcMillisSince1970);

The node adds its own uptime from then on. Without a clock,
node.clockIsSet() is false and the timestamps count from boot - a date in
January 1970, which a host should read as "not synced" and replace with the
time it received the message.

Units and ranges

The birth can say what a number means: engUnit, and the engLow/engHigh
range a SCADA gauge or trend should show (the property names Eclipse Tahu and
Ignition use).

node.declare("Tank/Level", MqttSparkplugDataType::Float, "m");   // units at declaration
const int level = node.indexOf(nullptr, "Tank/Level");
node.unitsAt(level, "m");
node.rangeAt(level, 0.0, 4.5);

They travel only in the birth, so a change shows at the next one.

Many metrics from one block

A wrapper binding a whole structure looks each metric up once and keeps its
index - indexOf(device, name) - then works by index every scan: setAt(),
valueAt(), changePendingAt(), commandPendingAt(), takeAt(),
deadbandAt(). The
index is stable for the life of the node; every call refuses one past
pointCount(). setAt() answers whether the value changed (false for the
same value again, and for a refusal). Names are found through a hash index
built as metrics are declared, so declaring a thousand costs a thousand
lookups, not half a million compares.

Two more by index, for a set sent on its own clock:

node.holdAt(elapsed, true);       // only when asked: not with every other report
node.reportAt(elapsed);           // ask: it goes in the next publishChanged()

A held metric is left out of publishChanged() unless reportAt() marked
it - for a set sent on its own clock, every period or on request. Births and
publishAll() carry it as it is. isHeldAt() says whether one is held.

What a metric is, and what it holds. pointAt(index) (and findPoint())
answer a metric's MqttSparkplugPoint: its name, type, alias, device, units,
range, deadband and write queue. Its value is not there - read it with
valueAt(index). The value, the one last sent and the marks that decide
whether it goes are a separate table of 32 bytes a metric, one cache line on
an ESP32, so a scan that sets a thousand values reads a thousand lines and
not five thousand.

Writes from SCADA

A metric declared with declareWritable() can be written by a host (an NCMD).
A write to any other metric is refused and counted in commandsRefused(). Take
writes with a callback:

void onCommand(const MqttSparkplugCommand& command) {
    if (strcmp(command.name, "Tank/Setpoint") == 0) {
        setpoint = (float)command.value.asNumber();
    }
}

node.onCommand(&onCommand);

or by polling, which suits a PLC scan:

MqttSparkplugValue value;
if (node.take("Tank/Setpoint", value)) setpoint = (float)value.asNumber();

// A device's metric (a DCMD) - a name is only unique within its device:
if (node.takeFor("Pump-2", "Enable", value)) enable = value.as.boolean;

The written value is the metric's value from then on, and the next
publishChanged() reports it back, so the SCADA system sees the write take
effect.

By default a write replaces the one before it, so a host writing 1 and then 0
faster than the application polls is seen to write 0 only. A PLC wants every
write, in order:

node.queueWrites(4);            // before the first write; 1..16 per metric
// each take() / takeAt() now hands over the OLDEST write, one at a time;
// peekAt() looks at it without taking it

Full, a metric's queue drops its oldest write and keeps the newest, and
counts the drop (writesLost(), writesLostAt(index)). A queued write
becomes the metric's value when it is taken, and is reported back then. Text
is copied, so a queued STRING survives the packets that follow it.

A write the node refuses is not given to onCommand() - so an application
that acts on what its handler is given never acts on one - but it can be seen,
with the reason, through onCommandRefused():

void onRefused(const MqttSparkplugCommand& command) {
    // command.outcome: UnknownDevice, UnknownMetric, ReadOnly, WrongType,
    // Undecodable or NotARequest - mq_sparkplugCommandOutcomeName() names it
    log(command.name, mq_sparkplugCommandOutcomeName(command.outcome));
}

node.onCommandRefused(&onRefused);

Between the two handlers every metric of every NCMD and DCMD is reported
exactly once; command.applied() says which side it came from.

Devices

A node can speak for devices behind it, each with its own metrics, its own
birth (DBIRTH) and death (DDEATH):

node.addDevice("Pump-2");
node.declareFor("Pump-2", "Speed", MqttSparkplugDataType::Float);
node.declareWritableFor("Pump-2", "Enable", MqttSparkplugDataType::Boolean);
// after connect():
node.deviceOnline("Pump-2");                     // DBIRTH, once the node has birthed
node.setFor("Pump-2", "Speed", mq_spFloat(1450.0f));
node.deviceOffline("Pump-2");                    // DDEATH, when the device stops answering

On a connection you already have

A device rarely has only Sparkplug to say: its own topics and its firmware
updates want the same broker session. Give the node the client instead of a
network, and it becomes a layer on that connection (see "Layers sharing one
client" in api.md):

MqttSparkplugNode node(client);    // client: begun, connected and polled by its owner
node.group("Plant1");
node.node("pump-01");
node.declare("level", MqttSparkplugDataType::Float);
node.connect();                    // starts the node on the shared session

// every pass, after client.poll():
node.poll();

The owner keeps the connection: server(), credentials(), keepAlive(),
only() and the TLS calls do nothing here, and node.poll() does not poll the
client. What Sparkplug needs of the session the node takes on every CONNECT:
the Will (its NDEATH - claimed, and refused while another layer owns it),
Clean Start, no session expiry, and a transmit buffer that holds the birth
certificate. Started on a session that is already up, it restarts the client
once so the next CONNECT carries the NDEATH; a primary host going offline
restarts it too. disconnect() publishes the NDEATH and leaves the connection
to everything else on it. MqttSparkplugHost takes a client the same way.

A primary host, and more than one broker

node.primaryHost("scada-1");                     // wait for this host before birthing
node.server("broker-a.plant");
node.addServer("broker-b.plant");                // up to four

With a primary host set, the node connects but does not birth until host
scada-1 says it is online. If that host goes offline, the node disconnects
and connects to the next broker in the list, as the specification
requires. If the host is replaced and the node sees the new session's online
(a newer timestamp) without the old one's offline, it treats the missed
offline as valid: NDEATH, reconnect to the same broker, and a new NBIRTH for
the new host. Without a primary host, the node births as soon as it connects,
which is what a single-broker site wants.

node.waitingFor() says, in words, why a connected node has not birthed yet:
for example no STATE from primary host 'scada-1'.

TLS

• On a board, TLS belongs to the network object: pass a
  WiFiClientSecure (with its CA certificate set) or an SSLClient, and give
  the server port 8883: node.server("mqtt.example.com", 8883).
• On a host, before connect():

node.useTls();                                   // verify against the system's certificates
node.trustFile("/etc/plant/ca.pem");             // or against this CA
node.pinnedKey("23c3ca67edd89db88be43bc88bd9806588f1766bd272e9e701c659365eb3ca1a");

Each makes the default port 8883. pinnedKey() trusts exactly one broker
key, which is what a plant broker with a self-made certificate wants;
security.md explains it and how to read the key off the
broker.

Knowing what is going on

Changing settings while running

server(), credentials() / clearCredentials(), group(), node(),
primaryHost(), only() and keepAlive() may be called again after a
disconnect(); the next connect() uses them and births with the next bdSeq.
rebind(MqttTransport&) moves the node to another transport you built - the
declarations, aliases, devices and write queues stay. With 5.0 the
disconnect() before it makes the broker publish the NDEATH.

Records with their own time, and store-and-forward

struct MqttSparkplugRecord { size_t index; MqttSparkplugValue value; uint64_t timestamp; };
bool publishRecords(const MqttSparkplugRecord* records, size_t count, bool historical);
uint64_t timestamp() const;   // the stamp the node would use now

publishRecords() sends values outside report by exception, each metric with
its own timestamp, in one NDATA (or DDATA: all records belong to one device or
all to the node). With historical every metric carries is_historical - the
specification's store-and-forward flag: "it may be desirable to send metrics
after they were acquired ... for batching, store and forward, or sending local
backup data during network communication losses" - so the host files them at
their own time and does not update the live tags. Without it each value
becomes the metric's live value and its last published one, so report by
exception does not send it twice. QoS 0 like every data message: keep what
must not be lost until you have seen the node stay born afterwards.

MqttSparkplugHistory is the bounded ring that goes with it: begin(depth)
allocates depth events of 24 bytes with mq_allocLarge() (PSRAM first),
add() drops the oldest when full and counts it in lost(), peek() /
drop() take them oldest first, and placeOnClock(utcNow, uptimeNow) moves
events recorded before the clock was set onto UTC. The OpenPLC block
SPARKPLUG_HISTORY is built on the two.

A host application

#include <ModBeeMqtt.h>

MqttSparkplugHost host;

void onMessage(const MqttSparkplugInbound& message) {
    // message.type: NBIRTH, NDATA, NDEATH, DBIRTH, ...
    // message.payload->metrics[i].name is filled in, even when the node sent an alias
}

int main() {
    host.hostId("scada-1");
    host.server("192.168.1.10");
    host.watch();                      // every group; or host.watch("Plant")
    host.onMessage(&onMessage);
    host.connect();

    for (;;) {
        host.poll();

        MqttSparkplugValue level;
        if (host.read("Plant", "Tank-01", "Tank/Level", level)) {
            // the latest value, by name
        }
    }
}

What the host does for you:

• Announces itself on spBv1.0/STATE/<host id>, with a retained Will
  that says offline, so nodes waiting for it as their primary host start and
  stop with it.
• Keeps every node's metric list from its birth, so data sent by alias
  arrives with names, and read() and readDevice() return the latest value
  of any metric by name. findNode() gives the whole picture for one node:
  online or not, its metrics, their types and the time of each value.
• Checks sequence numbers. With actAsPrimary(), a gap that does not fill
  within reorderTimeout() (5 seconds by default) makes it ask the node for a
  rebirth.

To write to a node:

host.command("Plant", "Tank-01", "Tank/Setpoint", mq_spFloat(3.5f));
host.commandDevice("Plant", "Tank-01", "Pump-2", "Enable", mq_spBool(true));
host.requestRebirth("Plant", "Tank-01");

The value must be the metric's declared type; the birth says what that is
(host.findNode(...)->find(name)->type), and the
sparkplug_host example converts with it.

Each host needs its own host id. It is also the MQTT client id, so two
hosts sharing one keep taking over each other's connection.

On a host computer MqttSparkplugHost host; uses the system's sockets, and
useTls(), trustFile() and pinnedKey() work as for the node. On a board,
pass the network object as for the node.

What is supported

Datatypes: Int8 to Int64, UInt8 to UInt64, Float, Double,
Boolean, String, Text, UUID, DateTime, Bytes, File, and metric
properties with scalar values (engineering units and quality live there).

Not supported yet: DataSet, Template, PropertySetList and the
array types. They are part of Sparkplug 3.0; this library does not carry
their values yet.

• A node cannot declare one - declare() refuses it.
• A host still sees a node that sends one (an Ignition UDT is a
  Template). The rest of the message is read as usual, and the metric itself
  is kept with its name, alias and datatype, valueUnsupported set and its
  value NULL - never a zero. The host's dictionary lists it; read() finds no
  value for it.

Data without datatypes. The specification says data and command
messages SHOULD NOT carry a metric's datatype, because the birth declared it.
This library's node includes it anyway (which is allowed), and both ends
accept messages without it: the host applies the type from the node's birth,
and the node the type it declared. A value that arrives in a field that
cannot hold that type is NULL rather than guessed.

Limits. A node's tables grow with what it declares, and its payloads and
send buffer are sized to its own birth certificate when it first connects -
so nothing is allocated in the steady state, and a node with five metrics pays
for five. On an ESP32 with PSRAM all of it is in PSRAM.

No fixed number of metrics or devices, on a host or a board: a node
declares as many metrics, under as many devices, as memory holds, and a
declaration memory cannot take is refused and node.outOfMemory() says so.
node.metricLimit(n) sets a limit of your own. A host likewise tracks as many
nodes, and metrics per node, as memory holds, and its decode payload grows to
the largest birth it meets (host.outOfMemory() when it could not).

To change one, set it for the whole build, not with a #define in the
sketch: the library's own files are compiled separately and would not see it.
That means -DMQ_SP_MAX_METRICS=64 in CMake, PlatformIO's build_flags, or
the Arduino IDE's compiler.cpp.extra_flags in platform.local.txt. An
OpenPLC project has no build flags of its own: there a define comes from the
board package.

What it costs on an ESP32: the node's tables are allocated as metrics are
declared, so its static RAM is a few hundred bytes. At run time a metric costs
320 bytes in the registry (288 for what it is, 32 for its value and state), a
text metric 280 more for the text it owns, and a writable one 80 for its write
queue (624 for a text metric) - all allocated with mq_allocLarge(), so in
PSRAM where the board has it. The outbound payload, the encode buffer and the
client's send buffer are sized to the birth certificate on top.

Sparkplug and the Kafka pipe

To Kafka, Sparkplug is MQTT like any other: the pipe passes the
topic and the protobuf bytes through untouched, in both directions. One
setting matters: key the records by group and node, so all of one node's
messages land in one Kafka partition and stay in order:

    - id: sparkplug
      mqtt-topic-filters: ["spBv1.0/#"]
      kafka-topic: mqtt-sparkplug
      key: "${MQTT_TOPIC_LEVEL_1}/${MQTT_TOPIC_LEVEL_3}"

How it is tested

Three independent checks, because tests written from the same reading of the
specification as the code would share its mistakes:

• The Eclipse Sparkplug TCK, the specification project's own test suite.
• Other implementations' codecs: Eclipse Tahu (Python) and the npm
  sparkplug-payload decode what this library encodes, and the reverse.
• A live node and host, on a real broker, judged by an independent Python
  host application.

Every requirement in the code names the statement of the specification it
implements, by its tck-id. extending.md has the
design history.