Compare commits
2 Commits
abcip-pr7-
...
abcip-pr8-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ac14ba9664 | ||
| 5978ea002d |
@@ -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
|
||||
/// <see cref="PlcTagHandle"/> and reconnects each device.</para>
|
||||
/// </remarks>
|
||||
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<DataChangeEventArgs>? OnDataChange;
|
||||
public event EventHandler<HostStatusChangedEventArgs>? 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<HostConnectivityStatus> 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 ----
|
||||
|
||||
/// <summary>
|
||||
/// 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
|
||||
/// <c>(DriverInstanceId, hostName)</c> 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.
|
||||
/// </summary>
|
||||
public string ResolveHost(string fullReference)
|
||||
{
|
||||
if (_tagsByName.TryGetValue(fullReference, out var def))
|
||||
return def.DeviceHostAddress;
|
||||
return _options.Devices.FirstOrDefault()?.HostAddress ?? DriverInstanceId;
|
||||
}
|
||||
|
||||
// ---- IReadable ----
|
||||
|
||||
/// <summary>
|
||||
@@ -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<string, PlcTagHandle> TagHandles { get; } =
|
||||
new(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
|
||||
@@ -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<HostStatusChangedEventArgs>();
|
||||
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<HostStatusChangedEventArgs>();
|
||||
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<HostStatusChangedEventArgs>();
|
||||
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<HostStatusChangedEventArgs>();
|
||||
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<bool> condition, TimeSpan timeout)
|
||||
{
|
||||
var deadline = DateTime.UtcNow + timeout;
|
||||
while (!condition() && DateTime.UtcNow < deadline)
|
||||
await Task.Delay(20);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user