feat(dcl): MxGatewayDataConnection adapter (connect/subscribe/read/write/wait/browse)
Implements IDataConnection + IBrowsableDataConnection over the IMxGatewayClient seam: connect/disconnect with once-only Disconnected guard + background event loop, subscribe/unsubscribe with tag routing, read/write batch with per-tag error classification, WriteBatchAndWait, and Galaxy browse mapping. Covers plan Tasks 6-10. Full unit coverage via FakeMxGatewayClient (12 tests).
This commit is contained in:
Joseph Doherty
@@ -0,0 +1,220 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Protocol;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Serialization;
|
||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters;
|
||||
|
||||
/// <summary>
|
||||
/// MxGateway adapter implementing <see cref="IDataConnection"/> + <see cref="IBrowsableDataConnection"/>.
|
||||
/// Maps IDataConnection concepts onto the MxAccess Gateway session model via the
|
||||
/// <see cref="IMxGatewayClient"/> seam:
|
||||
/// <list type="bullet">
|
||||
/// <item>Connect → OpenSession + Register, then a background event loop.</item>
|
||||
/// <item>Subscribe → AddItem + Advise; value changes arrive on the event stream.</item>
|
||||
/// <item>Read/Write → ReadBulk / WriteBulk.</item>
|
||||
/// <item>Browse → Galaxy repository BrowseChildren.</item>
|
||||
/// </list>
|
||||
/// Reconnection is driven by the <c>DataConnectionActor</c>: a stream fault raises
|
||||
/// <see cref="Disconnected"/>, the actor disposes this adapter, creates a fresh one,
|
||||
/// reconnects and re-subscribes all tags.
|
||||
/// </summary>
|
||||
public class MxGatewayDataConnection : IDataConnection, IBrowsableDataConnection
|
||||
{
|
||||
private readonly IMxGatewayClientFactory _clientFactory;
|
||||
private readonly ILogger<MxGatewayDataConnection> _logger;
|
||||
private IMxGatewayClient? _client;
|
||||
private ConnectionHealth _status = ConnectionHealth.Disconnected;
|
||||
private CancellationTokenSource? _eventLoopCts;
|
||||
|
||||
// subscriptionId → (tagPath, callback) so the event loop can route updates by tag,
|
||||
// plus tagPath → subscriptionId for reverse lookup. Concurrent because the event
|
||||
// loop reads from a background thread while Subscribe/Unsubscribe mutate.
|
||||
private readonly ConcurrentDictionary<string, (string TagPath, SubscriptionCallback Callback)> _subs = new();
|
||||
private readonly ConcurrentDictionary<string, string> _tagToSub = new();
|
||||
|
||||
// DataConnectionLayer mirror of OpcUaDataConnection's once-only guard: an int toggled
|
||||
// with Interlocked.Exchange so only the first caller raises Disconnected.
|
||||
// 0 = not fired, 1 = fired. Reset on (re)connect.
|
||||
private int _disconnectFired;
|
||||
|
||||
/// <summary>Initializes a new instance of <see cref="MxGatewayDataConnection"/>.</summary>
|
||||
/// <param name="clientFactory">Factory used to create gateway client instances.</param>
|
||||
/// <param name="logger">Logger instance.</param>
|
||||
public MxGatewayDataConnection(IMxGatewayClientFactory clientFactory, ILogger<MxGatewayDataConnection> logger)
|
||||
{
|
||||
_clientFactory = clientFactory;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public ConnectionHealth Status => _status;
|
||||
|
||||
/// <summary>Raised once when the gateway event stream faults (connection lost).</summary>
|
||||
public event Action? Disconnected;
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task ConnectAsync(IDictionary<string, string> connectionDetails, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var cfg = MxGatewayEndpointConfigSerializer.FromFlatDict(connectionDetails);
|
||||
Interlocked.Exchange(ref _disconnectFired, 0); // reset guard on (re)connect, like OPC UA
|
||||
_client = _clientFactory.Create();
|
||||
|
||||
await _client.ConnectAsync(new MxGatewayConnectionOptions(
|
||||
cfg.Endpoint,
|
||||
cfg.ApiKey,
|
||||
string.IsNullOrWhiteSpace(cfg.ClientName) ? "scadabridge" : cfg.ClientName,
|
||||
cfg.WriteUserId,
|
||||
cfg.UseTls,
|
||||
string.IsNullOrWhiteSpace(cfg.CaFile) ? null : cfg.CaFile,
|
||||
string.IsNullOrWhiteSpace(cfg.ServerName) ? null : cfg.ServerName,
|
||||
cfg.ReadTimeoutMs), cancellationToken);
|
||||
|
||||
_status = ConnectionHealth.Connected;
|
||||
|
||||
// Background event loop: route each value change to the matching subscription callback.
|
||||
_eventLoopCts = new CancellationTokenSource();
|
||||
_ = Task.Run(() => RunEventLoopAsync(_eventLoopCts.Token));
|
||||
}
|
||||
|
||||
private async Task RunEventLoopAsync(CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
await _client!.RunEventLoopAsync(update =>
|
||||
{
|
||||
if (_tagToSub.TryGetValue(update.TagPath, out var subId) && _subs.TryGetValue(subId, out var s))
|
||||
s.Callback(update.TagPath, new TagValue(update.Value, update.Quality, update.Timestamp));
|
||||
}, ct);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Normal shutdown (DisconnectAsync / DisposeAsync cancelled the loop).
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "MxGateway event stream faulted; signalling disconnect");
|
||||
RaiseDisconnected();
|
||||
}
|
||||
}
|
||||
|
||||
private void RaiseDisconnected()
|
||||
{
|
||||
if (Interlocked.Exchange(ref _disconnectFired, 1) == 0)
|
||||
{
|
||||
_status = ConnectionHealth.Disconnected;
|
||||
Disconnected?.Invoke();
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task DisconnectAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
_eventLoopCts?.Cancel();
|
||||
if (_client is not null)
|
||||
await _client.DisconnectAsync(cancellationToken);
|
||||
_status = ConnectionHealth.Disconnected;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var subId = await _client!.SubscribeAsync(tagPath, cancellationToken);
|
||||
_subs[subId] = (tagPath, callback);
|
||||
_tagToSub[tagPath] = subId;
|
||||
return subId;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_subs.TryRemove(subscriptionId, out var s))
|
||||
_tagToSub.TryRemove(s.TagPath, out _);
|
||||
await _client!.UnsubscribeAsync(subscriptionId, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var r = (await _client!.ReadAsync(new[] { tagPath }, cancellationToken)).Single();
|
||||
return ToReadResult(r);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var list = tagPaths.ToList();
|
||||
var results = await _client!.ReadAsync(list, cancellationToken);
|
||||
return results.ToDictionary(r => r.TagPath, ToReadResult);
|
||||
}
|
||||
|
||||
private static ReadResult ToReadResult(MxReadOutcome r) => r.Success
|
||||
? new ReadResult(true, new TagValue(r.Value, r.Quality, r.Timestamp), null)
|
||||
: new ReadResult(false, null, r.Error);
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<WriteResult> WriteAsync(string tagPath, object? value, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var w = (await _client!.WriteAsync(new[] { (tagPath, value) }, cancellationToken)).Single();
|
||||
return new WriteResult(w.Success, w.Error);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var results = await _client!.WriteAsync(values.Select(kv => (kv.Key, kv.Value)).ToList(), cancellationToken);
|
||||
return results.ToDictionary(w => w.TagPath, w => new WriteResult(w.Success, w.Error));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<bool> WriteBatchAndWaitAsync(
|
||||
IDictionary<string, object?> values, string flagPath, object? flagValue,
|
||||
string responsePath, object? responseValue, TimeSpan timeout, CancellationToken cancellationToken = default)
|
||||
{
|
||||
await WriteBatchAsync(values, cancellationToken);
|
||||
await WriteAsync(flagPath, flagValue, cancellationToken);
|
||||
|
||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeoutCts.CancelAfter(timeout);
|
||||
try
|
||||
{
|
||||
while (!timeoutCts.IsCancellationRequested)
|
||||
{
|
||||
var r = await ReadAsync(responsePath, timeoutCts.Token);
|
||||
// r.Value is a TagValue wrapper; compare its underlying scalar. String
|
||||
// projection tolerates numeric type differences across the gRPC boundary.
|
||||
if (r.Success && string.Equals(r.Value?.Value?.ToString(), responseValue?.ToString(), StringComparison.Ordinal))
|
||||
return true;
|
||||
await Task.Delay(TimeSpan.FromMilliseconds(200), timeoutCts.Token);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
// Timeout elapsed (the linked CTS, not the caller's token) — fall through to false.
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<BrowseChildrenResult> BrowseChildrenAsync(string? parentNodeId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (_status != ConnectionHealth.Connected || _client is null)
|
||||
throw new ConnectionNotConnectedException($"MxGateway connection is not connected (status: {_status}).");
|
||||
|
||||
var (children, truncated) = await _client.BrowseChildrenAsync(parentNodeId, cancellationToken);
|
||||
var nodes = children
|
||||
.Select(c => new BrowseNode(c.NodeId, c.DisplayName, c.NodeClass, c.HasChildren))
|
||||
.ToList();
|
||||
return new BrowseChildrenResult(nodes, truncated);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
_eventLoopCts?.Cancel();
|
||||
if (_client is not null)
|
||||
await _client.DisposeAsync();
|
||||
GC.SuppressFinalize(this);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user