sparkplugb

README

go-sparkplugb

Production-quality Eclipse Sparkplug B 3.0.0 library for Go.

go-sparkplugb handles the full Sparkplug B 3.0.0 protocol lifecycle for Edge Nodes, Host Applications and Applications: topic namespace, Protobuf codecs, typed metrics (including arrays, templates and datasets), sequence numbering, alias management, birth/death certificates, primary host awareness with server walking, and message reordering. Plug in any MQTT 3.1.1 or 5.0 client through the transport interface.

Quick links: Install · Edge Node · Host Application · Application · Packages · Examples · Conformance · Related projects

---

Features

Category	Details
Edge Node runtime	Full lifecycle (birth/death/rebirth), sequence numbering, alias support, primary host awareness, server walking across multiple brokers
Host Application runtime	STATE birth/death certificates, node/device registry, sequence validation, automatic rebirth requests, message reordering timeout
Application runtime	Non-primary-host consumer: subscribes to spBv1.0/#, tracks nodes/devices, alias resolution, sends NCMD/DCMD — no STATE messages
Typed metric codec	All scalar types, signed-integer handling (Int8–Int64 in wire fields), all array types (little-endian packed, bit-packed BooleanArray, null-terminated StringArray, DateTimeArray), quality codes
Templates	Template definitions (is_definition, version, member metrics, parameters) and instances (template_ref, parameter overrides)
DataSets	Columnar data with typed columns (num_of_columns, columns, types, rows), round-trip encode/decode
Topic namespace	Parser and builder for spBv1.0/{group}/{type}/{node}/[{device}] with validation
Protobuf types	Generated from the canonical Eclipse Tahu sparkplug_b.proto (pinned commit, SHA-256 verified)
MQTT transport	Interface-based — plug in Paho, EMQX, or any MQTT client

Installation

go get github.com/otfabric/go-sparkplugb

Requires Go 1.24 or later.

Packages

Package	Import	Description
proto	github.com/otfabric/go-sparkplugb/proto	Generated Protobuf types (EPL-2.0)
topic	github.com/otfabric/go-sparkplugb/topic	Topic namespace parsing, construction and validation
payload	github.com/otfabric/go-sparkplugb/payload	Typed metric encoding/decoding, templates, datasets, arrays, quality codes
edge	github.com/otfabric/go-sparkplugb/edge	Edge Node runtime (lifecycle, sequencing, aliases, server walking)
host	github.com/otfabric/go-sparkplugb/host	Host Application runtime (STATE, registry, seq validation, reorder timeout)
app	github.com/otfabric/go-sparkplugb/app	Application runtime (subscriber, node tracking, commands — no STATE)
mqtt	github.com/otfabric/go-sparkplugb/mqtt	MQTT transport interface and Paho adapter

---

Quick Start — Edge Node

package main

import (
    "context"
    "log"

    "github.com/otfabric/go-sparkplugb/edge"
    "github.com/otfabric/go-sparkplugb/mqtt"
    "github.com/otfabric/go-sparkplugb/payload"
    pb "github.com/otfabric/go-sparkplugb/proto"
)

func main() {
    transport := mqtt.NewPahoTransport()

    node, _ := edge.New(transport, edge.Config{
        GroupID:    "plant",
        EdgeNodeID: "gateway-01",
        BrokerURL:  "tcp://broker:1883",
    })

    ctx := context.Background()
    node.Connect(ctx)
    defer node.Close(ctx)

    temp, _ := payload.NewMetric("temperature", pb.DataType_Double, float64(22.5))
    node.PublishBirth(ctx, []*pb.Payload_Metric{temp})

    // Report by exception
    updated, _ := payload.NewMetric("temperature", pb.DataType_Double, float64(23.1))
    node.PublishNodeData(ctx, []*pb.Payload_Metric{updated})
}

Quick Start — Host Application

app, _ := host.New(transport, host.Config{
    HostID:    "scada-01",
    BrokerURL: "tcp://broker:1883",
    OnNodeBirth: func(group, node string, metrics []*pb.Payload_Metric) {
        log.Printf("node online: %s/%s with %d metrics", group, node, len(metrics))
    },
    OnNodeDeath: func(group, node string) {
        log.Printf("node offline: %s/%s", group, node)
    },
})

app.Connect(ctx)
defer app.Close(ctx)

Quick Start — Application

An Application subscribes to the Sparkplug namespace and tracks edge nodes/devices without publishing STATE birth/death certificates. Use it for dashboards, loggers, analytics pipelines, or any consumer that does not need to be a Primary Host.

application, _ := app.New(transport, app.Config{
    BrokerURL: "tcp://broker:1883",
    OnNodeBirth: func(group, node string, metrics []*pb.Payload_Metric) {
        log.Printf("node online: %s/%s with %d metrics", group, node, len(metrics))
    },
    OnNodeData: func(group, node string, metrics []*pb.Payload_Metric) {
        log.Printf("data from %s/%s: %d metrics", group, node, len(metrics))
    },
    OnNodeDeath: func(group, node string) {
        log.Printf("node offline: %s/%s", group, node)
    },
})

application.Connect(ctx)
defer application.Close(ctx)

// Send a command to an edge node
application.SendNodeCommand(ctx, "plant", "gw1", []*pb.Payload_Metric{rebirth})

Quick Start — Templates

// Define a template with member metrics and parameters.
tempMetric, _ := payload.NewMetric("temperature", pb.DataType_Double, float64(0))
unitParam, _ := payload.NewTemplateParameter("unit", pb.DataType_String, "celsius")

def, _ := payload.NewTemplateDefinition("SensorTemplate", "1.0",
    []*pb.Payload_Metric{tempMetric},
    []*pb.Payload_Template_Parameter{unitParam})

// Create an instance with parameter overrides.
instMetric, _ := payload.NewMetric("temperature", pb.DataType_Double, float64(72.5))
instParam, _ := payload.NewTemplateParameter("unit", pb.DataType_String, "fahrenheit")

inst, _ := payload.NewTemplateInstance("sensor1", "SensorTemplate",
    []*pb.Payload_Metric{instMetric},
    []*pb.Payload_Template_Parameter{instParam})

Quick Start — DataSets

columns := []string{"name", "value", "active"}
types := []pb.DataType{pb.DataType_String, pb.DataType_Double, pb.DataType_Boolean}
rows := [][]any{
    {"sensor1", float64(22.5), true},
    {"sensor2", float64(33.1), false},
}

m, _ := payload.NewDataSet("readings", columns, types, rows)

// Extract after decode
cols, dts, gotRows, _ := payload.GetDataSetValues(m)

---

Examples

Runnable examples are in the examples/ directory:

Example	Description
examples/edge-node	Edge Node with birth, device birth, and periodic data publishing
examples/host-app	Host Application with node/device lifecycle callbacks
examples/templates	Template definitions and instances with parameter overrides
examples/datasets	DataSet metrics with typed columnar data
examples/application	Application (non-host subscriber) with node/device event tracking
examples/aliases	Edge Node with metric aliases and alias resolution
examples/commands	Host Application sending NCMD and DCMD commands to edge nodes
examples/server-walking	Edge Node with multi-broker ServerURLs and primary host awareness

Run any example with:

MQTT_BROKER=tcp://localhost:1883 go run ./examples/edge-node

---

Configuration

Edge Node Config

Field	Type	Default	Description
GroupID	string	required	Sparkplug group identifier
EdgeNodeID	string	required	Edge node identifier
BrokerURL	string		MQTT broker URL (tcp://, ssl://)
ServerURLs	[]string		Multiple broker URLs for server walking
ClientID	string	auto	MQTT client ID
PrimaryHostID	string		Primary Host Application ID to wait for
UseAliases	bool	false	Use metric aliases in DATA messages
OnState	func		State change callback
OnNodeCmd	func		NCMD received callback

Host Application Config

Field	Type	Default	Description
HostID	string	required	Host Application identifier
BrokerURL	string		MQTT broker URL
ServerURLs	[]string		Multiple broker URLs
ClientID	string	auto	MQTT client ID
ReorderTimeout	time.Duration	0	Wait before requesting rebirth on sequence gap
OnNodeBirth	func		NBIRTH callback
OnDeviceBirth	func		DBIRTH callback
OnNodeData	func		NDATA callback
OnDeviceData	func		DDATA callback
OnNodeDeath	func		NDEATH callback
OnDeviceDeath	func		DDEATH callback

Application Config

Field	Type	Default	Description
BrokerURL	string		MQTT broker URL
ServerURLs	[]string		Multiple broker URLs (fallback)
ClientID	string	auto	MQTT client ID
OnState	func		State change callback
OnNodeBirth	func		NBIRTH callback
OnDeviceBirth	func		DBIRTH callback
OnNodeData	func		NDATA callback
OnDeviceData	func		DDATA callback
OnNodeDeath	func		NDEATH callback
OnDeviceDeath	func		DDEATH callback

Connection loss

Edge nodes, host applications and applications move to Disconnected (and fire OnState) as soon as the transport reports the connection gone — a broker drop, a client-ID takeover, or Disconnect called on the transport directly. Call Connect again to resume; edge nodes bump bdSeq and must re-publish NBIRTH. Custom mqtt.Transport implementations report this by calling ConnectOpts.OnConnectionLost with an error wrapping mqtt.ErrConnectionLost.

Interoperability

Every role is tested against independent implementations: the edge node against the host applications of Eclipse Tahu, pysparkplug, SparkplugNet, srad and sparkplughost; the host application and the application against the edge nodes of Eclipse Tahu, pysparkplug, SparkplugNet, srad and Eclipse Tahu sparkplug-client; on every fixture and on Mosquitto, HiveMQ CE, EMQX and NanoMQ. See INTEROPERABILITY.md for the matrix and the upstream bugs it found.

---

Specification Conformance

This library targets full conformance with the Eclipse Sparkplug B 3.0.0 specification. See docs/CONFORMANCE.md for the clause-by-clause requirements matrix.

Conformance Summary

Profile	Mandatory	Optional
Edge Node	Session, Termination, Data, Commands, Primary Host, Multi-Broker	Aliases, Templates, DataSets, Properties
Host Application	Session, Termination, Commands, Edge Termination	Message Reordering
Application	Subscribe, Track State, Alias Resolution, Seq Validation, Commands	—

---

Project Structure

go-sparkplugb
├── proto/            Generated Protobuf types (EPL-2.0)
├── topic/            Topic namespace parsing and construction
├── payload/          Typed metric codec, templates, datasets, arrays
├── edge/             Edge Node runtime
├── host/             Host Application runtime
├── app/              Application runtime (non-host consumer)
├── mqtt/             MQTT transport interface and Paho adapter
├── examples/
│   ├── edge-node/        Edge Node lifecycle
│   ├── host-app/         Host Application lifecycle
│   ├── templates/        Template definitions and instances
│   ├── aliases/          Metric aliases and alias resolution
│   ├── application/      Application (non-host consumer)
│   ├── commands/         Sending NCMD and DCMD commands
│   ├── datasets/         DataSet columnar metrics
│   └── server-walking/   Multi-broker server walking
└── docs/
    ├── CONFORMANCE.md    Specification conformance matrix
    └── LICENSING.md      License details

Building and Development

Target	Description
make test	Run all tests
make cover	Run tests with coverage report
make lint	Run golangci-lint
make build	Build examples
make check	Lint + test + coverage gate (90%)
make interop	Edge, host and application against the six reference stacks of sparkplugb-interop, on Mosquitto, HiveMQ CE, EMQX and NanoMQ (interop/ module, Docker)

git clone https://github.com/otfabric/go-sparkplugb.git
cd go-sparkplugb
make check

Related Projects

Repository	Description
sparkplugbctl	CLI for monitoring, publishing, simulating and debugging Sparkplug B infrastructure
sparkplugb-interop	Reference Sparkplug B edge nodes and host applications (Eclipse Tahu, pysparkplug, SparkplugNet, srad, sparkplug-client, sparkplughost) as fixture-driven containers, which make interop runs this library against

License

Original OT Fabric code is MIT licensed. The proto/ directory contains code derived from the Eclipse Tahu project under EPL-2.0. See docs/LICENSING.md for details.

Documentation

Index

• type ExponentialBackoff
  • func (b ExponentialBackoff) Delay(attempt int) time.Duration
  • func (b ExponentialBackoff) ShouldRetry(attempt int, err error) (bool, time.Duration)
• type FieldLogger
  • func NewSlogLogger(h slog.Handler) FieldLogger
  • func NewStdLogger(l *log.Logger, debug bool) FieldLogger
• type Logger
  • func ComponentLogger(l Logger, name string) Logger
  • func NopLogger() Logger
• type RetryPolicy
  • func NoRetry() RetryPolicy

Types

type ExponentialBackoff

type ExponentialBackoff struct {
	// Initial is the delay after the first failure; 500 ms when zero.
	Initial time.Duration
	// Max caps the delay; 30 s when zero.
	Max time.Duration
	// Attempts is the number of retries; 5 when zero, unbounded when
	// negative.
	Attempts int
	// NoJitter waits the delay itself, for tests.
	NoJitter bool
}

ExponentialBackoff is the RetryPolicy that retries after a delay that doubles from Initial up to Max, with jitter: the wait is drawn at random between half of the delay and the delay, so that devices that failed together do not come back together.

The zero value retries up to 5 times from 500 ms (Attempts 5, Initial 500 ms, Max 30 s).

func (ExponentialBackoff) Delay

func (b ExponentialBackoff) Delay(attempt int) time.Duration

Delay returns the wait after the failure of attempt, regardless of the error and of Attempts.

func (ExponentialBackoff) ShouldRetry

func (b ExponentialBackoff) ShouldRetry(attempt int, err error) (bool, time.Duration)

ShouldRetry implements RetryPolicy.

type FieldLogger

type FieldLogger interface {
	Logger
	With(kv ...any) FieldLogger
}

FieldLogger is a Logger that keeps fields (key-value pairs) with its lines: With returns one that adds kv to every line. The packages of this module name themselves (component=edge, ...) through it when a Logger is one.

func NewSlogLogger

func NewSlogLogger(h slog.Handler) FieldLogger

NewSlogLogger returns a FieldLogger that writes to the slog handler h, at the slog levels of the same names.

func NewStdLogger

func NewStdLogger(l *log.Logger, debug bool) FieldLogger

NewStdLogger returns a Logger that writes to l (the standard logger when nil), with the level in front of each line, from Info up; NewStdLogger with debug true writes Debug lines too.

type Logger

type Logger interface {
	Debugf(format string, args ...any)
	Infof(format string, args ...any)
	Warnf(format string, args ...any)
	Errorf(format string, args ...any)
}

Logger receives the log of the edge nodes, host applications and applications of this module: set it as the Logger of their configuration. A nil Logger logs nothing. Debug is the detail of every publish and subscribe; Info, what changes (a state transition, a rebirth); Warn, what failed and is retried; Error, what fails a call or stops work.

A Logger is called from many goroutines, and must not block.

func ComponentLogger

func ComponentLogger(l Logger, name string) Logger

ComponentLogger returns the Logger a package of this module uses for l: NopLogger for nil, and for a FieldLogger one with component=name.

func NopLogger

func NopLogger() Logger

NopLogger returns a Logger that logs nothing.

type RetryPolicy

type RetryPolicy interface {
	ShouldRetry(attempt int, err error) (retry bool, wait time.Duration)
}

RetryPolicy decides whether a failed connection or publish is made again, and when: the edge nodes, host applications and applications of this module take one for their reconnection. attempt is the number of the attempt that failed, from 0. A policy is called from many goroutines.

func NoRetry

func NoRetry() RetryPolicy

NoRetry returns a RetryPolicy that never retries.