TwinCAT follow-up — Native ADS notifications for ISubscribable. Closes task #189 — upgrades TwinCATDriver's subscription path from polling (shared PollGroupEngine) to native AdsClient.AddDeviceNotificationExAsync so the PLC pushes changes on its own cycle rather than the driver polling. Strictly better for latency + CPU — TC2 and TC3 runtimes notify on value change with sub-millisecond latency from the PLC cycle. ITwinCATClient gains AddNotificationAsync — takes symbolPath + TwinCATDataType + optional bitIndex + cycleTime + onChange callback + CancellationToken; returns an ITwinCATNotificationHandle whose Dispose tears the notification down on the wire. Bit-within-word reads supported — the parent word value arrives via the notification, driver extracts the bit before invoking the callback (same ExtractBit path as the read surface from PR 2). AdsTwinCATClient — subscribes to AdsClient.AdsNotificationEx in the ctor, maintains a ConcurrentDictionary<uint, NotificationRegistration> keyed on the server-side notification handle. AddDeviceNotificationExAsync returns Task<ResultHandle> with Handle + ErrorCode; non-NoError throws InvalidOperationException so the driver can catch + retry. Notification event args carry Handle + Value + DataType; lookup in _notifications dict routes the value through any bit-extraction + calls the consumer callback. Consumer-side exceptions are swallowed so a misbehaving callback can't crash the ADS notification thread. Dispose unsubscribes from AdsNotificationEx + clears the dict + disposes AdsClient. NotificationRegistration is ITwinCATNotificationHandle — Dispose fires DeleteDeviceNotificationAsync as fire-and-forget with CancellationToken.None (caller has already committed to teardown; blocking would slow shutdown). TwinCATDriverOptions.UseNativeNotifications — new bool, default true. When true the driver uses native notifications; when false it falls through to the shared PollGroupEngine (same semantics as other libplctag-backed drivers, also a safety valve for targets with notification limits). TwinCATDriver.SubscribeAsync dual-path — if UseNativeNotifications false delegate into _poll.Subscribe (unchanged behavior from PR 3). If true, iterate fullReferences, resolve each to its device's client via EnsureConnectedAsync (reuses PR 2's per-device connection cache), parse the SymbolPath via TwinCATSymbolPath (preserves bit-in-word support), call ITwinCATClient.AddNotificationAsync with a closure over the FullReference (not the ADS symbol — OPC UA subscribers addressed the driver-side name). Per-registration callback bridges (_, value) → OnDataChange event with a fresh DataValueSnapshot (Good status, current UtcNow timestamps). Any mid-registration failure triggers a try/catch that disposes every already-registered handle before rethrowing, keeping the driver in a clean never-existed state rather than half-registered. UnsubscribeAsync dispatches on handle type — NativeSubscriptionHandle disposes all its cached ITwinCATNotificationHandles; anything else delegates to _poll.Unsubscribe for the poll fallback. ShutdownAsync tears down native subs first (so AdsClient-level cleanup happens before the client itself disposes), then PollGroupEngine, then per-device probe CTS + client. NativeSubscriptionHandle DiagnosticId prefixes with twincat-native-sub- so Admin UI + logs can distinguish the paths. 9 new unit tests in TwinCATNativeNotificationTests — native subscribe registers one notification per tag, pushed value via FireNotification fires OnDataChange with the right FullReference (driver-side, not ADS symbol), unsubscribe disposes all notifications, unsubscribe halts future notifications, partial-failure cleanup via FailAfterNAddsFake (first succeeds, second throws → first gets torn down + Notifications count returns to 0 + AddCallCount=2 proving the test actually exercised both calls), shutdown disposes subscriptions, poll fallback works when UseNativeNotifications=false (no native handles created + initial-data push still fires), handle DiagnosticId distinguishes native vs poll. Existing poll-mode ISubscribable tests in TwinCATCapabilityTests updated with UseNativeNotifications=false so they continue testing the poll path specifically — both poll + native paths have test coverage now. TwinCATDriverTests got Probe.Enabled=false added because the default factory creates a real AdsClient which was flakily affected by parallel test execution sharing AMS router state. Total TwinCAT unit tests now 93/93 passing (+8 from PR 3's 85 counting the new native tests + 2 existing tests that got options tweaks). Full solution builds 0 errors; Modbus / AbCip / AbLegacy / other drivers untouched. TwinCAT driver is now feature-complete end-to-end — read / write / discover / native-subscribe / probe / host-resolve, with poll-mode as a safety valve. Unblocks closing task #120 for TwinCAT; remaining sub-task: FOCAS + task #188 (symbol-browsing — lower priority than FOCAS since real config flows still use pre-declared tags).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,3 +1,4 @@
|
||||
using System.Collections.Concurrent;
|
||||
using TwinCAT.Ads;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
|
||||
@@ -17,6 +18,12 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
|
||||
internal sealed class AdsTwinCATClient : ITwinCATClient
|
||||
{
|
||||
private readonly AdsClient _client = new();
|
||||
private readonly ConcurrentDictionary<uint, NotificationRegistration> _notifications = new();
|
||||
|
||||
public AdsTwinCATClient()
|
||||
{
|
||||
_client.AdsNotificationEx += OnAdsNotificationEx;
|
||||
}
|
||||
|
||||
public bool IsConnected => _client.IsConnected;
|
||||
|
||||
@@ -95,7 +102,79 @@ internal sealed class AdsTwinCATClient : ITwinCATClient
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose() => _client.Dispose();
|
||||
public async Task<ITwinCATNotificationHandle> AddNotificationAsync(
|
||||
string symbolPath,
|
||||
TwinCATDataType type,
|
||||
int? bitIndex,
|
||||
TimeSpan cycleTime,
|
||||
Action<string, object?> onChange,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var clrType = MapToClrType(type);
|
||||
// NotificationSettings takes cycle + max-delay in 100ns units. AdsTransMode.OnChange
|
||||
// fires when the value differs; OnCycle fires every cycle. OnChange is the right default
|
||||
// for OPC UA data-change semantics — the PLC already has the best view of "has this
|
||||
// changed" so we let it decide.
|
||||
var cycleTicks = (uint)Math.Max(1, cycleTime.Ticks / TimeSpan.TicksPerMillisecond * 10_000);
|
||||
var settings = new NotificationSettings(AdsTransMode.OnChange, (int)cycleTicks, 0);
|
||||
|
||||
// AddDeviceNotificationExAsync returns Task<ResultHandle>; AdsNotificationEx fires
|
||||
// with the handle as part of the event args so we use the handle as the correlation
|
||||
// key into _notifications.
|
||||
var result = await _client.AddDeviceNotificationExAsync(
|
||||
symbolPath, settings, userData: null, clrType, args: null, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
if (result.ErrorCode != AdsErrorCode.NoError)
|
||||
throw new InvalidOperationException(
|
||||
$"AddDeviceNotificationExAsync failed with ADS error {result.ErrorCode} for {symbolPath}");
|
||||
|
||||
var reg = new NotificationRegistration(symbolPath, type, bitIndex, onChange, this, result.Handle);
|
||||
_notifications[result.Handle] = reg;
|
||||
return reg;
|
||||
}
|
||||
|
||||
private void OnAdsNotificationEx(object? sender, AdsNotificationExEventArgs args)
|
||||
{
|
||||
if (!_notifications.TryGetValue(args.Handle, out var reg)) return;
|
||||
var value = args.Value;
|
||||
if (reg.BitIndex is int bit && reg.Type == TwinCATDataType.Bool && value is not bool)
|
||||
value = ExtractBit(value, bit);
|
||||
try { reg.OnChange(reg.SymbolPath, value); } catch { /* consumer-side errors don't crash the ADS thread */ }
|
||||
}
|
||||
|
||||
internal async Task DeleteNotificationAsync(uint handle, CancellationToken cancellationToken)
|
||||
{
|
||||
_notifications.TryRemove(handle, out _);
|
||||
try { await _client.DeleteDeviceNotificationAsync(handle, cancellationToken).ConfigureAwait(false); }
|
||||
catch { /* best-effort tear-down; target may already be gone */ }
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
_client.AdsNotificationEx -= OnAdsNotificationEx;
|
||||
_notifications.Clear();
|
||||
_client.Dispose();
|
||||
}
|
||||
|
||||
private sealed class NotificationRegistration(
|
||||
string symbolPath,
|
||||
TwinCATDataType type,
|
||||
int? bitIndex,
|
||||
Action<string, object?> onChange,
|
||||
AdsTwinCATClient owner,
|
||||
uint handle) : ITwinCATNotificationHandle
|
||||
{
|
||||
public string SymbolPath { get; } = symbolPath;
|
||||
public TwinCATDataType Type { get; } = type;
|
||||
public int? BitIndex { get; } = bitIndex;
|
||||
public Action<string, object?> OnChange { get; } = onChange;
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
// Fire-and-forget AMS call — caller has already committed to the tear-down.
|
||||
_ = owner.DeleteNotificationAsync(handle, CancellationToken.None);
|
||||
}
|
||||
}
|
||||
|
||||
private static Type MapToClrType(TwinCATDataType type) => type switch
|
||||
{
|
||||
|
||||
@@ -46,8 +46,31 @@ public interface ITwinCATClient : IDisposable
|
||||
/// Used by <see cref="Core.Abstractions.IHostConnectivityProbe"/>'s probe loop.
|
||||
/// </summary>
|
||||
Task<bool> ProbeAsync(CancellationToken cancellationToken);
|
||||
|
||||
/// <summary>
|
||||
/// Register a cyclic / on-change ADS notification for a symbol. Returns a handle whose
|
||||
/// <see cref="IDisposable.Dispose"/> tears the notification down. Callback fires on the
|
||||
/// thread libplctag / AdsClient uses for notifications — consumers should marshal to
|
||||
/// their own scheduler before doing work of any size.
|
||||
/// </summary>
|
||||
/// <param name="symbolPath">ADS symbol path (e.g. <c>MAIN.bStart</c>).</param>
|
||||
/// <param name="type">Declared type; drives the native layout + callback value boxing.</param>
|
||||
/// <param name="bitIndex">For BOOL-within-word tags — the bit to extract from the parent word.</param>
|
||||
/// <param name="cycleTime">Minimum interval between change notifications (native-floor depends on target).</param>
|
||||
/// <param name="onChange">Invoked with <c>(symbolPath, boxedValue)</c> per notification.</param>
|
||||
/// <param name="cancellationToken">Cancels the initial registration; does not tear down an established notification.</param>
|
||||
Task<ITwinCATNotificationHandle> AddNotificationAsync(
|
||||
string symbolPath,
|
||||
TwinCATDataType type,
|
||||
int? bitIndex,
|
||||
TimeSpan cycleTime,
|
||||
Action<string, object?> onChange,
|
||||
CancellationToken cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>Opaque handle for a registered ADS notification. <see cref="IDisposable.Dispose"/> tears it down.</summary>
|
||||
public interface ITwinCATNotificationHandle : IDisposable { }
|
||||
|
||||
/// <summary>Factory for <see cref="ITwinCATClient"/>s. One client per device.</summary>
|
||||
public interface ITwinCATClientFactory
|
||||
{
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.TwinCAT;
|
||||
@@ -78,6 +79,12 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
|
||||
|
||||
public async Task ShutdownAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
// Native subs first — disposing the handles is cheap + lets the client close its
|
||||
// notifications before the AdsClient itself goes away.
|
||||
foreach (var sub in _nativeSubs.Values)
|
||||
foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } }
|
||||
_nativeSubs.Clear();
|
||||
|
||||
await _poll.DisposeAsync().ConfigureAwait(false);
|
||||
foreach (var state in _devices.Values)
|
||||
{
|
||||
@@ -238,18 +245,83 @@ public sealed class TwinCATDriver : IDriver, IReadable, IWritable, ITagDiscovery
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
// ---- ISubscribable (polling overlay via shared engine) ----
|
||||
// ---- ISubscribable (native ADS notifications with poll fallback) ----
|
||||
|
||||
public Task<ISubscriptionHandle> SubscribeAsync(
|
||||
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken) =>
|
||||
Task.FromResult(_poll.Subscribe(fullReferences, publishingInterval));
|
||||
private readonly ConcurrentDictionary<long, NativeSubscription> _nativeSubs = new();
|
||||
private long _nextNativeSubId;
|
||||
|
||||
/// <summary>
|
||||
/// Subscribe via native ADS notifications when <see cref="TwinCATDriverOptions.UseNativeNotifications"/>
|
||||
/// is <c>true</c>, otherwise fall through to the shared <see cref="PollGroupEngine"/>.
|
||||
/// Native path registers one <see cref="ITwinCATNotificationHandle"/> per tag against the
|
||||
/// target's PLC runtime — the PLC pushes changes on its own cycle so we skip the poll
|
||||
/// loop entirely. Unsub path disposes the handles.
|
||||
/// </summary>
|
||||
public async Task<ISubscriptionHandle> SubscribeAsync(
|
||||
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
|
||||
{
|
||||
if (!_options.UseNativeNotifications)
|
||||
return _poll.Subscribe(fullReferences, publishingInterval);
|
||||
|
||||
var id = Interlocked.Increment(ref _nextNativeSubId);
|
||||
var handle = new NativeSubscriptionHandle(id);
|
||||
var registrations = new List<ITwinCATNotificationHandle>(fullReferences.Count);
|
||||
var now = DateTime.UtcNow;
|
||||
|
||||
try
|
||||
{
|
||||
foreach (var reference in fullReferences)
|
||||
{
|
||||
if (!_tagsByName.TryGetValue(reference, out var def)) continue;
|
||||
if (!_devices.TryGetValue(def.DeviceHostAddress, out var device)) continue;
|
||||
|
||||
var client = await EnsureConnectedAsync(device, cancellationToken).ConfigureAwait(false);
|
||||
var parsed = TwinCATSymbolPath.TryParse(def.SymbolPath);
|
||||
var symbolName = parsed?.ToAdsSymbolName() ?? def.SymbolPath;
|
||||
var bitIndex = parsed?.BitIndex;
|
||||
|
||||
var reg = await client.AddNotificationAsync(
|
||||
symbolName, def.DataType, bitIndex, publishingInterval,
|
||||
(_, value) => OnDataChange?.Invoke(this,
|
||||
new DataChangeEventArgs(handle, reference, new DataValueSnapshot(
|
||||
value, TwinCATStatusMapper.Good, DateTime.UtcNow, DateTime.UtcNow))),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
registrations.Add(reg);
|
||||
}
|
||||
}
|
||||
catch
|
||||
{
|
||||
// On any registration failure, tear down everything we got so far + rethrow. Leaves
|
||||
// the subscription in a clean "never existed" state rather than a half-registered
|
||||
// state the caller has to clean up.
|
||||
foreach (var r in registrations) { try { r.Dispose(); } catch { } }
|
||||
throw;
|
||||
}
|
||||
|
||||
_nativeSubs[id] = new NativeSubscription(handle, registrations);
|
||||
return handle;
|
||||
}
|
||||
|
||||
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
|
||||
{
|
||||
if (handle is NativeSubscriptionHandle native && _nativeSubs.TryRemove(native.Id, out var sub))
|
||||
{
|
||||
foreach (var r in sub.Registrations) { try { r.Dispose(); } catch { } }
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
_poll.Unsubscribe(handle);
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private sealed record NativeSubscriptionHandle(long Id) : ISubscriptionHandle
|
||||
{
|
||||
public string DiagnosticId => $"twincat-native-sub-{Id}";
|
||||
}
|
||||
|
||||
private sealed record NativeSubscription(
|
||||
NativeSubscriptionHandle Handle,
|
||||
IReadOnlyList<ITwinCATNotificationHandle> Registrations);
|
||||
|
||||
// ---- IHostConnectivityProbe ----
|
||||
|
||||
public IReadOnlyList<HostConnectivityStatus> GetHostStatuses() =>
|
||||
|
||||
@@ -11,6 +11,18 @@ public sealed class TwinCATDriverOptions
|
||||
public IReadOnlyList<TwinCATTagDefinition> Tags { get; init; } = [];
|
||||
public TwinCATProbeOptions Probe { get; init; } = new();
|
||||
public TimeSpan Timeout { get; init; } = TimeSpan.FromSeconds(2);
|
||||
|
||||
/// <summary>
|
||||
/// When <c>true</c> (default), <c>SubscribeAsync</c> registers native ADS notifications
|
||||
/// via <c>AddDeviceNotificationExAsync</c> — the PLC pushes changes on its own cycle
|
||||
/// rather than the driver polling. Strictly better for latency + CPU when the target
|
||||
/// supports it (TC2 + TC3 PLC runtimes always do; some soft-PLC / third-party ADS
|
||||
/// implementations may not). When <c>false</c>, the driver falls through to the shared
|
||||
/// <see cref="Core.Abstractions.PollGroupEngine"/> — same semantics as the other
|
||||
/// libplctag-backed drivers. Set <c>false</c> for deployments where the AMS router has
|
||||
/// notification limits you can't raise.
|
||||
/// </summary>
|
||||
public bool UseNativeNotifications { get; init; } = true;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user