Hello,
thank you for your interest in our products.
Some application do not need this (I would call it "local" store-and-forward, since no intermediate component is involved). And yes, I agree, for some applications it is absolutely necessary.
Rapid Toolkit for Sparkplug currently does not have built-in support for this. In its basic form, it can be put together relatively easily, as the "building blocks", the necessary objects and methods, are there. We can think of providing full support in next version.
One possible approach currently is to use the LockPublishing and UnlockPublishing methods. When LockPublishing is called, any metric data updates in the edge node are simply accumulated in memory, and not really sent to the broker. When UnlockPublishing is called, the accumulated metric data (if any) are commonly sent to the server. So, it is possible to call LockPublishing when the connection to the broker is lost, and call UnlockPublishing when it is regained. The code below illustrates this. If you want to use, make sure to use the very recent NuGet package (5.83.1167 or later), because some minor change in the component was needed to make it work.
// This example shows how to locally store the data updates when the edge node is offline and automatically push the stored
// data to the broker when the edge node is back online.
//
// You can use any Sparkplug application, including our SparkplugCmd utility and the SparkplugApplicationConsoleDemo
// program, to subscribe to the edge node data.
//
// Find all latest examples here:
www.doc-that.com/files/onlinedocs/OPCLabs-ConnectivityStudio/Latest/examples.html
.
// Sparkplug examples in C# on GitHub:
github.com/OPCLabs/Examples-ConnectivityStudio-CSharp
.
// Missing some example? Ask us for it on our Online Forums,
forum.opclabs.com/forum/index
! You do not have to own
// a commercial license in order to use Online Forums, and we reply to every post.
using OpcLabs.EasySparkplug;
using System;
using Timer = System.Timers.Timer;
namespace SparkplugDocExamples.EdgeNode._EasySparkplugEdgeNode
{
class CustomStoreAndForward
{
static public void Main1()
{
// Note that the default port for the "mqtt" scheme is 1883.
var hostDescriptor = new SparkplugHostDescriptor("mqtt://localhost");
// Instantiate the edge node object and hook the SystemConnectionStateChanged event.
var edgeNode = new EasySparkplugEdgeNode(hostDescriptor, "easyGroup", "easySparkplugDemo");
edgeNode.SystemConnectionStateChanged += (sender, eventArgs) =>
{
// Display the new connection state (such as when the connection to the broker succeeds or fails).
Console.WriteLine($"{nameof(EasySparkplugEdgeNode.SystemConnectionStateChanged)}: {eventArgs}");
};
// Create a read-only data metric.
var metric = SparkplugMetric.CreateIn(edgeNode, "ReadThisMetric")
.ValueType<int>()
.Writable(false);
// Count how many data updates have been buffered while the edge node is offline.
int metricBufferedCount = 0;
// Hook the ProducerOnlineChanged event to handle system connection state changes.
edgeNode.ProducerOnlineChanged += (sender, eventArgs) =>
{
var node = (EasySparkplugEdgeNode)sender;
// Display the new producer online state.
Console.WriteLine($"{nameof(EasySparkplugEdgeNode.ProducerOnlineChanged)}; {nameof(node.ProducerOnline)}={node.ProducerOnline}");
// If the edge node is back online, unlock the metric the publishing to push all the buffered data to the
// broker.
if (node.ProducerOnline)
node.UnlockPublishing();
// Otherwise, lock the publishing and start buffering the data updates.
else
{
node.LockPublishing();
metricBufferedCount = 0;
}
};
metric.Starting += (sender, args) =>
{
// Lock the publishing initially (ProducerOnline will change to true when the connection to the broker, and
// we will unlock then).
edgeNode.LockPublishing();
// Create a timer for pushing the data to the metric. In a real edge node or device, the activity may also come
// from other sources.
var timer = new Timer
{
Interval = 1000, // 1 second
AutoReset = true,
};
// Update the read data of the metric with random value whenever the timer interval elapses.
var random = new Random();
timer.Elapsed += (s, a) =>
{
// If the edge node is online, always update the metric with new data.
if (edgeNode.ProducerOnline)
metric.UpdateReadData(random.Next());
// Otherwise, if the edge node is offline, and we have not buffered too many updates, update the metric
// with new data (which will be buffered until the edge node is back online).
else if (metricBufferedCount < 100)
{
metric.UpdateReadData(random.Next());
metricBufferedCount++;
}
};
// Associate the timer with the data variable.
metric.State = timer;
timer.Start();
};
metric.Stopped += (sender, args) =>
{
// Obtain the timer associated with the metric.
var timer = (Timer)((SparkplugMetric)sender).State;
// Stop the timer.
timer.Stop();
};
// Start the edge node.
Console.WriteLine("The edge node is starting...");
edgeNode.Start();
Console.WriteLine("The edge node is started.");
Console.WriteLine();
// Let the user decide when to stop.
Console.WriteLine("Press Enter to stop the edge node...");
Console.ReadLine();
// Stop the edge node.
Console.WriteLine("The edge node is stopping...");
edgeNode.Stop();
Console.WriteLine("The edge node is stopped.");
}
}
}
It should be considered how many data gets buffered, for two reasons:
- If the MQTT disconnection is long, there will be ever increasing demand on memory to store the data. There should be some limit to that.
- When UnlockPublishing is called, all accumulated metric data changes are published in a single payload. This works fine inside our library (the max size is really high), but there is a possibility that the payload size will exceed some limits on the broker or receiver side.
This is currently handled in the example by not buffering more than 100 updates; extra ones are lost.
Let me know you thoughts on this.
Best regards
