dohertj2/scadalink-design
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(dcl): add StaleTagMonitor for heartbeat-based disconnect detection

Browse Source 
Composable StaleTagMonitor class in Commons fires a Stale event when no
value is received within a configurable max silence period. Integrated
into both LmxProxyDataConnection and OpcUaDataConnection adapters via
optional HeartbeatTagPath/HeartbeatMaxSilence connection config keys.
When stale, the adapter fires Disconnected triggering the standard
reconnect cycle. 10 unit tests cover timer behavior.
This commit is contained in:
Joseph Doherty
2026-03-24 14:28:11 -04:00
parent 02a7e8abc6
commit d4397910f0
4 changed files with 285 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
src/ScadaLink.DataConnectionLayer/Adapters/LmxProxyDataConnection.cs
View File

@@ -30,6 +30,8 @@ public class LmxProxyDataConnection : IDataConnection
private readonly Dictionary<string, ILmxSubscription> _subscriptions = new();
private volatile bool _disconnectFired;
private StaleTagMonitor? _staleMonitor;
private string? _heartbeatSubscriptionId;
public LmxProxyDataConnection(ILmxProxyClientFactory clientFactory, ILogger<LmxProxyDataConnection> logger)
{
@@ -57,10 +59,44 @@ public class LmxProxyDataConnection : IDataConnection
_disconnectFired = false;
_logger.LogInformation("LmxProxy connected to {Host}:{Port}", _host, _port);
// Heartbeat stale tag monitoring (optional)
await StartHeartbeatMonitorAsync(connectionDetails, cancellationToken);
}
private async Task StartHeartbeatMonitorAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken)
{
if (!connectionDetails.TryGetValue("HeartbeatTagPath", out var heartbeatTag) || string.IsNullOrWhiteSpace(heartbeatTag))
return;
var maxSilenceSeconds = connectionDetails.TryGetValue("HeartbeatMaxSilence", out var silenceStr)
&& int.TryParse(silenceStr, out var sec) ? sec : 30;
_staleMonitor?.Dispose();
_staleMonitor = new StaleTagMonitor(TimeSpan.FromSeconds(maxSilenceSeconds));
_staleMonitor.Stale += () =>
{
_logger.LogWarning("LmxProxy heartbeat tag '{Tag}' stale — no update in {Seconds}s", heartbeatTag, maxSilenceSeconds);
RaiseDisconnected();
};
try
{
_heartbeatSubscriptionId = await SubscribeAsync(heartbeatTag, (_, _) => _staleMonitor.OnValueReceived(), cancellationToken);
_staleMonitor.Start();
_logger.LogInformation("LmxProxy heartbeat monitor started for '{Tag}' with {Seconds}s max silence", heartbeatTag, maxSilenceSeconds);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to subscribe to heartbeat tag '{Tag}' — stale monitor not active", heartbeatTag);
_staleMonitor.Dispose();
_staleMonitor = null;
}
}
public async Task DisconnectAsync(CancellationToken cancellationToken = default)
{
StopHeartbeatMonitor();
if (_client != null)
{
await _client.DisconnectAsync();
@@ -200,8 +236,16 @@ public class LmxProxyDataConnection : IDataConnection
}
}
private void StopHeartbeatMonitor()
{
_staleMonitor?.Dispose();
_staleMonitor = null;
_heartbeatSubscriptionId = null;
}
public async ValueTask DisposeAsync()
{
StopHeartbeatMonitor();
foreach (var subscription in _subscriptions.Values)
{
try { await subscription.DisposeAsync(); }

44
src/ScadaLink.DataConnectionLayer/Adapters/OpcUaDataConnection.cs
View File

@@ -1,5 +1,6 @@
using Microsoft.Extensions.Logging;
using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Types;
using ScadaLink.Commons.Types.Enums;
namespace ScadaLink.DataConnectionLayer.Adapters;
@@ -26,6 +27,8 @@ public class OpcUaDataConnection : IDataConnection
/// Maps subscription IDs to their tag paths for cleanup.
/// </summary>
private readonly Dictionary<string, string> _subscriptionHandles = new();
private StaleTagMonitor? _staleMonitor;
private string? _heartbeatSubscriptionId;
public OpcUaDataConnection(IOpcUaClientFactory clientFactory, ILogger<OpcUaDataConnection> logger)
{
@@ -67,6 +70,38 @@ public class OpcUaDataConnection : IDataConnection
_status = ConnectionHealth.Connected;
_disconnectFired = false;
_logger.LogInformation("OPC UA connected to {Endpoint}", _endpointUrl);
// Heartbeat stale tag monitoring (optional)
await StartHeartbeatMonitorAsync(connectionDetails, cancellationToken);
}
private async Task StartHeartbeatMonitorAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken)
{
if (!connectionDetails.TryGetValue("HeartbeatTagPath", out var heartbeatTag) || string.IsNullOrWhiteSpace(heartbeatTag))
return;
var maxSilenceSeconds = ParseInt(connectionDetails, "HeartbeatMaxSilence", 30);
_staleMonitor?.Dispose();
_staleMonitor = new StaleTagMonitor(TimeSpan.FromSeconds(maxSilenceSeconds));
_staleMonitor.Stale += () =>
{
_logger.LogWarning("OPC UA heartbeat tag '{Tag}' stale — no update in {Seconds}s", heartbeatTag, maxSilenceSeconds);
RaiseDisconnected();
};
try
{
_heartbeatSubscriptionId = await SubscribeAsync(heartbeatTag, (_, _) => _staleMonitor.OnValueReceived(), cancellationToken);
_staleMonitor.Start();
_logger.LogInformation("OPC UA heartbeat monitor started for '{Tag}' with {Seconds}s max silence", heartbeatTag, maxSilenceSeconds);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to subscribe to heartbeat tag '{Tag}' — stale monitor not active", heartbeatTag);
_staleMonitor.Dispose();
_staleMonitor = null;
}
}
internal static int ParseInt(IDictionary<string, string> d, string key, int defaultValue)
@@ -86,6 +121,7 @@ public class OpcUaDataConnection : IDataConnection
public async Task DisconnectAsync(CancellationToken cancellationToken = default)
{
StopHeartbeatMonitor();
if (_client != null)
{
_client.ConnectionLost -= OnClientConnectionLost;
@@ -201,8 +237,16 @@ public class OpcUaDataConnection : IDataConnection
return false;
}
private void StopHeartbeatMonitor()
{
_staleMonitor?.Dispose();
_staleMonitor = null;
_heartbeatSubscriptionId = null;
}
public async ValueTask DisposeAsync()
{
StopHeartbeatMonitor();
if (_client != null)
{
_client.ConnectionLost -= OnClientConnectionLost;