feat(runtime,host): close F7 — driver subscribe + write paths + Host DI
Some checks failed
v2-ci / build (push) Failing after 42s
v2-ci / unit-tests (tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Security.Tests) (push) Has been skipped
v2-ci / integration (push) Has been skipped
Some checks failed
v2-ci / build (push) Failing after 42s
v2-ci / unit-tests (tests/Core/ZB.MOM.WW.OtOpcUa.Cluster.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests) (push) Has been skipped
v2-ci / unit-tests (tests/Server/ZB.MOM.WW.OtOpcUa.Security.Tests) (push) Has been skipped
v2-ci / integration (push) Has been skipped
Three pieces landed in one batch, closing F7-residual + Host DI #106: Runtime/DriverInstanceActor: - Subscribe / Unsubscribe message contracts; the Connected state handles them via IDriver.ISubscribable. On every OnDataChange event the actor publishes AttributeValuePublished to its parent (DriverHostActor → OpcUaPublishActor). OPC UA StatusCode is mapped to the 3-state OpcUaQuality enum via severity bits (00=Good, 01=Uncertain, 10/11=Bad). - DetachSubscription tears the handler off the driver on DisconnectObserved, Unsubscribe, and PostStop so a stale handler never pushes to a dead actor. - WriteAttribute now dispatches IWritable.WriteAsync (batch of one) with a 5s CancellationTokenSource; status-code propagated to WriteAttributeResult on non-Good results. Host: - New ProjectReferences to Core + every cross-platform driver assembly (AbCip/AbLegacy/FOCAS/Galaxy/Modbus/S7/TwinCAT). Galaxy is net10 (gRPC client to mxaccessgw); the COM-bound net48 Wonderware Historian driver stays out of the Host's reference closure — its .Client gRPC wrapper is what binds for historian needs. - New DriverFactoryBootstrap.AddOtOpcUaDriverFactories() registers a singleton DriverFactoryRegistry, invokes each driver's Register(registry, loggerFactory), and binds IDriverFactory to DriverFactoryRegistryAdapter. Replaces the F7 NullDriverFactory default so deploys actually materialise real IDriver instances on driver-role nodes. ShouldStub() still gates per-platform behaviour at spawn time. - Program.cs wires AddOtOpcUaDriverFactories() before AddAkka so the runtime extension can resolve IDriverFactory from DI. Tests: Runtime 46 -> 52 (+6): - Write returns success when StatusCode = Good - Write propagates non-Good status code in failure Reason - Subscribe forwards OnDataChange to parent as AttributeValuePublished - Quality translation: Uncertain (0x40...) and Bad (0x80...) - Subscribe against non-ISubscribable returns failure - DisconnectObserved detaches handler so late events are dropped All 6 v2 test suites green: 152 tests passing. Closes F7. F7-residual sub-tasks #110 (subscribe) and #111 (write) both shipped. Host DI binding #106 shipped.
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Hosting;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Host.Drivers;
|
||||
|
||||
/// <summary>
|
||||
/// Wires every cross-platform driver assembly's <c>Register(registry, loggerFactory)</c>
|
||||
/// extension into a single <see cref="DriverFactoryRegistry"/> singleton and binds the
|
||||
/// v2 <see cref="IDriverFactory"/> abstraction to a <see cref="DriverFactoryRegistryAdapter"/>
|
||||
/// over it. Replaces the F7 seam's <c>NullDriverFactory</c> default so deploys actually
|
||||
/// materialise real <see cref="IDriver"/> instances on driver-role nodes.
|
||||
///
|
||||
/// Skipped entirely on admin-only nodes — they never run drivers, so the registry doesn't
|
||||
/// need to exist (Program.cs guards via the <c>hasDriver</c> flag).
|
||||
/// </summary>
|
||||
public static class DriverFactoryBootstrap
|
||||
{
|
||||
/// <summary>
|
||||
/// Register the cross-platform driver factories + bind <see cref="IDriverFactory"/>.
|
||||
/// Must be called BEFORE <c>services.AddAkka</c> so the runtime extension can resolve
|
||||
/// <see cref="IDriverFactory"/> from DI when spawning <c>DriverHostActor</c>.
|
||||
/// </summary>
|
||||
public static IServiceCollection AddOtOpcUaDriverFactories(this IServiceCollection services)
|
||||
{
|
||||
services.AddSingleton<DriverFactoryRegistry>(sp =>
|
||||
{
|
||||
var registry = new DriverFactoryRegistry();
|
||||
var loggerFactory = sp.GetService<ILoggerFactory>();
|
||||
Register(registry, loggerFactory);
|
||||
return registry;
|
||||
});
|
||||
services.AddSingleton<IDriverFactory>(sp =>
|
||||
new DriverFactoryRegistryAdapter(sp.GetRequiredService<DriverFactoryRegistry>()));
|
||||
return services;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Invoke every cross-platform driver's <c>Register</c> extension. New driver assemblies
|
||||
/// get added here — one line per type. ShouldStub() in <c>DriverInstanceActor</c> still
|
||||
/// handles platform/role-dependent stubbing (e.g. Galaxy on macOS), so registering a
|
||||
/// factory here doesn't mean it always runs in production.
|
||||
/// </summary>
|
||||
private static void Register(DriverFactoryRegistry registry, ILoggerFactory? loggerFactory)
|
||||
{
|
||||
Driver.AbCip.AbCipDriverFactoryExtensions.Register(registry);
|
||||
Driver.AbLegacy.AbLegacyDriverFactoryExtensions.Register(registry, loggerFactory);
|
||||
Driver.FOCAS.FocasDriverFactoryExtensions.Register(registry);
|
||||
Driver.Galaxy.GalaxyDriverFactoryExtensions.Register(registry, loggerFactory);
|
||||
Driver.Modbus.ModbusDriverFactoryExtensions.Register(registry, loggerFactory);
|
||||
Driver.S7.S7DriverFactoryExtensions.Register(registry);
|
||||
Driver.TwinCAT.TwinCATDriverFactoryExtensions.Register(registry);
|
||||
}
|
||||
}
|
||||
@@ -8,6 +8,7 @@ using ZB.MOM.WW.OtOpcUa.Cluster;
|
||||
using ZB.MOM.WW.OtOpcUa.Configuration;
|
||||
using ZB.MOM.WW.OtOpcUa.ControlPlane;
|
||||
using ZB.MOM.WW.OtOpcUa.Host;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.Drivers;
|
||||
using ZB.MOM.WW.OtOpcUa.Host.Health;
|
||||
using ZB.MOM.WW.OtOpcUa.Runtime;
|
||||
using ZB.MOM.WW.OtOpcUa.Security;
|
||||
@@ -40,7 +41,13 @@ builder.Services.AddOtOpcUaConfigDb(builder.Configuration);
|
||||
builder.Services.AddOtOpcUaCluster(builder.Configuration);
|
||||
|
||||
if (hasDriver)
|
||||
{
|
||||
builder.Services.AddOtOpcUaRuntime();
|
||||
// Bind every cross-platform driver factory before AddAkka resolves IDriverFactory — replaces
|
||||
// the F7-default NullDriverFactory with a real DriverFactoryRegistryAdapter so DriverHostActor
|
||||
// can materialise real IDriver instances on deploy.
|
||||
builder.Services.AddOtOpcUaDriverFactories();
|
||||
}
|
||||
|
||||
// Akka cluster bootstrap. Role-specific singletons are registered on the AkkaConfigurationBuilder
|
||||
// from inside the configurator lambda. AddAkka spins the ActorSystem at host start.
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Commons\ZB.MOM.WW.OtOpcUa.Commons.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Cluster\ZB.MOM.WW.OtOpcUa.Cluster.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Configuration\ZB.MOM.WW.OtOpcUa.Configuration.csproj"/>
|
||||
<ProjectReference Include="..\..\Core\ZB.MOM.WW.OtOpcUa.Core\ZB.MOM.WW.OtOpcUa.Core.csproj"/>
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Security\ZB.MOM.WW.OtOpcUa.Security.csproj"/>
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.ControlPlane\ZB.MOM.WW.OtOpcUa.ControlPlane.csproj"/>
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.Runtime\ZB.MOM.WW.OtOpcUa.Runtime.csproj"/>
|
||||
@@ -28,6 +29,22 @@
|
||||
<ProjectReference Include="..\ZB.MOM.WW.OtOpcUa.AdminUI\ZB.MOM.WW.OtOpcUa.AdminUI.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<!-- Cross-platform driver assemblies. Each Register(registry, loggerFactory) extension is
|
||||
called from DriverFactoryBootstrap on driver-role nodes; the F7 seam (IDriverFactory)
|
||||
then exposes the registry to DriverHostActor. Galaxy is net10 because it talks gRPC to
|
||||
the out-of-process mxaccessgw worker — the COM-bound net48 piece is over there.
|
||||
Historian.Wonderware (the net48 COM-bridge driver) is intentionally excluded; the
|
||||
net10 .Client gRPC wrapper is what production binds when the historian role is needed. -->
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.AbCip\ZB.MOM.WW.OtOpcUa.Driver.AbCip.csproj"/>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.AbLegacy\ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.csproj"/>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.FOCAS\ZB.MOM.WW.OtOpcUa.Driver.FOCAS.csproj"/>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Galaxy\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.csproj"/>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Modbus\ZB.MOM.WW.OtOpcUa.Driver.Modbus.csproj"/>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.S7\ZB.MOM.WW.OtOpcUa.Driver.S7.csproj"/>
|
||||
<ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.TwinCAT\ZB.MOM.WW.OtOpcUa.Driver.TwinCAT.csproj"/>
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<!-- OpenTelemetry.Api transitively via Akka; Opc.Ua.Core transitively via OpcUaServer. -->
|
||||
<NuGetAuditSuppress Include="https://github.com/advisories/GHSA-g94r-2vxg-569j"/>
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
using Akka.Actor;
|
||||
using Akka.Event;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.OpcUa;
|
||||
using ZB.MOM.WW.OtOpcUa.Commons.Types;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
@@ -31,6 +32,14 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
public sealed record ApplyResult(bool Success, string? Reason, CorrelationId Correlation);
|
||||
public sealed record WriteAttribute(string TagId, object Value);
|
||||
public sealed record WriteAttributeResult(bool Success, string? Reason);
|
||||
public sealed record Subscribe(IReadOnlyList<string> FullReferences, TimeSpan PublishingInterval);
|
||||
public sealed record SubscriptionEstablished(string DiagnosticId, int ReferenceCount);
|
||||
public sealed record SubscriptionFailed(string Reason);
|
||||
public sealed record Unsubscribe;
|
||||
/// <summary>Published to the actor's parent whenever the subscribed IDriver fires
|
||||
/// <see cref="ISubscribable.OnDataChange"/>. The parent forwards to OpcUaPublishActor.</summary>
|
||||
public sealed record AttributeValuePublished(string FullReference, object? Value, OpcUaQuality Quality, DateTime TimestampUtc);
|
||||
private sealed record DataChangeForward(string FullReference, DataValueSnapshot Snapshot);
|
||||
public sealed class RetryConnect
|
||||
{
|
||||
public static readonly RetryConnect Instance = new();
|
||||
@@ -43,6 +52,12 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
private readonly ILoggingAdapter _log = Context.GetLogger();
|
||||
private string? _currentConfigJson;
|
||||
|
||||
/// <summary>Active subscription handle (null when not subscribed). Lifetime is one-per-actor —
|
||||
/// re-subscribe across reconnects is the consumer's responsibility today (subscribe-once
|
||||
/// semantics keep the actor simple; mux-driven re-subscribe is tracked as F8b/#113).</summary>
|
||||
private ISubscriptionHandle? _subscriptionHandle;
|
||||
private EventHandler<DataChangeEventArgs>? _dataChangeHandler;
|
||||
|
||||
public ITimerScheduler Timers { get; set; } = null!;
|
||||
|
||||
public static Props Props(IDriver driver, TimeSpan? reconnectInterval = null, bool startStubbed = false) =>
|
||||
@@ -111,9 +126,13 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
{
|
||||
_log.Warning("DriverInstance {Id}: disconnect observed ({Reason}); reconnecting",
|
||||
_driverInstanceId, msg.Reason);
|
||||
DetachSubscription();
|
||||
Become(Reconnecting);
|
||||
});
|
||||
Receive<WriteAttribute>(HandleWrite);
|
||||
ReceiveAsync<WriteAttribute>(HandleWriteAsync);
|
||||
ReceiveAsync<Subscribe>(HandleSubscribeAsync);
|
||||
ReceiveAsync<Unsubscribe>(_ => UnsubscribeAsync());
|
||||
Receive<DataChangeForward>(OnDataChangeForward);
|
||||
}
|
||||
|
||||
private void Reconnecting()
|
||||
@@ -162,22 +181,137 @@ public sealed class DriverInstanceActor : ReceiveActor, IWithTimers
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleWrite(WriteAttribute msg)
|
||||
private async Task HandleWriteAsync(WriteAttribute msg)
|
||||
{
|
||||
// Per-tag write requires IWritable capability discovery. Skeleton stub — see follow-up F7.
|
||||
if (_driver is IWritable writable)
|
||||
{
|
||||
// Future: writable.WriteAsync(msg.TagId, msg.Value, ct) and Pipe back to Sender.
|
||||
Sender.Tell(new WriteAttributeResult(false, "Write path not yet implemented (F7)"));
|
||||
}
|
||||
else
|
||||
if (_driver is not IWritable writable)
|
||||
{
|
||||
Sender.Tell(new WriteAttributeResult(false, "Driver does not implement IWritable"));
|
||||
return;
|
||||
}
|
||||
|
||||
var replyTo = Sender;
|
||||
var request = new[] { new WriteRequest(msg.TagId, msg.Value) };
|
||||
// Bound the write so a hung backend can't pin this actor forever — decision #44/#45 keeps
|
||||
// retry off by default, but a stalled call still needs an answer.
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
try
|
||||
{
|
||||
var results = await writable.WriteAsync(request, cts.Token).ConfigureAwait(false);
|
||||
if (results is { Count: 1 } && IsGoodStatus(results[0].StatusCode))
|
||||
{
|
||||
replyTo.Tell(new WriteAttributeResult(true, null));
|
||||
return;
|
||||
}
|
||||
var status = results is { Count: > 0 } ? results[0].StatusCode : 0xFFFFFFFF;
|
||||
replyTo.Tell(new WriteAttributeResult(false, $"StatusCode=0x{status:X8}"));
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
replyTo.Tell(new WriteAttributeResult(false, "write timeout"));
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
replyTo.Tell(new WriteAttributeResult(false, ex.Message));
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleSubscribeAsync(Subscribe msg)
|
||||
{
|
||||
if (_driver is not ISubscribable subscribable)
|
||||
{
|
||||
Sender.Tell(new SubscriptionFailed("Driver does not implement ISubscribable"));
|
||||
return;
|
||||
}
|
||||
if (_subscriptionHandle is not null)
|
||||
{
|
||||
// Subscribe-twice — drop the prior subscription before establishing the new one.
|
||||
await UnsubscribeAsync().ConfigureAwait(false);
|
||||
}
|
||||
|
||||
var replyTo = Sender;
|
||||
var self = Self;
|
||||
try
|
||||
{
|
||||
_dataChangeHandler = (_, args) => self.Tell(new DataChangeForward(args.FullReference, args.Snapshot));
|
||||
subscribable.OnDataChange += _dataChangeHandler;
|
||||
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
|
||||
_subscriptionHandle = await subscribable
|
||||
.SubscribeAsync(msg.FullReferences, msg.PublishingInterval, cts.Token)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
replyTo.Tell(new SubscriptionEstablished(_subscriptionHandle.DiagnosticId, msg.FullReferences.Count));
|
||||
_log.Info("DriverInstance {Id}: subscribed to {Count} refs ({Diag})",
|
||||
_driverInstanceId, msg.FullReferences.Count, _subscriptionHandle.DiagnosticId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
DetachSubscription();
|
||||
_log.Warning(ex, "DriverInstance {Id}: subscribe failed", _driverInstanceId);
|
||||
replyTo.Tell(new SubscriptionFailed(ex.Message));
|
||||
}
|
||||
}
|
||||
|
||||
private async Task UnsubscribeAsync()
|
||||
{
|
||||
if (_driver is not ISubscribable subscribable || _subscriptionHandle is null)
|
||||
{
|
||||
DetachSubscription();
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(5));
|
||||
await subscribable.UnsubscribeAsync(_subscriptionHandle, cts.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_log.Warning(ex, "DriverInstance {Id}: unsubscribe threw (continuing)", _driverInstanceId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
DetachSubscription();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Tear down the event handler + null the handle. Called from Unsubscribe path, on
|
||||
/// PostStop, and on Connected → Reconnecting transitions so a stale handler doesn't push
|
||||
/// data-change events to an actor that has lost its driver connection.</summary>
|
||||
private void DetachSubscription()
|
||||
{
|
||||
if (_driver is ISubscribable subscribable && _dataChangeHandler is not null)
|
||||
{
|
||||
subscribable.OnDataChange -= _dataChangeHandler;
|
||||
}
|
||||
_dataChangeHandler = null;
|
||||
_subscriptionHandle = null;
|
||||
}
|
||||
|
||||
private void OnDataChangeForward(DataChangeForward msg)
|
||||
{
|
||||
var quality = QualityFromStatus(msg.Snapshot.StatusCode);
|
||||
var ts = msg.Snapshot.SourceTimestampUtc ?? msg.Snapshot.ServerTimestampUtc;
|
||||
Context.Parent.Tell(new AttributeValuePublished(msg.FullReference, msg.Snapshot.Value, quality, ts));
|
||||
}
|
||||
|
||||
/// <summary>Translate an OPC UA status code to the 3-state <see cref="OpcUaQuality"/> projection
|
||||
/// the publish actor consumes. Severity bits (top 2): 00 = Good, 01 = Uncertain, 10/11 = Bad.</summary>
|
||||
private static OpcUaQuality QualityFromStatus(uint statusCode)
|
||||
{
|
||||
var severity = statusCode >> 30;
|
||||
return severity switch
|
||||
{
|
||||
0 => OpcUaQuality.Good,
|
||||
1 => OpcUaQuality.Uncertain,
|
||||
_ => OpcUaQuality.Bad,
|
||||
};
|
||||
}
|
||||
|
||||
private static bool IsGoodStatus(uint statusCode) => (statusCode >> 30) == 0;
|
||||
|
||||
protected override void PostStop()
|
||||
{
|
||||
DetachSubscription();
|
||||
try { _driver.ShutdownAsync(CancellationToken.None).GetAwaiter().GetResult(); }
|
||||
catch (Exception ex) { _log.Warning(ex, "DriverInstance {Id}: ShutdownAsync threw on PostStop", _driverInstanceId); }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user