diff --git a/src/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs b/src/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs index 5d740a5..79c7e7c 100644 --- a/src/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs +++ b/src/ZB.MOM.WW.OtOpcUa.Driver.AbCip/AbCipDriver.cs @@ -20,7 +20,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip; /// from native-heap growth that the CLR allocator can't see; it tears down every /// and reconnects each device. /// -public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable, IDisposable, IAsyncDisposable +public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, ISubscribable, + IHostConnectivityProbe, IPerCallHostResolver, IDisposable, IAsyncDisposable { private readonly AbCipDriverOptions _options; private readonly string _driverInstanceId; @@ -33,6 +34,7 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, private DriverHealth _health = new(DriverState.Unknown, null, null); public event EventHandler? OnDataChange; + public event EventHandler? OnHostStatusChanged; public AbCipDriver(AbCipDriverOptions options, string driverInstanceId, IAbCipTagFactory? tagFactory = null, @@ -70,9 +72,6 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } foreach (var tag in _options.Tags) { - // UDT tags with declared Members fan out into synthetic member-tag entries addressable - // by composed full-reference. Parent structure tag also stored so discovery can emit a - // folder for it. _tagsByName[tag.Name] = tag; if (tag.DataType == AbCipDataType.Structure && tag.Members is { Count: > 0 }) { @@ -89,6 +88,17 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, } } } + + // Probe loops — one per device when enabled + a ProbeTagPath is configured. + if (_options.Probe.Enabled && !string.IsNullOrWhiteSpace(_options.Probe.ProbeTagPath)) + { + foreach (var state in _devices.Values) + { + state.ProbeCts = new CancellationTokenSource(); + var ct = state.ProbeCts.Token; + _ = Task.Run(() => ProbeLoopAsync(state, ct), ct); + } + } _health = new DriverHealth(DriverState.Healthy, DateTime.UtcNow, null); } catch (Exception ex) @@ -109,7 +119,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, { await _poll.DisposeAsync().ConfigureAwait(false); foreach (var state in _devices.Values) + { + try { state.ProbeCts?.Cancel(); } catch { } + state.ProbeCts?.Dispose(); + state.ProbeCts = null; state.DisposeHandles(); + } _devices.Clear(); _tagsByName.Clear(); _health = new DriverHealth(DriverState.Unknown, _health.LastSuccessfulRead, null); @@ -127,6 +142,90 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, return Task.CompletedTask; } + // ---- IHostConnectivityProbe ---- + + public IReadOnlyList GetHostStatuses() => + [.. _devices.Values.Select(s => new HostConnectivityStatus(s.Options.HostAddress, s.HostState, s.HostStateChangedUtc))]; + + private async Task ProbeLoopAsync(DeviceState state, CancellationToken ct) + { + var probeParams = new AbCipTagCreateParams( + Gateway: state.ParsedAddress.Gateway, + Port: state.ParsedAddress.Port, + CipPath: state.ParsedAddress.CipPath, + LibplctagPlcAttribute: state.Profile.LibplctagPlcAttribute, + TagName: _options.Probe.ProbeTagPath!, + Timeout: _options.Probe.Timeout); + + IAbCipTagRuntime? probeRuntime = null; + while (!ct.IsCancellationRequested) + { + var success = false; + try + { + probeRuntime ??= _tagFactory.Create(probeParams); + // Lazy-init on first attempt; re-init after a transport failure has caused the + // native handle to be destroyed. + if (!state.ProbeInitialized) + { + await probeRuntime.InitializeAsync(ct).ConfigureAwait(false); + state.ProbeInitialized = true; + } + await probeRuntime.ReadAsync(ct).ConfigureAwait(false); + success = probeRuntime.GetStatus() == 0; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + break; + } + catch + { + // Wire / init error — tear down the probe runtime so the next tick re-creates it. + try { probeRuntime?.Dispose(); } catch { } + probeRuntime = null; + state.ProbeInitialized = false; + } + + TransitionDeviceState(state, success ? HostState.Running : HostState.Stopped); + + try { await Task.Delay(_options.Probe.Interval, ct).ConfigureAwait(false); } + catch (OperationCanceledException) { break; } + } + + try { probeRuntime?.Dispose(); } catch { } + } + + private void TransitionDeviceState(DeviceState state, HostState newState) + { + HostState old; + lock (state.ProbeLock) + { + old = state.HostState; + if (old == newState) return; + state.HostState = newState; + state.HostStateChangedUtc = DateTime.UtcNow; + } + OnHostStatusChanged?.Invoke(this, + new HostStatusChangedEventArgs(state.Options.HostAddress, old, newState)); + } + + // ---- IPerCallHostResolver ---- + + /// + /// Resolve the device host address for a given tag full-reference. Per plan decision #144 + /// the Phase 6.1 resilience pipeline keys its bulkhead + breaker on + /// (DriverInstanceId, hostName) so multi-PLC drivers get per-device isolation — + /// one dead PLC trips only its own breaker. Unknown references fall back to the + /// first configured device's host address rather than throwing — the invoker handles the + /// mislookup at the capability level when the actual read returns BadNodeIdUnknown. + /// + public string ResolveHost(string fullReference) + { + if (_tagsByName.TryGetValue(fullReference, out var def)) + return def.DeviceHostAddress; + return _options.Devices.FirstOrDefault()?.HostAddress ?? DriverInstanceId; + } + // ---- IReadable ---- /// @@ -457,6 +556,12 @@ public sealed class AbCipDriver : IDriver, IReadable, IWritable, ITagDiscovery, public AbCipDeviceOptions Options { get; } = options; public AbCipPlcFamilyProfile Profile { get; } = profile; + public object ProbeLock { get; } = new(); + public HostState HostState { get; set; } = HostState.Unknown; + public DateTime HostStateChangedUtc { get; set; } = DateTime.UtcNow; + public CancellationTokenSource? ProbeCts { get; set; } + public bool ProbeInitialized { get; set; } + public Dictionary TagHandles { get; } = new(StringComparer.OrdinalIgnoreCase); diff --git a/tests/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs b/tests/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs new file mode 100644 index 0000000..7a1f8a0 --- /dev/null +++ b/tests/ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests/AbCipHostProbeTests.cs @@ -0,0 +1,227 @@ +using System.Collections.Concurrent; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.AbCip; + +namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip.Tests; + +[Trait("Category", "Unit")] +public sealed class AbCipHostProbeTests +{ + [Fact] + public async Task GetHostStatuses_returns_one_entry_per_device() + { + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = + [ + new AbCipDeviceOptions("ab://10.0.0.5/1,0"), + new AbCipDeviceOptions("ab://10.0.0.6/1,0"), + ], + Probe = new AbCipProbeOptions { Enabled = false }, + }, "drv-1"); + await drv.InitializeAsync("{}", CancellationToken.None); + + var statuses = drv.GetHostStatuses(); + statuses.Count.ShouldBe(2); + statuses.Select(s => s.HostName).ShouldBe(["ab://10.0.0.5/1,0", "ab://10.0.0.6/1,0"], ignoreOrder: true); + statuses.ShouldAllBe(s => s.State == HostState.Unknown); + } + + [Fact] + public async Task Probe_with_successful_read_transitions_to_Running() + { + var factory = new FakeAbCipTagFactory { Customise = p => new FakeAbCipTag(p) { Status = 0 } }; + var transitions = new ConcurrentQueue(); + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(100), + Timeout = TimeSpan.FromMilliseconds(50), + ProbeTagPath = "@raw_cpu_type", + }, + }, "drv-1", factory); + drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); + + await drv.InitializeAsync("{}", CancellationToken.None); + await WaitForAsync(() => transitions.Any(t => t.NewState == HostState.Running), TimeSpan.FromSeconds(2)); + + transitions.Select(t => t.NewState).ShouldContain(HostState.Running); + drv.GetHostStatuses().Single().State.ShouldBe(HostState.Running); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Probe_with_read_failure_transitions_to_Stopped() + { + var factory = new FakeAbCipTagFactory + { + Customise = p => new FakeAbCipTag(p) { ThrowOnRead = true }, + }; + var transitions = new ConcurrentQueue(); + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions + { + Enabled = true, + Interval = TimeSpan.FromMilliseconds(100), + Timeout = TimeSpan.FromMilliseconds(50), + ProbeTagPath = "@raw_cpu_type", + }, + }, "drv-1", factory); + drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); + + await drv.InitializeAsync("{}", CancellationToken.None); + await WaitForAsync(() => transitions.Any(t => t.NewState == HostState.Stopped), TimeSpan.FromSeconds(2)); + + drv.GetHostStatuses().Single().State.ShouldBe(HostState.Stopped); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Probe_disabled_when_Enabled_is_false() + { + var factory = new FakeAbCipTagFactory(); + var transitions = new ConcurrentQueue(); + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions { Enabled = false, ProbeTagPath = "@raw_cpu_type" }, + }, "drv-1", factory); + drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); + + await drv.InitializeAsync("{}", CancellationToken.None); + await Task.Delay(300); + + transitions.ShouldBeEmpty(); + drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Probe_skipped_when_ProbeTagPath_is_null() + { + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions { Enabled = true, ProbeTagPath = null }, + }, "drv-1"); + + await drv.InitializeAsync("{}", CancellationToken.None); + await Task.Delay(200); + + drv.GetHostStatuses().Single().State.ShouldBe(HostState.Unknown); + await drv.ShutdownAsync(CancellationToken.None); + } + + [Fact] + public async Task Probe_loops_across_multiple_devices_independently() + { + var factory = new FakeAbCipTagFactory + { + // Device A returns ok, Device B throws on read. + Customise = p => p.Gateway == "10.0.0.5" + ? new FakeAbCipTag(p) + : new FakeAbCipTag(p) { ThrowOnRead = true }, + }; + var transitions = new ConcurrentQueue(); + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = + [ + new AbCipDeviceOptions("ab://10.0.0.5/1,0"), + new AbCipDeviceOptions("ab://10.0.0.6/1,0"), + ], + Probe = new AbCipProbeOptions + { + Enabled = true, Interval = TimeSpan.FromMilliseconds(100), + Timeout = TimeSpan.FromMilliseconds(50), ProbeTagPath = "@raw_cpu_type", + }, + }, "drv-1", factory); + drv.OnHostStatusChanged += (_, e) => transitions.Enqueue(e); + + await drv.InitializeAsync("{}", CancellationToken.None); + await WaitForAsync(() => transitions.Count >= 2, TimeSpan.FromSeconds(3)); + + transitions.ShouldContain(t => t.HostName == "ab://10.0.0.5/1,0" && t.NewState == HostState.Running); + transitions.ShouldContain(t => t.HostName == "ab://10.0.0.6/1,0" && t.NewState == HostState.Stopped); + await drv.ShutdownAsync(CancellationToken.None); + } + + // ---- IPerCallHostResolver ---- + + [Fact] + public async Task ResolveHost_returns_declared_device_for_known_tag() + { + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = + [ + new AbCipDeviceOptions("ab://10.0.0.5/1,0"), + new AbCipDeviceOptions("ab://10.0.0.6/1,0"), + ], + Tags = + [ + new AbCipTagDefinition("A", "ab://10.0.0.5/1,0", "A", AbCipDataType.DInt), + new AbCipTagDefinition("B", "ab://10.0.0.6/1,0", "B", AbCipDataType.DInt), + ], + Probe = new AbCipProbeOptions { Enabled = false }, + }, "drv-1"); + await drv.InitializeAsync("{}", CancellationToken.None); + + drv.ResolveHost("A").ShouldBe("ab://10.0.0.5/1,0"); + drv.ResolveHost("B").ShouldBe("ab://10.0.0.6/1,0"); + } + + [Fact] + public async Task ResolveHost_falls_back_to_first_device_for_unknown_reference() + { + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.5/1,0")], + Probe = new AbCipProbeOptions { Enabled = false }, + }, "drv-1"); + await drv.InitializeAsync("{}", CancellationToken.None); + + drv.ResolveHost("does-not-exist").ShouldBe("ab://10.0.0.5/1,0"); + } + + [Fact] + public async Task ResolveHost_falls_back_to_DriverInstanceId_when_no_devices() + { + var drv = new AbCipDriver(new AbCipDriverOptions(), "drv-1"); + await drv.InitializeAsync("{}", CancellationToken.None); + + drv.ResolveHost("anything").ShouldBe("drv-1"); + } + + [Fact] + public async Task ResolveHost_for_UDT_member_walks_to_synthesised_definition() + { + var drv = new AbCipDriver(new AbCipDriverOptions + { + Devices = [new AbCipDeviceOptions("ab://10.0.0.7/1,0")], + Tags = + [ + new AbCipTagDefinition("Motor1", "ab://10.0.0.7/1,0", "Motor1", AbCipDataType.Structure, + Members: [new AbCipStructureMember("Speed", AbCipDataType.DInt)]), + ], + Probe = new AbCipProbeOptions { Enabled = false }, + }, "drv-1"); + await drv.InitializeAsync("{}", CancellationToken.None); + + drv.ResolveHost("Motor1.Speed").ShouldBe("ab://10.0.0.7/1,0"); + } + + private static async Task WaitForAsync(Func condition, TimeSpan timeout) + { + var deadline = DateTime.UtcNow + timeout; + while (!condition() && DateTime.UtcNow < deadline) + await Task.Delay(20); + } +}