sparkplughost

README

Sparkplug B Host Application Go library

Go based package to use for implementing Sparkplug B Host Applications

Sparkplug Host Library for Go

The Sparkplug Host Library for Go is designed to assist developers in creating custom Host Applications adhering to the Sparkplug standard. It simplifies the implementation of MQTT-based communication for industrial IoT environments, allowing seamless integration with Sparkplug-compliant edge nodes and devices.

Features

• Sparkplug B Standard Support: Implement Sparkplug B standard communication effortlessly.
• Customizable Callbacks: Receive real-time updates with customizable callback functions for metrics, births, commands, and other events.
• Command Writing Support: Use the library to write commands to both edge nodes and devices.
• Flexible Design: Create dynamic and tailored Host Applications suited to your specific use case.
• Fully compliant: The library fully passes the Eclipse Sparkplug TCK. Latest execution logs are available here.

Installation

go get -u github.com/EvergenEnergy/sparkplughost

This package requires Go version 1.21 or higher.

Usage

// 1. Configure your MQTT Brokers
brokerConfig := sparkplughost.MqttBrokerConfig{BrokerURL: "tcp://broker.hivemq.com:1883"}

// 2. Create a handler function that will be called whenever a metric is added or updated
metricHandler := func(metric sparkplughost.HostMetric) {
    log.Printf("Received metric callback:%v\n", metric)
}

// 3. Create the `HostApplication`. See options.go for further configuration options.
host, err := sparkplughost.NewHostApplication(
    []sparkplughost.MqttBrokerConfig{brokerConfig},
    "my-host-id",
    sparkplughost.WithMetricHandler(metricHandler),
)
if err != nil {
    panic(err)
}

// 4. Start the application. The `Run` function will block until
// `ctx` is cancelled.
go func() {
    if err := host.Run(ctx); err != nil {
        panic(err)
    }
}()

// 5. Optionally, use the `HostApplication` methods to send commands
// to Edge Nodes and Devices
// host.SendEdgeNodeCommand(sparkplughost.EdgeNodeDescriptor{GroupID:"group-id", EdgeNodeID:"edge-node-id"}, metrics)
// host.SendDeviceCommand(sparkplughost.EdgeNodeDescriptor{GroupID:"group-id", EdgeNodeID:"edge-node-id"},deviceID, metrics)

Samples are available in the examples directory for reference.

Important Notice

This library is under heavy development, and the API may change at any point.

We are actively working towards stabilizing the API, and once it reaches a stable state, a v1.0.0 version will be released. From that point forward, the library will adhere to the normal guarantees of semantic versioning.

Contributing

We welcome contributions from the community! Please check our Contribution Guidelines for details on how to contribute to this project.

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

Documentation

Index

• type CachingHandler
  • func NewCachingHandler() *CachingHandler
  • func (c *CachingHandler) AllMetrics() []HostMetric
  • func (c *CachingHandler) DeviceMetrics(descriptor EdgeNodeDescriptor, deviceID string) []HostMetric
  • func (c *CachingHandler) EdgeNodeMetrics(descriptor EdgeNodeDescriptor) []HostMetric
  • func (c *CachingHandler) GroupMetrics(groupID string) []HostMetric
  • func (c *CachingHandler) HandleMetric(metric HostMetric)
• type EdgeNodeDescriptor
  • func (e EdgeNodeDescriptor) String() string
• type HostApplication
  • func NewHostApplication(brokerConfigs []MqttBrokerConfig, hostID string, opts ...Option) (*HostApplication, error)
  • func (h *HostApplication) Run(ctx context.Context) error
  • func (h *HostApplication) SendDeviceCommand(descriptor EdgeNodeDescriptor, deviceID string, ...) error
  • func (h *HostApplication) SendDeviceRebirthRequest(descriptor EdgeNodeDescriptor, deviceID string) error
  • func (h *HostApplication) SendEdgeNodeCommand(descriptor EdgeNodeDescriptor, metrics []*protobuf.Payload_Metric) error
  • func (h *HostApplication) SendEdgeNodeRebirthRequest(descriptor EdgeNodeDescriptor) error
• type HostMetric
• type MetricHandler
• type MetricQuality
• type MqttBrokerConfig
• type Option
  • func WithLogger(logger *slog.Logger) Option
  • func WithMetricHandler(metricHandler MetricHandler) Option
  • func WithReorderTimeout(timeout time.Duration) Option

Types

type CachingHandler

type CachingHandler struct {
	// contains filtered or unexported fields
}

CachingHandler is a MetricHandler that stores the latest known state of every metric in memory. It allows clients to get a snapshot view of all the current Edge Nodes and Devices on-demand. All operations in this handler are safe for concurrent use.

func NewCachingHandler

func NewCachingHandler() *CachingHandler

NewCachingHandler returns a new handler ready to be used.

func (*CachingHandler) AllMetrics

func (c *CachingHandler) AllMetrics() []HostMetric

AllMetrics returns all currently known metrics in the MQTT infrastructure.

func (*CachingHandler) DeviceMetrics

func (c *CachingHandler) DeviceMetrics(descriptor EdgeNodeDescriptor, deviceID string) []HostMetric

DeviceMetrics returns all known metrics for a specific Device.

func (*CachingHandler) EdgeNodeMetrics

func (c *CachingHandler) EdgeNodeMetrics(descriptor EdgeNodeDescriptor) []HostMetric

EdgeNodeMetrics returns all known metrics for a specific Edge Node. This includes the metrics for Devices associated with that Edge Node.

func (*CachingHandler) GroupMetrics

func (c *CachingHandler) GroupMetrics(groupID string) []HostMetric

GroupMetrics returns all known metrics for a specific Group. This includes all Edge Nodes and Devices in that group.

func (*CachingHandler) HandleMetric

func (c *CachingHandler) HandleMetric(metric HostMetric)

HandleMetric is a MetricHandler function that can be passed to WithMetricHandler when creating the HostApplication.

type EdgeNodeDescriptor

type EdgeNodeDescriptor struct {
	GroupID    string
	EdgeNodeID string
}

EdgeNodeDescriptor is the combination of the Group ID and Edge Node ID. No two Edge Nodes within a Sparkplug environment can have the same Group ID and same Edge Node ID.

func (EdgeNodeDescriptor) String

func (e EdgeNodeDescriptor) String() string

type HostApplication

type HostApplication struct {
	// contains filtered or unexported fields
}

func NewHostApplication

func NewHostApplication(brokerConfigs []MqttBrokerConfig, hostID string, opts ...Option) (*HostApplication, error)

func (*HostApplication) Run

func (h *HostApplication) Run(ctx context.Context) error

Run will connect to the mqtt broker and block until ctx is canceled.

func (*HostApplication) SendDeviceCommand

func (h *HostApplication) SendDeviceCommand(descriptor EdgeNodeDescriptor, deviceID string, metrics []*protobuf.Payload_Metric) error

func (*HostApplication) SendDeviceRebirthRequest

func (h *HostApplication) SendDeviceRebirthRequest(descriptor EdgeNodeDescriptor, deviceID string) error

func (*HostApplication) SendEdgeNodeCommand

func (h *HostApplication) SendEdgeNodeCommand(descriptor EdgeNodeDescriptor, metrics []*protobuf.Payload_Metric) error

func (*HostApplication) SendEdgeNodeRebirthRequest

func (h *HostApplication) SendEdgeNodeRebirthRequest(descriptor EdgeNodeDescriptor) error

type HostMetric

type HostMetric struct {
	EdgeNodeDescriptor EdgeNodeDescriptor
	DeviceID           string
	Metric             *protobuf.Payload_Metric
	Quality            MetricQuality
}

HostMetric represents the view this Host Application has of a particular edge node or device metric.

type MetricHandler

type MetricHandler func(HostMetric)

MetricHandler is a callback type which can be set to be executed upon the change of any of the known Edge Node or Device metrics. This includes when a metric is first received during BIRTH messages as well as updates through DATA or DEATH messages.

type MetricQuality

type MetricQuality string

MetricQuality will be "STALE" when a given edge node or device looses connection to the MQTT broker. This represents that the data was accurate at a time, but now that the MQTT session has been lost can no longer be considered current or up to date.

const (
	MetricQualityGood  MetricQuality = "GOOD"
	MetricQualityStale MetricQuality = "STALE"
)

type MqttBrokerConfig

type MqttBrokerConfig struct {
	// URL of the broker. The format should be scheme://host:port
	// (e.g., tcp://localhost:1883). Required.
	BrokerURL string
	// Username if required by the broker. Optional.
	Username string
	// Password if required by the broker. Optional.
	Password string
	// SSL/TLS configuration to be used when connecting to an MQTT broker.
	// This can be used for brokers where the authentication needs to happen
	// via client certificates instead of username + password. Optional.
	TLSConfig *tls.Config
}

MqttBrokerConfig contains the configuration parameters for each of the MQTT Brokers to be used.

type Option

type Option func(*config)

Option allows clients to configure the Host Application.

func WithLogger

func WithLogger(logger *slog.Logger) Option

WithLogger sets a `*slog.Logger` instance to use by the Host application. This allows clients to enable/disable DEBUG and INFO messages. The default logger sends everything to `io.Discard`.

func WithMetricHandler

func WithMetricHandler(metricHandler MetricHandler) Option

WithMetricHandler sets a MetricHandler to be called when metrics are created or updated by this Host Application.

func WithReorderTimeout

func WithReorderTimeout(timeout time.Duration) Option

WithReorderTimeout sets a timeout on how long to wait before requesting a Rebirth when receiving messages out of order. Default: 5 seconds.