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

refactor(dcl): OpcUaDataConnection uses OpcUaEndpointConfig via FromFlatDict

Browse Source
This commit is contained in:
Joseph Doherty
2026-05-12 00:57:09 -04:00
parent 80d4d3e252
commit f98d29fc36
1 changed files with 29 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,8 @@
using Microsoft.Extensions.Logging;
using ScadaLink.Commons.Interfaces.Protocol;
using ScadaLink.Commons.Serialization;
using ScadaLink.Commons.Types;
using ScadaLink.Commons.Types.DataConnections;
using ScadaLink.Commons.Types.Enums;
namespace ScadaLink.DataConnectionLayer.Adapters;
@@ -43,23 +45,23 @@ public class OpcUaDataConnection : IDataConnection
public async Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
{
_endpointUrl = connectionDetails.TryGetValue("endpoint", out var url)
? url
: connectionDetails.TryGetValue("EndpointUrl", out var url2)
? url2
: "opc.tcp://localhost:4840";
var config = OpcUaEndpointConfigSerializer.FromFlatDict(connectionDetails);
_endpointUrl = string.IsNullOrWhiteSpace(config.EndpointUrl)
? "opc.tcp://localhost:4840"
: config.EndpointUrl;
var options = new OpcUaConnectionOptions(
SessionTimeoutMs: ParseInt(connectionDetails, "SessionTimeoutMs", 60000),
OperationTimeoutMs: ParseInt(connectionDetails, "OperationTimeoutMs", 15000),
PublishingIntervalMs: ParseInt(connectionDetails, "PublishingIntervalMs", 1000),
KeepAliveCount: ParseInt(connectionDetails, "KeepAliveCount", 10),
LifetimeCount: ParseInt(connectionDetails, "LifetimeCount", 30),
MaxNotificationsPerPublish: ParseInt(connectionDetails, "MaxNotificationsPerPublish", 100),
SamplingIntervalMs: ParseInt(connectionDetails, "SamplingIntervalMs", 1000),
QueueSize: ParseInt(connectionDetails, "QueueSize", 10),
SecurityMode: connectionDetails.TryGetValue("SecurityMode", out var secMode) ? secMode : "None",
AutoAcceptUntrustedCerts: ParseBool(connectionDetails, "AutoAcceptUntrustedCerts", true));
SessionTimeoutMs: config.SessionTimeoutMs,
OperationTimeoutMs: config.OperationTimeoutMs,
PublishingIntervalMs: config.PublishingIntervalMs,
KeepAliveCount: config.KeepAliveCount,
LifetimeCount: config.LifetimeCount,
MaxNotificationsPerPublish: config.MaxNotificationsPerPublish,
SamplingIntervalMs: config.SamplingIntervalMs,
QueueSize: config.QueueSize,
SecurityMode: config.SecurityMode.ToString(),
AutoAcceptUntrustedCerts: config.AutoAcceptUntrustedCerts);
_status = ConnectionHealth.Connecting;
@@ -71,49 +73,40 @@ public class OpcUaDataConnection : IDataConnection
_disconnectFired = false;
_logger.LogInformation("OPC UA connected to {Endpoint}", _endpointUrl);
// Heartbeat stale tag monitoring (optional)
await StartHeartbeatMonitorAsync(connectionDetails, cancellationToken);
await StartHeartbeatMonitorAsync(config.Heartbeat, cancellationToken);
}
private async Task StartHeartbeatMonitorAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken)
private async Task StartHeartbeatMonitorAsync(OpcUaHeartbeatConfig? heartbeat, CancellationToken cancellationToken)
{
if (!connectionDetails.TryGetValue("HeartbeatTagPath", out var heartbeatTag) || string.IsNullOrWhiteSpace(heartbeatTag))
if (heartbeat is null || string.IsNullOrWhiteSpace(heartbeat.TagPath))
return;
var maxSilenceSeconds = ParseInt(connectionDetails, "HeartbeatMaxSilence", 30);
_staleMonitor?.Dispose();
_staleMonitor = new StaleTagMonitor(TimeSpan.FromSeconds(maxSilenceSeconds));
_staleMonitor = new StaleTagMonitor(TimeSpan.FromSeconds(heartbeat.MaxSilenceSeconds));
_staleMonitor.Stale += () =>
{
_logger.LogWarning("OPC UA heartbeat tag '{Tag}' stale — no update in {Seconds}s", heartbeatTag, maxSilenceSeconds);
_logger.LogWarning("OPC UA heartbeat tag '{Tag}' stale — no update in {Seconds}s",
heartbeat.TagPath, heartbeat.MaxSilenceSeconds);
RaiseDisconnected();
};
try
{
_heartbeatSubscriptionId = await SubscribeAsync(heartbeatTag, (_, _) => _staleMonitor.OnValueReceived(), cancellationToken);
_heartbeatSubscriptionId = await SubscribeAsync(heartbeat.TagPath,
(_, _) => _staleMonitor.OnValueReceived(), cancellationToken);
_staleMonitor.Start();
_logger.LogInformation("OPC UA heartbeat monitor started for '{Tag}' with {Seconds}s max silence", heartbeatTag, maxSilenceSeconds);
_logger.LogInformation("OPC UA heartbeat monitor started for '{Tag}' with {Seconds}s max silence",
heartbeat.TagPath, heartbeat.MaxSilenceSeconds);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to subscribe to heartbeat tag '{Tag}' — stale monitor not active", heartbeatTag);
_logger.LogWarning(ex, "Failed to subscribe to heartbeat tag '{Tag}' — stale monitor not active",
heartbeat.TagPath);
_staleMonitor.Dispose();
_staleMonitor = null;
}
}
internal static int ParseInt(IDictionary<string, string> d, string key, int defaultValue)
{
return d.TryGetValue(key, out var str) && int.TryParse(str, out var val) ? val : defaultValue;
}
internal static bool ParseBool(IDictionary<string, string> d, string key, bool defaultValue)
{
return d.TryGetValue(key, out var str) && bool.TryParse(str, out var val) ? val : defaultValue;
}
private void OnClientConnectionLost()
{
RaiseDisconnected();