Simplify and refactor

README.md

@@ -16,13 +16,18 @@ The module is configured using a JSON file. `connector.json` represents the conn
 ```json
 {
     "serverUrl": "opc.tcp://<HOST>:<PORT>/<SERVERNAME>",
-    "nodeIds": [
-        "ns=3;i=1002",
-        "ns=3;i=1001"
-    ],
-    "opcUaServerCertificateIssuer": "Name of certificate issuer", // used to verify untrusted server certificates
-    "opcUaServerCertificateSubject": "Issuer subject",            // used to verify untrusted server certificates
-    "opcUaServerAutoAcceptUntrustedCertificates": false
+    "publishingIntervalSeconds": 1.0,
+    "nodes": [
+        {
+            "id": "ns=3;i=1002",
+            "subscribeIntervalSeconds": 1.0,
+            "readIntervalSeconds": 1.0
+        },
+        {
+            "id": "ns=3;i=1001",
+            "readIntervalSeconds": 1.0
+        }
+    ]
 }
 ```
 

Source/BatchConfiguration.cs

@@ -7,17 +7,8 @@
 
 namespace RaaLabs.Edge.Connectors.OPCUA;
 
-/// <summary>
-/// Config class for batching settings.
-/// </summary>
 public class BatchConfiguration : IEdgeHubOutgoingEventBatchConfiguration
 {
-    /// <summary>
-    /// The batch size.
-    /// </summary>
     public int BatchSize { get; set; } = int.Parse(Environment.GetEnvironmentVariable("EDGEHUB_BATCH_SIZE") ?? "250", CultureInfo.CurrentCulture);
-    /// <summary>
-    /// The batch interval in milliseconds.
-    /// </summary>
     public int Interval { get; set; } = int.Parse(Environment.GetEnvironmentVariable("EDGEHUB_BATCH_INTERVAL") ?? "5000", CultureInfo.CurrentCulture);
 }

Source/Client.cs

@@ -0,0 +1,88 @@
+// Copyright (c) RaaLabs. All rights reserved.
+// Licensed under the GPLv2 License. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Diagnostics;
+using System.Threading;
+using System.Threading.Tasks;
+using Opc.Ua;
+using Opc.Ua.Client;
+using Serilog;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+public class Client : ICreateSessions
+{
+    private readonly ApplicationConfiguration _application;
+    private readonly ConfiguredEndpoint _endpoint;
+    private readonly UserIdentity _identity;
+    private readonly ISessionFactory _factory;
+    private readonly IMetricsHandler _metrics;
+    private readonly ILogger _logger;
+
+    public Client(ConnectorConfiguration config, ISessionFactory factory, IMetricsHandler metrics, ILogger logger)
+    {
+        _application = ConnectorDescription();
+        _application.ClientConfiguration = new()
+        {
+            DefaultSessionTimeout = (int)TimeSpan.FromHours(1).TotalMilliseconds,
+        };
+
+        (_endpoint, _identity) = ConnectAnonymouslyToAnyServerOn(config.ServerUrl);
+        _factory = factory;
+        _metrics = metrics;
+        _logger = logger;
+    }
+
+    public async Task<ISession> ConnectToServer(CancellationToken cancellationToken)
+    {
+        try
+        {
+            var sessionName = $"RaaEdge OPCUA Connector {Guid.NewGuid()}";
+            _logger.Information("Creating session '{SessionName}' with OPCUA server on '{EndpointUrl}'", sessionName, _endpoint.EndpointUrl);
+            _metrics.NumberOfSessionConnectionAttempts(1);
+            var timer = Stopwatch.StartNew();
+
+            var session = await _factory.CreateAsync(
+                _application,
+                _endpoint,
+                updateBeforeConnect: false,
+                checkDomain: false,
+                sessionName,
+                (uint)_application.ClientConfiguration.DefaultSessionTimeout,
+                _identity,
+                [],
+                cancellationToken
+            ).ConfigureAwait(false);
+
+            _metrics.NumberOfSessionConnections(1);
+            _metrics.SessionConnectionTime(timer.Elapsed.TotalSeconds);
+            return session;
+        }
+        catch (Exception error)
+        {
+            _logger.Error(error, "Failed to connect to server");
+            throw;
+        }
+    }
+
+    private static (ConfiguredEndpoint, UserIdentity) ConnectAnonymouslyToAnyServerOn(string url)
+    {
+        var descriptor = new EndpointDescription(url);
+        descriptor.UserIdentityTokens.Add(new UserTokenPolicy(UserTokenType.Anonymous));
+        descriptor.Server.ApplicationUri = null;
+        var endpoint = new ConfiguredEndpoint(null, descriptor);
+
+        var identity = new UserIdentity(new AnonymousIdentityToken());
+        return (endpoint, identity);
+    }
+
+    private static ApplicationConfiguration ConnectorDescription() => new()
+    {
+        ApplicationType = ApplicationType.Client,
+
+        ApplicationName = "RaaEDGE OPC UA Connector",
+        ApplicationUri = "https://github.com/RaaLabs/Connectors.OPCUA",
+        ProductUri = "https://github.com/RaaLabs",
+    };
+}

Source/Connector.cs

@@ -0,0 +1,59 @@
+// Copyright (c) RaaLabs. All rights reserved.
+// Licensed under the GPLv2 License. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using RaaLabs.Edge.Connectors.OPCUA.Events;
+using RaaLabs.Edge.Modules.EventHandling;
+using Serilog;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+public class Connector : IRunAsync, IProduceEvent<OpcuaDatapointOutput>
+{
+    private readonly ICreateSessions _sessions;
+    private readonly IRetrieveData _retriever;
+    private readonly ICreateDatapointsFromDataValues _datapoints;
+    private readonly ILogger _logger;
+    private readonly IMetricsHandler _metrics;
+
+    public Connector(ICreateSessions sessions, IRetrieveData retriever, ICreateDatapointsFromDataValues datapoints, ILogger logger, IMetricsHandler metrics)
+    {
+        _sessions = sessions;
+        _retriever = retriever;
+        _datapoints = datapoints;
+        _logger = logger;
+        _metrics = metrics;
+    }
+
+    public event AsyncEventEmitter<OpcuaDatapointOutput>? SendDatapoint;
+
+    public async Task Run()
+    {
+        while (true)
+        {
+            try
+            {
+                _logger.Information("Initiating connection to server");
+                var connection = await _sessions.ConnectToServer(CancellationToken.None).ConfigureAwait(false);
+
+                _logger.Information("Starting data reader");
+                await _retriever.ReadDataForever(connection, ConvertAndSendDataValue, CancellationToken.None).ConfigureAwait(false);
+
+                _logger.Warning("Data reader stopped");
+            }
+            catch (Exception error)
+            {
+                _logger.Error(error, "Failure occured while connecting or reading data");
+                await Task.Delay(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
+            }
+        }
+    }
+
+    private Task ConvertAndSendDataValue(NodeValue value)
+    {
+        _metrics.NumberOfMessagesSent(1);
+        return SendDatapoint!(_datapoints.CreateDatapointFrom(value));
+    }
+}

Source/ConnectorConfiguration.cs

@@ -0,0 +1,24 @@
+// Copyright (c) RaaLabs. All rights reserved.
+// Licensed under the GPLv2 License. See LICENSE file in the project root for full license information.
+
+using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
+using RaaLabs.Edge.Modules.Configuration;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+[Name("configuration.json"), RestartOnChange, ExcludeFromCodeCoverage]
+public class ConnectorConfiguration : IConfiguration
+{
+    public required string ServerUrl { get; init; }
+    public double PublishIntervalSeconds { get; init; } = 1.0;
+    public IList<NodeConfiguration> Nodes { get; init; } = [];
+    public string? Source { get; set; }
+}
+
+public class NodeConfiguration
+{
+    public required string NodeId { get; init; }
+    public double? SubscribeIntervalSeconds { get; init; }
+    public double? ReadIntervalSeconds { get; init; }
+}

Source/DataPointParser.cs

@@ -0,0 +1,65 @@
+// Copyright (c) RaaLabs. All rights reserved.
+// Licensed under the GPLv2 License. See LICENSE file in the project root for full license information.
+
+using System;
+using RaaLabs.Edge.Connectors.OPCUA.Events;
+using Serilog;
+using StatusCode = Opc.Ua.StatusCode;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+public class DataPointParser : ICreateDatapointsFromDataValues
+{
+    private readonly ConnectorConfiguration _configuration;
+    private readonly ILogger _logger;
+    private readonly IMetricsHandler _metrics;
+    private readonly TimeProvider _clock;
+
+    public DataPointParser(ConnectorConfiguration configuration, ILogger logger, IMetricsHandler metrics, TimeProvider clock)
+    {
+        _configuration = configuration;
+        _logger = logger;
+        _metrics = metrics;
+        _clock = clock;
+    }
+
+    public OpcuaDatapointOutput CreateDatapointFrom(NodeValue nodeValue)
+    {
+        LogWarningIfStatusCodeIsNotGood(nodeValue);
+        LogWarningIfFaultyTimestamp(nodeValue, nodeValue.Value.ServerTimestamp, "Server");
+        LogWarningIfFaultyTimestamp(nodeValue, nodeValue.Value.SourceTimestamp, "Source");
+
+        return new()
+        {
+            Source = _configuration.Source ?? "OPCUA",
+            Tag = nodeValue.Node.ToString(),
+            Value = nodeValue.Value.Value,
+            Timestamp = _clock.GetUtcNow().ToUnixTimeMilliseconds()
+        };
+    }
+
+    private void LogWarningIfStatusCodeIsNotGood(NodeValue nodeValue)
+    {
+        if (StatusCode.IsGood(nodeValue.Value.StatusCode) && StatusCode.IsNotBad(nodeValue.Value.StatusCode)) return;
+
+        _logger.Warning("Bad status code for node {NodeId} - {StatusCode}", nodeValue.Node, nodeValue.Value.StatusCode);
+        _metrics.NumberOfBadStatusCodesFor(1, nodeValue.Node.ToString()!);
+    }
+
+    private void LogWarningIfFaultyTimestamp(NodeValue nodeValue, DateTime timestamp, string timestampType)
+    {
+        var dateTimeOffset = (DateTimeOffset) timestamp;
+        var utcNow = _clock.GetUtcNow();
+
+        if (dateTimeOffset > utcNow + TimeSpan.FromMinutes(15))
+        {
+            _logger.Warning("Timestamp more than 15 minutes from the future for node {NodeValueNode} - {Timestamp}. Timestamp from {Source} is never used as property for OpcuaDatapointOutput", nodeValue.Node, timestamp, timestampType);
+            _metrics.NumberOfFutureTimestampsFor(1, nodeValue.Node.ToString());
+        }
+        else if (dateTimeOffset < utcNow - TimeSpan.FromMinutes(15))
+        {
+            _logger.Warning("Timestamp older than 15 minutes for node {NodeValueNode} - {Timestamp}. Timestamp from {Source} is never used as property for OpcuaDatapointOutput", nodeValue.Node, timestamp, timestampType);
+            _metrics.NumberOfOldTimestampsFor(1, nodeValue.Node.ToString());
+        }
+    }
+}

Source/DataReader.cs

@@ -0,0 +1,98 @@
+// Copyright (c) RaaLabs. All rights reserved.
+// Licensed under the GPLv2 License. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Opc.Ua;
+using Opc.Ua.Client;
+using Serilog;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+public class DataReader : IRetrieveData
+{
+    private readonly TimeSpan _publishInterval;
+    private readonly List<(NodeId, TimeSpan)> _subscribeNodes;
+    private readonly List<(NodeId, TimeSpan)> _readNodes;
+    private readonly ICanSubscribeToNodes _subscriber;
+    private readonly ICanReadNodes _reader;
+    private readonly ILogger _logger;
+
+    public DataReader(ConnectorConfiguration config, ICanSubscribeToNodes subscriber, ICanReadNodes reader, ILogger logger)
+    {
+        _publishInterval = TimeSpan.FromSeconds(config.PublishIntervalSeconds);
+        _subscribeNodes = config.Nodes
+            .Where(_ => _.SubscribeIntervalSeconds is not null)
+            .Select(_ => (new NodeId(_.NodeId), TimeSpan.FromSeconds(_.SubscribeIntervalSeconds!.Value)))
+            .ToList();
+        _readNodes = config.Nodes
+            .Where(_ => _.ReadIntervalSeconds is not null)
+            .Select(_ => (new NodeId(_.NodeId), TimeSpan.FromSeconds(_.ReadIntervalSeconds!.Value)))
+            .ToList();
+        _subscriber = subscriber;
+        _reader = reader;
+        _logger = logger;
+    }
+
+    public async Task ReadDataForever(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken)
+    {
+        using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+
+        var (subscription, reader) = (Task.CompletedTask, Task.CompletedTask);
+        try
+        {
+            _logger.Information("Starting data reader for {SubscriptionCount} subscriptions and {ReadCount} reads", _subscribeNodes.Count, _readNodes.Count);
+
+            subscription = SubscribeOrSleep(connection, handleValue, cts.Token);
+            reader = ReadOrSleep(connection, handleValue, cts.Token);
+
+            await Task.WhenAny(subscription, reader).ConfigureAwait(false);
+
+            ThrowIfFailedWithError(subscription);
+            ThrowIfFailedWithError(reader);
+
+            _logger.Warning("Reading data completed, it should not...");
+        }
+        catch (Exception error) when (error is not OperationCanceledException)
+        {
+            _logger.Error(error, "Failure occured while reading data");
+        }
+
+        cts.Cancel();
+        try
+        {
+            _logger.Information("Waiting for subscription and reader to complete");
+            await subscription.ConfigureAwait(false);
+            await reader.ConfigureAwait(false);
+        }
+        catch (OperationCanceledException)
+        {
+            // Ignore
+        }
+    }
+
+    private Task SubscribeOrSleep(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken) =>
+        _subscribeNodes.Count switch
+        {
+            0 => Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken),
+            _ => Task.Run(() => _subscriber.SubscribeToChangesFor(connection, _publishInterval, _subscribeNodes, handleValue, cancellationToken), cancellationToken)
+        };
+
+    private Task ReadOrSleep(ISession connection, Func<NodeValue, Task> handleValue, CancellationToken cancellationToken) =>
+        _readNodes.Count switch
+        {
+            0 => Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken),
+            _ => Task.Run(() => _reader.ReadNodesForever(connection, _readNodes, handleValue, cancellationToken), cancellationToken)
+        };
+
+    private static void ThrowIfFailedWithError(Task task)
+    {
+        if (task.Exception?.GetBaseException() is {} error and not OperationCanceledException)
+        {
+            throw error;
+        }
+    }
+}

Source/ICanReadNodes.cs

@@ -0,0 +1,16 @@
+// Copyright (c) RaaLabs. All rights reserved.
+// Licensed under the GPLv2 License. See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Opc.Ua;
+using Opc.Ua.Client;
+
+namespace RaaLabs.Edge.Connectors.OPCUA;
+
+public interface ICanReadNodes
+{
+    Task ReadNodesForever(ISession connection, IEnumerable<(NodeId node, TimeSpan readInterval)> nodes, Func<NodeValue,Task> handleValue, CancellationToken cancellationToken);
+}