diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs index a5443512..a9754706 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs @@ -41,12 +41,14 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable private readonly Task _predecessorClosed; private readonly byte[] _route; - // Never disposed: callers may still be queued on it when DisposeAsync runs, and Release on a - // disposed SemaphoreSlim throws. It owns no unmanaged handle unless AvailableWaitHandle is touched. - private readonly SemaphoreSlim _state = new(1, 1); + private readonly object _sync = new(); + // Cancels only the waits BEFORE an attempt (predecessor close, throttle interval), never an attempt + // already on the wire. Never disposed: it has no timer, and late registrations must stay legal. + private readonly CancellationTokenSource _disposing = new(); private readonly ITimer? _keepAlive; - private Open? _open; // written under _state; read lock-free via Volatile - private int _disposed; // checked under _state before any open; read lock-free elsewhere + private Task? _pendingOpen; // the one shared open in progress; under _sync + private Open? _open; // written under _sync (Volatile); read lock-free via Volatile + private int _disposed; // set under _sync; read lock-free via Volatile private int _disposeStarted; private long _lastSendTimestamp; private int _keepAlivesSent; @@ -58,7 +60,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable /// Clock for the throttle, deadlines and keep-alive. /// Completes when the previous connection for the same key has /// finished closing, so two connections to one device never overlap. - public CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time, + internal CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time, Task? predecessorClosed = null) { _key = key ?? throw new ArgumentNullException(nameof(key)); @@ -100,37 +102,68 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable /// Returns the live (client, connection) pair, opening one if needed. The pair is returned /// rather than re-read from fields afterwards, so a concurrent dispose can never hand the caller a /// null (finding 16). - private async Task EnsureOpenAsync(CancellationToken ct) + /// All callers share ONE open, which runs under its own deadline only. A caller's token (or the + /// scheduler's stop token) ends that caller's wait, never the open itself: cancelling a Forward + /// Open mid-flight could leave a connection the target accepted but nobody ever closes. A completed + /// open is stored even if disposal began meanwhile, so Forward Closes it. + private Task EnsureOpenAsync(CancellationToken ct) { - if (Volatile.Read(ref _open) is { Client.IsConnected: true } live) return live; + if (Volatile.Read(ref _open) is { Client.IsConnected: true } live) return Task.FromResult(live); + Task pending; + TaskCompletionSource? started = null; + lock (_sync) + { + ThrowIfDisposed(); + if (_open is { Client.IsConnected: true } current) return Task.FromResult(current); + if (_pendingOpen is null) + { + started = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _pendingOpen = started.Task; + } + pending = _pendingOpen; + } + if (started is not null) _ = RunOpenAsync(started); + return pending.WaitAsync(ct); + } + + /// The shared open. Bounded by (ruling R9-3); its outcome, success or + /// failure, goes to every caller waiting on it. + private async Task RunOpenAsync(TaskCompletionSource result) + { using var deadline = new CancellationTokenSource(OpenBudget, _time); - using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token); + using var beforeAttempt = CancellationTokenSource.CreateLinkedTokenSource(deadline.Token, _disposing.Token); + Open? opened = null; + Exception? error = null; try { - await _state.WaitAsync(linked.Token).ConfigureAwait(false); - try + Open? stale; + lock (_sync) { - ThrowIfDisposed(); - var current = _open; - if (current is { Client.IsConnected: true }) return current; - if (current is not null) - { - Volatile.Write(ref _open, null); - await current.Client.DisposeAsync().ConfigureAwait(false); // faulted: closes without UnRegister - } - await _predecessorClosed.WaitAsync(linked.Token).ConfigureAwait(false); - var opened = await _throttle.RunAsync(OpenAsync, linked.Token).ConfigureAwait(false); - Volatile.Write(ref _open, opened); - return opened; + stale = _open; + Volatile.Write(ref _open, null); } - finally { _state.Release(); } + if (stale is not null) await stale.Client.DisposeAsync().ConfigureAwait(false); // faulted: no UnRegister + await _predecessorClosed.WaitAsync(beforeAttempt.Token).ConfigureAwait(false); + opened = await _throttle.RunAsync(_ => OpenAsync(deadline.Token), beforeAttempt.Token).ConfigureAwait(false); } - catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested) + catch (Exception ex) { - throw new AbTransportException(AbStatus.Timeout, - $"Opening {_key.Host}:{_key.Port} did not complete within {OpenBudget.TotalMilliseconds:F0} ms."); + error = Volatile.Read(ref _disposed) != 0 + ? new AbTransportException(AbStatus.Abort, "Connection disposed.", ex) + : ex is OperationCanceledException && deadline.IsCancellationRequested + ? new AbTransportException(AbStatus.Timeout, + $"Opening {_key.Host}:{_key.Port} did not complete within {OpenBudget.TotalMilliseconds:F0} ms.", ex) + : ex; } + + lock (_sync) + { + if (opened is not null) Volatile.Write(ref _open, opened); + _pendingOpen = null; + } + if (opened is not null) result.SetResult(opened); + else result.SetException(error!); } private async Task OpenAsync(CancellationToken ct) @@ -252,18 +285,28 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable { if (Interlocked.Exchange(ref _disposeStarted, 1) != 0) return; - // Flag first: every caller still queued on _state sees it the moment it gets the lock and throws - // Abort without opening, so this waits behind at most the one open already in progress (bounded - // by OpenBudget), never behind a queue of them. - Volatile.Write(ref _disposed, 1); - await _state.WaitAsync().ConfigureAwait(false); + // Under the lock, so no caller can start a new open after this point. Then cancel the shared + // open's pre-attempt waits; an attempt already on the wire is allowed to finish (bounded by + // OpenBudget) so that whatever it opened is Forward Closed below. + Task? pending; + lock (_sync) + { + Volatile.Write(ref _disposed, 1); + pending = _pendingOpen; + } + await _disposing.CancelAsync().ConfigureAwait(false); + if (pending is not null) + { + try { await pending.ConfigureAwait(false); } + catch (Exception) { /* its callers have the error; nothing was opened */ } + } + Open? open; - try + lock (_sync) { open = _open; Volatile.Write(ref _open, null); } - finally { _state.Release(); } if (_keepAlive is not null) await _keepAlive.DisposeAsync().ConfigureAwait(false); if (open is null) return; @@ -272,7 +315,7 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable try { var budget = TimeSpan.FromSeconds(1); - using var cts = new CancellationTokenSource(budget); + using var cts = new CancellationTokenSource(budget, _time); await open.Client.SendRRDataAsync( ConnectionManagerServices.ForwardClose(open.ConnectionSerial, _o.VendorId, _o.OriginatorSerial, _route), budget, cts.Token).ConfigureAwait(false); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs index f0498555..9756d613 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs @@ -1,12 +1,25 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; /// -/// Identity of one physical CIP connection: exactly one exists per key -/// (press safety). and compare case-insensitively; -/// and compare exactly. +/// Identity of one physical CIP connection: exactly one (and one Forward Open +/// throttle) exists per key (press safety). and compare +/// case-insensitively; compares exactly; is canonicalised (all +/// whitespace removed, so "1,0" and " 1, 0 " are the same device) and then compares exactly. /// public sealed record DeviceKey(string Host, int Port, string CipPath, string PlcAttribute) { + private readonly string _cipPath = Canonicalise(CipPath); + + /// The route, whitespace-free. Canonicalised on construction and on with. + public string CipPath + { + get => _cipPath; + init => _cipPath = Canonicalise(value); + } + + private static string Canonicalise(string? path) => + path is null ? "" : string.Concat(path.Where(c => !char.IsWhiteSpace(c))); + public bool Equals(DeviceKey? other) => other is not null && string.Equals(Host, other.Host, StringComparison.OrdinalIgnoreCase) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs index f25b24a5..778f1995 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs @@ -12,9 +12,11 @@ public sealed class ForwardOpenThrottle private readonly SemaphoreSlim _inFlight = new(1, 1); private readonly TimeSpan _minInterval; private readonly TimeProvider _time; - private DateTimeOffset? _lastStart; // only touched while holding _inFlight + private long? _lastStartTimestamp; // monotonic (TimeProvider.GetTimestamp); only touched while holding _inFlight - public ForwardOpenThrottle(TimeSpan minInterval, TimeProvider time) + /// Internal: the only legitimate owner is , which keeps exactly + /// one per device. A second throttle for the same device would halve the guaranteed interval. + internal ForwardOpenThrottle(TimeSpan minInterval, TimeProvider time) { if (minInterval < TimeSpan.Zero || minInterval == Timeout.InfiniteTimeSpan) throw new ArgumentOutOfRangeException(nameof(minInterval), minInterval, "Interval must be finite and non-negative."); @@ -34,9 +36,10 @@ public sealed class ForwardOpenThrottle try { - if (_lastStart is { } last) + if (_lastStartTimestamp is { } last) { - var wait = last + _minInterval - _time.GetUtcNow(); + // Monotonic: a wall-clock step (NTP, DST, an operator) can never shorten the interval. + var wait = _minInterval - _time.GetElapsedTime(last); if (wait > TimeSpan.Zero) { try { await Task.Delay(wait, _time, ct).ConfigureAwait(false); } @@ -44,7 +47,7 @@ public sealed class ForwardOpenThrottle { throw Throttled("the minimum interval had not elapsed"); } } } - _lastStart = _time.GetUtcNow(); + _lastStartTimestamp = _time.GetTimestamp(); return await open(ct).ConfigureAwait(false); } finally { _inFlight.Release(); } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs index b254f80c..21cfd740 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs @@ -16,9 +16,12 @@ public sealed class DeviceSessionPool /// The process-wide pool the drivers use. Tests build their own. public static DeviceSessionPool Shared { get; } = new(); - internal sealed class Entry(CipConnection connection, RequestScheduler scheduler, ConnectionOptions options) + internal sealed class Entry(CipConnection connection, RequestScheduler scheduler, ConnectionOptions options, + Task predecessorClosed) { public CipConnection Connection { get; } = connection; + /// Completes when every earlier connection for this key has finished closing. + public Task PredecessorClosed { get; } = predecessorClosed; public RequestScheduler Scheduler { get; } = scheduler; public ConnectionOptions Options { get; } = options; public int RefCount; // under the pool lock @@ -30,7 +33,7 @@ public sealed class DeviceSessionPool private readonly Dictionary _throttles = new(); private readonly Dictionary _closing = new(); private readonly TimeProvider _time; - private readonly ILogger _logger; + private volatile ILogger _logger; public DeviceSessionPool(TimeProvider? time = null, ILogger? logger = null) { @@ -38,6 +41,14 @@ public sealed class DeviceSessionPool _logger = logger ?? NullLogger.Instance; } + /// Give a logger (it is built before any host exists), so its + /// options-mismatch and close-failure warnings reach the service log. Call once at startup. + public static void ConfigureSharedLogger(ILoggerFactory loggerFactory) + { + ArgumentNullException.ThrowIfNull(loggerFactory); + Shared._logger = loggerFactory.CreateLogger(); + } + /// Test hook: keys with a live entry. internal int EntryCount { get { lock (_lock) return _entries.Count; } } @@ -54,9 +65,9 @@ public sealed class DeviceSessionPool { if (!_throttles.TryGetValue(key, out var throttle)) _throttles[key] = throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, _time); - var predecessor = _closing.GetValueOrDefault(key); + var predecessor = _closing.GetValueOrDefault(key) ?? Task.CompletedTask; var connection = new CipConnection(key, options, throttle, _time, predecessor); - entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options); + entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options, predecessor); _entries[key] = entry; } else if (!entry.Options.Equals(options) && !entry.MismatchLogged) @@ -75,27 +86,45 @@ public sealed class DeviceSessionPool internal async ValueTask ReleaseAsync(DeviceKey key, Entry entry) { TaskCompletionSource closed; + Task chain; lock (_lock) { if (--entry.RefCount > 0) return; _entries.Remove(key); closed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); - _closing[key] = closed.Task; + // The successor must wait for THIS close and for every earlier one still running. An entry + // that never opened closes instantly; without the chain it would hide a slow predecessor + // (A closing, B acquired+released unopened, C opens while A's Forward Close is still pending). + chain = Task.WhenAll(entry.PredecessorClosed, closed.Task); + _closing[key] = chain; } - - // Outside the lock: scheduler first (fails its waiters, stops sending), then Forward Close + - // UnRegisterSession via the connection. - try - { - await entry.Scheduler.DisposeAsync().ConfigureAwait(false); - await entry.Connection.DisposeAsync().ConfigureAwait(false); - } - finally + _ = chain.ContinueWith(_ => { lock (_lock) { - if (_closing.TryGetValue(key, out var t) && t == closed.Task) _closing.Remove(key); + if (_closing.TryGetValue(key, out var t) && t == chain) _closing.Remove(key); } + }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); + + // Outside the lock: scheduler first (fails its waiters, stops sending), then Forward Close + + // UnRegisterSession via the connection. The connection is disposed even if the scheduler's + // dispose throws: otherwise its keep-alive would hold the press connection open forever. + try + { + try { await entry.Scheduler.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) + { + _logger.LogWarning(ex, "Disposing the request scheduler for {Host}:{Port} failed; closing the connection anyway.", + key.Host, key.Port); + } + try { await entry.Connection.DisposeAsync().ConfigureAwait(false); } + catch (Exception ex) + { + _logger.LogWarning(ex, "Closing the connection to {Host}:{Port} failed.", key.Host, key.Port); + } + } + finally + { closed.TrySetResult(); // never faults: a successor only needs to know the close is over } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs index bf2124de..6a4d3d74 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs @@ -106,6 +106,23 @@ public sealed class ForwardOpenThrottleTests invoked.ShouldBeFalse(); } + [Fact] + public async Task A_wall_clock_step_does_not_shorten_the_interval() + { + var time = new ManualTimeProvider(); + var throttle = new ForwardOpenThrottle(Second, time); + await throttle.RunAsync(_ => Task.FromResult(1), TestCt); + + time.WallClockOffset = TimeSpan.FromHours(1); // NTP / DST / operator steps the clock forward + var started = 0; + var second = throttle.RunAsync(_ => { Interlocked.Increment(ref started); return Task.FromResult(2); }, TestCt); + + await time.WaitForTimerDueAtAsync(time.Start + Second, Guard); // still parked for the full interval + Volatile.Read(ref started).ShouldBe(0); + time.Advance(Second); + (await second.WaitAsync(Guard, TestCt)).ShouldBe(2); + } + [Fact] public async Task Refusing_target_sees_at_most_one_forward_open_per_interval() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs index 065ed32c..d9cdb478 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs @@ -153,6 +153,109 @@ public sealed class DeviceSessionPoolTests schedulers.Keys.GroupBy(k => k.Item2).ShouldAllBe(g => g.Select(k => k.Item1).Distinct().Count() == 1); } + [Fact] + public async Task Device_keys_ignore_whitespace_in_the_cip_path() + { + var a = new DeviceKey("plc1", 44818, "1,0", "ControlLogix"); + var b = new DeviceKey("plc1", 44818, " 1, 0 ", "ControlLogix"); + b.CipPath.ShouldBe("1,0"); + b.ShouldBe(a); + b.GetHashCode().ShouldBe(a.GetHashCode()); + (a with { CipPath = "1, 2" }).CipPath.ShouldBe("1,2"); + + var pool = new DeviceSessionPool(new ManualTimeProvider()); + await using var la = pool.Acquire(a, Options()); + await using var lb = pool.Acquire(b, Options()); + lb.Scheduler.ShouldBeSameAs(la.Scheduler); // one connection, one throttle + pool.EntryCount.ShouldBe(1); + } + + [Fact] + public async Task A_new_connection_waits_for_every_earlier_close_not_just_the_latest() + { + // A is closing (its Forward Close is held); B is acquired and released without ever opening, so + // its own close is instant; C must still not open until A's close has finished. + await using var target = new FakeCipTarget(); + var closeHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + target.ForwardCloseGate = () => closeHeld.Task; + var time = new ManualTimeProvider(); + var pool = new DeviceSessionPool(time); + var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"); + + var a = pool.Acquire(key, Options()); + await a.Scheduler.SendAsync(ReadTag, TestCt); + var aClosing = a.DisposeAsync().AsTask(); + await FakeCipTarget.Until(() => target.ForwardCloses == 1, Guard, "A's Forward Close"); + + await pool.Acquire(key, Options()).DisposeAsync(); // B: never opened + time.Advance(TimeSpan.FromSeconds(5)); // the throttle is no obstacle for C + + await using var c = pool.Acquire(key, Options()); + var cSend = c.Scheduler.SendAsync(ReadTag, TestCt); + await Task.Delay(300, TestCt); + target.ForwardOpens.ShouldBe(1); // C is held back behind A + cSend.IsCompleted.ShouldBeFalse(); + + closeHeld.SetResult(); + await aClosing.WaitAsync(Guard, TestCt); + (await cSend.WaitAsync(Guard, TestCt)).GeneralStatus.ShouldBe((byte)0); + target.ForwardOpens.ShouldBe(2); + } + + [Fact] + public async Task Disposing_during_an_open_still_forward_closes_what_the_target_accepted() + { + // The scheduler's stop token (and any caller's) ends only the WAIT; the Forward Open on the wire + // completes, and the connection it created is closed. + await using var target = new FakeCipTarget(); + var openHeld = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + target.ForwardOpenGate = () => openHeld.Task; + var pool = new DeviceSessionPool(new ManualTimeProvider()); + var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"); + + var lease = pool.Acquire(key, Options()); + var send = lease.Scheduler.SendAsync(ReadTag, TestCt); + await FakeCipTarget.Until(() => target.ForwardOpens == 1, Guard, "the Forward Open"); + var disposing = lease.DisposeAsync().AsTask(); + + var failed = await Should.ThrowAsync(() => send.WaitAsync(Guard, TestCt)); + failed.Status.ShouldBe(AbStatus.BadConnection); // the scheduler's waiter is released at once + target.ForwardCloses.ShouldBe(0); + + openHeld.SetResult(); // the target accepts, late + await disposing.WaitAsync(Guard, TestCt); + + target.ForwardCloses.ShouldBe(1); + target.ForwardCloseRequests.Single().AsSpan(8, 8).ToArray() + .ShouldBe(target.ForwardOpenRequests.Single().AsSpan(16, 8).ToArray()); + target.ConnectedRequests.ShouldBe(0); + } + + [Fact] + public async Task The_shared_pool_logger_can_be_configured() + { + var logger = new CapturingLogger(); + DeviceSessionPool.ConfigureSharedLogger(new SingleLoggerFactory(logger)); + try + { + var key = new DeviceKey($"shared-logger-{Guid.NewGuid():N}", 44818, "1,0", "ControlLogix"); + await using var a = DeviceSessionPool.Shared.Acquire(key, Options(4002)); + await using var b = DeviceSessionPool.Shared.Acquire(key, Options(504)); + logger.Warnings.ShouldBe(1); + } + finally + { + DeviceSessionPool.ConfigureSharedLogger(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance); + } + } + + private sealed class SingleLoggerFactory(ILogger logger) : ILoggerFactory + { + public ILogger CreateLogger(string categoryName) => logger; + public void AddProvider(ILoggerProvider provider) { } + public void Dispose() { } + } + [Fact] public void Shared_is_one_process_wide_instance() => DeviceSessionPool.Shared.ShouldBeSameAs(DeviceSessionPool.Shared); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs index 0ce9f4eb..4341a66c 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs @@ -37,6 +37,12 @@ internal sealed class FakeCipTarget : IAsyncDisposable /// Connected requests only: may delay, or return null to never answer. Defaults to . public Func>? ConnectedHandler { get; set; } + /// Awaited before a Forward Open is answered (the request is already counted). + public Func? ForwardOpenGate { get; set; } + + /// Awaited before a Forward Close is answered (the request is already counted). + public Func? ForwardCloseGate { get; set; } + public int ForwardOpens { get { lock (_forwardOpens) return _forwardOpens.Count; } } public int ForwardCloses { get { lock (_forwardCloses) return _forwardCloses.Count; } } public int UnconnectedSends { get { lock (_unconnectedEmbedded) return _unconnectedEmbedded.Count; } } @@ -88,11 +94,13 @@ internal sealed class FakeCipTarget : IAsyncDisposable if (IsForwardOpen(cip)) { lock (_forwardOpens) _forwardOpens.Add(cip); + if (ForwardOpenGate is { } openGate) await openGate(); return FakeEipServer.RRReply(frame, ForwardOpenReply(cip)); } if (IsForwardClose(cip)) { lock (_forwardCloses) _forwardCloses.Add(cip); + if (ForwardCloseGate is { } closeGate) await closeGate(); var w = new CipWriter().U8(0xCE).U8(0).U8(0).U8(0).Bytes(cip.AsSpan(8, 8)).U8(0).U8(0); return FakeEipServer.RRReply(frame, w.ToArray()); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs index 49b5ee77..f7c940c7 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs @@ -15,9 +15,13 @@ internal sealed class ManualTimeProvider : TimeProvider public DateTimeOffset Start { get; } - public override DateTimeOffset GetUtcNow() { lock (_lock) return _now; } + /// Steps the wall clock () without moving monotonic time + /// () or any timer, as an NTP or operator clock change does. + public TimeSpan WallClockOffset { get; set; } - public override long GetTimestamp() => GetUtcNow().UtcTicks; + public override DateTimeOffset GetUtcNow() { lock (_lock) return _now + WallClockOffset; } + + public override long GetTimestamp() { lock (_lock) return _now.UtcTicks; } public override long TimestampFrequency => TimeSpan.TicksPerSecond;