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

feat: wire real LmxProxy gRPC client into Data Connection Layer

Browse Source 
Replace stub ILmxProxyClient with production proto-generated gRPC client
(RealLmxProxyClient) that connects to LmxProxy servers with x-api-key
metadata header authentication. Includes pre-generated proto stubs for
ARM64 Docker compatibility, updated adapter with proper quality mapping
(Good/Uncertain/Bad), subscription via server-streaming RPC, and 20 unit
tests covering all operations. Updated Component-DataConnectionLayer.md
to reflect the actual implementation.
This commit is contained in:
Joseph Doherty
2026-03-18 11:57:18 -04:00
parent da683d4fe9
commit e837eae2cc
10 changed files with 7022 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,13 +5,13 @@ using ScadaLink.Commons.Types.Enums;
namespace ScadaLink.DataConnectionLayer.Adapters;
/// <summary>
/// WP-8: LmxProxy adapter implementing IDataConnection.
/// Maps IDataConnection to LmxProxy SDK calls.
/// LmxProxy adapter implementing IDataConnection.
/// Maps IDataConnection operations to the LmxProxy SDK client.
///
/// LmxProxy-specific behavior:
/// - Session-based connection with 30s keep-alive
/// - gRPC streaming for subscriptions
/// - SessionId management (required for all operations)
/// - Session-based connection with automatic 30s keep-alive (managed by SDK)
/// - gRPC streaming for subscriptions via ILmxSubscription handles
/// - API key authentication via x-api-key gRPC metadata header
/// </summary>
public class LmxProxyDataConnection : IDataConnection
{
@@ -19,11 +19,10 @@ public class LmxProxyDataConnection : IDataConnection
private readonly ILogger<LmxProxyDataConnection> _logger;
private ILmxProxyClient? _client;
private string _host = "localhost";
private int _port = 5000;
private int _port = 50051;
private ConnectionHealth _status = ConnectionHealth.Disconnected;
private Timer? _keepAliveTimer;
private readonly Dictionary<string, string> _subscriptionHandles = new();
private readonly Dictionary<string, ILmxSubscription> _subscriptions = new();
public LmxProxyDataConnection(ILmxProxyClientFactory clientFactory, ILogger<LmxProxyDataConnection> logger)
{
@@ -38,82 +37,56 @@ public class LmxProxyDataConnection : IDataConnection
_host = connectionDetails.TryGetValue("Host", out var host) ? host : "localhost";
if (connectionDetails.TryGetValue("Port", out var portStr) && int.TryParse(portStr, out var port))
_port = port;
connectionDetails.TryGetValue("ApiKey", out var apiKey);
_status = ConnectionHealth.Connecting;
_client = _clientFactory.Create();
_client = _clientFactory.Create(_host, _port, apiKey);
var sessionId = await _client.OpenSessionAsync(_host, _port, cancellationToken);
await _client.ConnectAsync(cancellationToken);
_status = ConnectionHealth.Connected;
// Start 30s keep-alive timer per design spec
_keepAliveTimer = new Timer(
async _ => await SendKeepAliveAsync(),
null,
TimeSpan.FromSeconds(30),
TimeSpan.FromSeconds(30));
_logger.LogInformation("LmxProxy connected to {Host}:{Port}, sessionId={SessionId}", _host, _port, sessionId);
_logger.LogInformation("LmxProxy connected to {Host}:{Port}", _host, _port);
}
public async Task DisconnectAsync(CancellationToken cancellationToken = default)
{
_keepAliveTimer?.Dispose();
_keepAliveTimer = null;
if (_client != null)
{
await _client.CloseSessionAsync(cancellationToken);
await _client.DisconnectAsync();
_status = ConnectionHealth.Disconnected;
_logger.LogInformation("LmxProxy disconnected from {Host}:{Port}", _host, _port);
}
}
public async Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
{
EnsureConnected();
var handle = await _client!.SubscribeTagAsync(
tagPath,
(path, value, timestamp, isGood) =>
{
var quality = isGood ? QualityCode.Good : QualityCode.Bad;
callback(path, new TagValue(value, quality, new DateTimeOffset(timestamp, TimeSpan.Zero)));
},
cancellationToken);
_subscriptionHandles[handle] = tagPath;
return handle;
}
public async Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
{
if (_client != null)
{
await _client.UnsubscribeTagAsync(subscriptionId, cancellationToken);
_subscriptionHandles.Remove(subscriptionId);
}
}
public async Task<ReadResult> ReadAsync(string tagPath, CancellationToken cancellationToken = default)
{
EnsureConnected();
var (value, timestamp, isGood) = await _client!.ReadTagAsync(tagPath, cancellationToken);
var quality = isGood ? QualityCode.Good : QualityCode.Bad;
var vtq = await _client!.ReadAsync(tagPath, cancellationToken);
var quality = MapQuality(vtq.Quality);
var tagValue = new TagValue(vtq.Value, quality, new DateTimeOffset(vtq.TimestampUtc, TimeSpan.Zero));
if (!isGood)
return new ReadResult(false, null, "LmxProxy read returned bad quality");
return new ReadResult(true, new TagValue(value, quality, new DateTimeOffset(timestamp, TimeSpan.Zero)), null);
return vtq.Quality == LmxQuality.Bad
? new ReadResult(false, tagValue, "LmxProxy read returned bad quality")
: new ReadResult(true, tagValue, null);
}
public async Task<IReadOnlyDictionary<string, ReadResult>> ReadBatchAsync(IEnumerable<string> tagPaths, CancellationToken cancellationToken = default)
{
EnsureConnected();
var vtqs = await _client!.ReadBatchAsync(tagPaths, cancellationToken);
var results = new Dictionary<string, ReadResult>();
foreach (var tagPath in tagPaths)
foreach (var (tag, vtq) in vtqs)
{
results[tagPath] = await ReadAsync(tagPath, cancellationToken);
var quality = MapQuality(vtq.Quality);
var tagValue = new TagValue(vtq.Value, quality, new DateTimeOffset(vtq.TimestampUtc, TimeSpan.Zero));
results[tag] = vtq.Quality == LmxQuality.Bad
? new ReadResult(false, tagValue, "LmxProxy read returned bad quality")
: new ReadResult(true, tagValue, null);
}
return results;
}
@@ -121,20 +94,35 @@ public class LmxProxyDataConnection : IDataConnection
{
EnsureConnected();
var success = await _client!.WriteTagAsync(tagPath, value, cancellationToken);
return success
? new WriteResult(true, null)
: new WriteResult(false, "LmxProxy write failed");
try
{
await _client!.WriteAsync(tagPath, value!, cancellationToken);
return new WriteResult(true, null);
}
catch (Exception ex)
{
return new WriteResult(false, ex.Message);
}
}
public async Task<IReadOnlyDictionary<string, WriteResult>> WriteBatchAsync(IDictionary<string, object?> values, CancellationToken cancellationToken = default)
{
var results = new Dictionary<string, WriteResult>();
foreach (var (tagPath, value) in values)
EnsureConnected();
try
{
results[tagPath] = await WriteAsync(tagPath, value, cancellationToken);
var nonNullValues = values.Where(kv => kv.Value != null)
.ToDictionary(kv => kv.Key, kv => kv.Value!);
await _client!.WriteBatchAsync(nonNullValues, cancellationToken);
return values.Keys.ToDictionary(k => k, _ => new WriteResult(true, null))
as IReadOnlyDictionary<string, WriteResult>;
}
catch (Exception ex)
{
return values.Keys.ToDictionary(k => k, _ => new WriteResult(false, ex.Message))
as IReadOnlyDictionary<string, WriteResult>;
}
return results;
}
public async Task<bool> WriteBatchAndWaitAsync(
@@ -162,10 +150,40 @@ public class LmxProxyDataConnection : IDataConnection
return false;
}
public async Task<string> SubscribeAsync(string tagPath, SubscriptionCallback callback, CancellationToken cancellationToken = default)
{
EnsureConnected();
var subscription = await _client!.SubscribeAsync(
[tagPath],
(path, vtq) =>
{
var quality = MapQuality(vtq.Quality);
callback(path, new TagValue(vtq.Value, quality, new DateTimeOffset(vtq.TimestampUtc, TimeSpan.Zero)));
},
cancellationToken);
var subscriptionId = Guid.NewGuid().ToString("N");
_subscriptions[subscriptionId] = subscription;
return subscriptionId;
}
public async Task UnsubscribeAsync(string subscriptionId, CancellationToken cancellationToken = default)
{
if (_subscriptions.Remove(subscriptionId, out var subscription))
{
await subscription.DisposeAsync();
}
}
public async ValueTask DisposeAsync()
{
_keepAliveTimer?.Dispose();
_keepAliveTimer = null;
foreach (var subscription in _subscriptions.Values)
{
try { await subscription.DisposeAsync(); }
catch { /* best-effort cleanup */ }
}
_subscriptions.Clear();
if (_client != null)
{
@@ -181,16 +199,11 @@ public class LmxProxyDataConnection : IDataConnection
throw new InvalidOperationException("LmxProxy client is not connected.");
}
private async Task SendKeepAliveAsync()
private static QualityCode MapQuality(LmxQuality quality) => quality switch
{
try
{
if (_client?.IsConnected == true)
await _client.SendKeepAliveAsync();
}
catch (Exception ex)
{
_logger.LogWarning(ex, "LmxProxy keep-alive failed for {Host}:{Port}", _host, _port);
}
}
LmxQuality.Good => QualityCode.Good,
LmxQuality.Uncertain => QualityCode.Uncertain,
LmxQuality.Bad => QualityCode.Bad,
_ => QualityCode.Bad
};
}