Merge pull request #105 from virajdere/feature/durable-mqtt-relay

PatrickRitchie · web-flow · commit cef10c6e4a41 · 2025-09-09T22:07:03.000-04:00
Add Durable Relay Feature to MQTT Relay Module
diff --git a/agent/MTConnect.NET-Agent/agent.config.yaml b/agent/MTConnect.NET-Agent/agent.config.yaml
@@ -94,6 +94,7 @@ modules:
 # - mqtt-relay: # - Add MQTT Relay module (Entity Structure)
 #     server: localhost  
 #     port: 1883
+#     durableRelay: true
 #     currentInterval: 10000
 #     sampleInterval: 500
 #     documentFormat: JSON
@@ -103,6 +104,7 @@ modules:
 # - mqtt-relay: # - Add MQTT Relay module (TLS)
 #     server: localhost
 #     port: 8883
+#     durableRelay: true
 #     currentInterval: 10000
 #     sampleInterval: 500
 #     tls:
diff --git a/agent/MTConnect.NET-Applications-Agents/agent.config.default.yaml b/agent/MTConnect.NET-Applications-Agents/agent.config.default.yaml
@@ -24,6 +24,7 @@ modules:
 # - mqtt-relay: # - Add MQTT Relay module
 #     server: localhost
 #     port: 1883
+#     durableRelay: true
 #     currentInterval: 5000
 #     sampleInterval: 500
 
diff --git a/agent/Modules/MTConnect.NET-AgentModule-MqttRelay/Module.cs b/agent/Modules/MTConnect.NET-AgentModule-MqttRelay/Module.cs
@@ -35,6 +35,9 @@ public class Module : MTConnectAgentModule
         private readonly MqttFactory _mqttFactory;
         private readonly IMqttClient _mqttClient;
         private CancellationTokenSource _stop;
+        private static readonly object _lastSentSequenceLock = new object();
+        private long _totalIncomingObservations = 0;
+        private long _lastSentSequence = 0;
 
 
         public Module(IMTConnectAgentBroker mtconnectAgent, object configuration) : base(mtconnectAgent)
@@ -204,6 +207,10 @@ private async Task Worker()
                         // Initialize Entity Server (if configured)
                         if (_entityServer != null)
                         {
+                            if (_configuration.DurableRelay)
+                            {
+                                await RelayBufferedObservations();
+                            }
                             await PublishDevices();
                             await PublishCurrentObservations();
                             await PublishAssets();
@@ -243,7 +250,121 @@ private async Task Worker()
             } while (!_stop.Token.IsCancellationRequested);
         }
 
+        private static IEnumerable<IObservationOutput> GetAllObservations(IStreamsResponseOutputDocument doc)
+        {
+            if (doc == null || doc.Streams == null) yield break;
+
+            foreach (var deviceStream in doc.Streams)
+            {
+                if (deviceStream.ComponentStreams == null) continue;
+                foreach (var componentStream in deviceStream.ComponentStreams)
+                {
+                    if (componentStream.Observations == null) continue;
+                    foreach (var obs in componentStream.Observations)
+                    {
+                        yield return obs;
+                    }
+                }
+            }
+        }
+
+        private async Task RelayBufferedObservations()
+        {
+            if (Agent is IMTConnectAgentBroker broker)
+            {
+                ulong lastSent = GetLastSentSequence();
+                ulong from = Math.Max(lastSent + 1, broker.FirstSequence);
+                ulong to = broker.LastSequence;
+
+                Log(MTConnectLogLevel.Information, $"[MQTT Relay] RelayBufferedObservations: lastSent={lastSent}, broker.FirstSequence={broker.FirstSequence}, broker.LastSequence={broker.LastSequence}");
+
+                long missed = (long)(to - lastSent);
+
+                if (lastSent + 1 < broker.FirstSequence)
+                {
+                    Log(MTConnectLogLevel.Warning, $"[MQTT Relay] Some observations could not be relayed because they were overwritten in the buffer (lastSent={lastSent}, firstAvailable={broker.FirstSequence}).");
+                }
+                if (missed > 0 && from <= to)
+                {
+                    Log(MTConnectLogLevel.Warning, $"[MQTT Relay] Network reconnected. {to - from + 1} observations will now be sent (from Sequence {from} to {to}).");
+                }
+
+                if (from <= to)
+                {
+                    var doc = broker.GetDeviceStreamsResponseDocument(from, to);
+                    var observations = GetAllObservations(doc)
+                        .OrderBy(o => o.Sequence)
+                        .ToList();
+
+                    Log(MTConnectLogLevel.Information, $"[MQTT Relay] Relaying buffered observations: {observations.Count} from Sequence {from} to {to}");
+
+                    foreach (var obs in observations)
+                    {
+                        var x = new Observation();
+                        x.DeviceUuid = obs.DeviceUuid;
+                        x.DataItemId = obs.DataItemId;
+                        x.DataItem = obs.DataItem;
+                        x.Name = obs.Name;
+                        x.Category = obs.Category;
+                        x.Type = obs.Type;
+                        x.SubType = obs.SubType;
+                        x.Representation = obs.Representation;
+                        x.CompositionId = obs.CompositionId;
+                        x.InstanceId = obs.InstanceId;
+                        x.Sequence = obs.Sequence;
+                        x.Timestamp = obs.Timestamp;
+                        x.AddValues(obs.Values);
+
+                        var result = await _entityServer.PublishObservation(_mqttClient, x);
+                        if (result != null && result.IsSuccess)
+                        {
+                            SetLastSentSequence(x.Sequence);
+                        }
+                    }
 
+                    Log(MTConnectLogLevel.Information, $"[MQTT Relay] Buffered observation relay complete. {observations.Count} missed observations sent.");
+                    var unsent = broker.LastSequence > (ulong)_lastSentSequence ? broker.LastSequence - (ulong)_lastSentSequence : 0;
+                    if (unsent > 0)
+                        Log(MTConnectLogLevel.Information, $"[MQTT Relay] {unsent} new observations arrived during relay and will be sent next.");
+                }
+                else
+                {
+                    Log(MTConnectLogLevel.Information, "[MQTT Relay] No buffered observations to relay.");
+                }
+            }
+        }
+
+        private static string GetLastSentSequenceFilePath()
+        {
+            var bufferDir = Path.Combine(AppContext.BaseDirectory, "buffer");
+            Directory.CreateDirectory(bufferDir);
+            return Path.Combine(bufferDir, "mqttrelay_last_sent.seq");
+        }
+
+        private ulong GetLastSentSequence()
+        {
+            if (!_configuration.DurableRelay) return 0; // Default
+            var path = GetLastSentSequenceFilePath();
+            lock (_lastSentSequenceLock)
+            {
+                if (File.Exists(path))
+                {
+                    var text = File.ReadAllText(path);
+                    if (ulong.TryParse(text, out var seq)) return seq;
+                }
+            }
+            return 0;
+        }
+
+        private void SetLastSentSequence(ulong seq)
+        {
+            if (!_configuration.DurableRelay) return; // Default
+            var path = GetLastSentSequenceFilePath();
+            lock (_lastSentSequenceLock)
+            {
+                File.WriteAllText(path, seq.ToString());
+            }
+        }
 
         private async void ProbeReceived(IDevice device, IDevicesResponseDocument responseDocument)
         {
@@ -553,16 +674,47 @@ private async void AgentDeviceAdded(object sender, IDevice device)
 
         private async void AgentObservationAdded(object sender, IObservation observation)
         {
-            if (_entityServer != null)
+            if (_configuration.DurableRelay)
             {
-                if (observation.Category == DataItemCategory.CONDITION)
+                Interlocked.Increment(ref _totalIncomingObservations);
+
+                var lastSent = GetLastSentSequence();
+                _lastSentSequence = (long)lastSent;
+
+                if (_entityServer != null)
                 {
-                    var conditionObservations = Agent.GetCurrentObservations(observation.DeviceUuid, observation.DataItemId);
-                    await PublishObservations(conditionObservations);
+                    if (observation.Category == DataItemCategory.CONDITION)
+                    {
+                        var conditionObservations = Agent.GetCurrentObservations(observation.DeviceUuid, observation.DataItemId);
+                        var result = await _entityServer.PublishObservations(_mqttClient, conditionObservations.OfType<IObservation>());
+                        if (result != null && result.IsSuccess && conditionObservations != null && conditionObservations.Any())
+                        {
+                            SetLastSentSequence(conditionObservations.Max(o => o.Sequence));
+                        }
+                    }
+                    else
+                    {
+                        var result = await _entityServer.PublishObservation(_mqttClient, observation);
+                        if (result != null && result.IsSuccess)
+                        {
+                            SetLastSentSequence(observation.Sequence);
+                        }
+                    }
                 }
-                else
+            }
+            else
+            {
+                if (_entityServer != null)
                 {
-                    await _entityServer.PublishObservation(_mqttClient, observation);
+                    if (observation.Category == DataItemCategory.CONDITION)
+                    {
+                        var conditionObservations = Agent.GetCurrentObservations(observation.DeviceUuid, observation.DataItemId);
+                        await PublishObservations(conditionObservations);
+                    }
+                    else
+                    {
+                        await _entityServer.PublishObservation(_mqttClient, observation);
+                    }
                 }
             }
         }
diff --git a/agent/Modules/MTConnect.NET-AgentModule-MqttRelay/MqttRelayModuleConfiguration.cs b/agent/Modules/MTConnect.NET-AgentModule-MqttRelay/MqttRelayModuleConfiguration.cs
@@ -95,6 +95,11 @@ public class MqttRelayModuleConfiguration : IMTConnectMqttDocumentServerConfigur
         /// </summary>
         public int SampleInterval { get; set; }
 
+        /// <summary>
+        /// Sets whether to send buffered observation on successful reconnect
+        /// </summary>
+        public bool DurableRelay { get; set; } = false;
+
 
         public MqttRelayModuleConfiguration()
         {
diff --git a/agent/Modules/MTConnect.NET-AgentModule-MqttRelay/README.md b/agent/Modules/MTConnect.NET-AgentModule-MqttRelay/README.md
@@ -59,6 +59,8 @@ modules:
 
 * `documentFormat` - The Document Format ID to use to format the payload
 
+* `durableRelay` - Enables durable relay mode. When set to `true`, the agent will persist unsent observations to disk and automatically relay missed observations to the MQTT broker after a network reconnect or agent restart.
+
 * `tls` - Sets the TLS settings
 
     * `pfx` - The PFX certificate settings
diff --git a/libraries/MTConnect.NET-MQTT/MTConnectMqttEntityServer.cs b/libraries/MTConnect.NET-MQTT/MTConnectMqttEntityServer.cs
@@ -111,7 +111,7 @@ private MqttApplicationMessage CreateMessage(IDevice device)
         }
 
 
-        public async Task PublishObservation(IMqttClient mqttClient, IObservation observation)
+        public async Task<MqttClientPublishResult> PublishObservation(IMqttClient mqttClient, IObservation observation)
         {
             if (mqttClient != null && mqttClient.IsConnected && observation != null)
             {
@@ -127,15 +127,17 @@ public async Task PublishObservation(IMqttClient mqttClient, IObservation observ
                     {
                         if (SendError != null) SendError.Invoke(this, message.Topic);
                     }
+                    return result;
                 }
                 catch (Exception ex)
                 {
                     if (ClientError != null) ClientError.Invoke(this, ex);
                 }
             }
+            return null;
         }
 
-        public async Task PublishObservations(IMqttClient mqttClient, IEnumerable<IObservation> observations)
+        public async Task<MqttClientPublishResult> PublishObservations(IMqttClient mqttClient, IEnumerable<IObservation> observations)
         {
             if (mqttClient != null && mqttClient.IsConnected && !observations.IsNullOrEmpty())
             {
@@ -151,12 +153,14 @@ public async Task PublishObservations(IMqttClient mqttClient, IEnumerable<IObser
                     {
                         if (SendError != null) SendError.Invoke(this, message.Topic);
                     }
+                    return result;
                 }
                 catch (Exception ex)
                 {
                     if (ClientError != null) ClientError.Invoke(this, ex);
                 }
             }
+            return null;
         }
 
         public async Task PublishObservation(MqttServer mqttServer, IObservation observation)
diff --git a/tests/IntegrationTests/ClientAgentCommunicationTests.cs b/tests/IntegrationTests/ClientAgentCommunicationTests.cs
@@ -162,7 +162,7 @@ private void AddCuttingTools()
             var measurements = new List<IMeasurement>();
             measurements.Add(new MTConnect.Assets.CuttingTools.Measurements.FunctionalLengthMeasurement(7.6543));
             measurements.Add(new MTConnect.Assets.CuttingTools.Measurements.CuttingDiameterMaxMeasurement(0.375));
-            cuttingToolLifeCycle.Measurements = measurements;
+            cuttingToolLifeCycle.Measurements = measurements.OfType<IToolingMeasurement>();
 
             var cuttingItems = new List<ICuttingItem>();
             cuttingItems.Add(new CuttingItem
@@ -239,7 +239,7 @@ private void AddCuttingTools()
             return tcs.Task;
         }
 
-        private static void GenerateDevicesXml(
+        internal static void GenerateDevicesXml(
             string machineId,
             string machineName,
             string fileName,

Original file line number	Diff line number	Diff line change
`@@ -111,7 +111,7 @@ private MqttApplicationMessage CreateMessage(IDevice device)`
`111`	`111`	`}`
`112`	`112`
`113`	`113`
`114`		`- public async Task PublishObservation(IMqttClient mqttClient, IObservation observation)`
	`114`	`+ public async Task<MqttClientPublishResult> PublishObservation(IMqttClient mqttClient, IObservation observation)`
`115`	`115`	`{`
`116`	`116`	`if (mqttClient != null && mqttClient.IsConnected && observation != null)`
`117`	`117`	`{`
`@@ -127,15 +127,17 @@ public async Task PublishObservation(IMqttClient mqttClient, IObservation observ`
`127`	`127`	`{`
`128`	`128`	`if (SendError != null) SendError.Invoke(this, message.Topic);`
`129`	`129`	`}`
	`130`	`+ return result;`
`130`	`131`	`}`
`131`	`132`	`catch (Exception ex)`
`132`	`133`	`{`
`133`	`134`	`if (ClientError != null) ClientError.Invoke(this, ex);`
`134`	`135`	`}`
`135`	`136`	`}`
	`137`	`+ return null;`
`136`	`138`	`}`
`137`	`139`
`138`		`- public async Task PublishObservations(IMqttClient mqttClient, IEnumerable<IObservation> observations)`
	`140`	`+ public async Task<MqttClientPublishResult> PublishObservations(IMqttClient mqttClient, IEnumerable<IObservation> observations)`
`139`	`141`	`{`
`140`	`142`	`if (mqttClient != null && mqttClient.IsConnected && !observations.IsNullOrEmpty())`
`141`	`143`	`{`
`@@ -151,12 +153,14 @@ public async Task PublishObservations(IMqttClient mqttClient, IEnumerable<IObser`
`151`	`153`	`{`
`152`	`154`	`if (SendError != null) SendError.Invoke(this, message.Topic);`
`153`	`155`	`}`
	`156`	`+ return result;`
`154`	`157`	`}`
`155`	`158`	`catch (Exception ex)`
`156`	`159`	`{`
`157`	`160`	`if (ClientError != null) ClientError.Invoke(this, ex);`
`158`	`161`	`}`
`159`	`162`	`}`
	`163`	`+ return null;`
`160`	`164`	`}`
`161`	`165`
`162`	`166`	`public async Task PublishObservation(MqttServer mqttServer, IObservation observation)`