From f3bb3e8a29e28770eeee308e5d61664a8d6c74a0 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:25:33 +0200 Subject: [PATCH 1/4] feat(rawcan): add callback-style Subscribe(onNext, predicate) convenience API A colleague asked for a way to subscribe to incoming CAN frames with a callback instead of driving the ISubscription.Frames async-enumerable manually. Rather than a bigger cross-protocol unification (the protocol layers already have their own event-based APIs -- J1939's MessageReceived, ISO-TP's DatagramReceived, CANopen's PDO/SDO events -- and unifying those under one generic signature would either duplicate them for one line of savings or flatten their typed payloads to a lowest common denominator), this adds a small, additive-only extension method scoped to the raw-CAN layer, where the underlying complexity (per-subscription bounded buffering, backpressure) is real enough to be worth hiding. CanBusServiceExtensions.Subscribe(onNext, predicate, bufferCapacity) is built entirely on the existing ISubscription pull API, so it inherits the same FR-RAW-011 guarantee for free: a slow onNext only falls behind and drops its own subscription's oldest frames -- it can never delay delivery to other subscriptions or to the bus's own FrameObserved event, because the dispatch hot path never waits on a subscriber's consumer. A throwing onNext is isolated per frame and routed through the existing BackgroundExceptionOccurred fault channel via a new small internal CanBusService.RaiseBackgroundException helper (also used to de-duplicate the identical inline try/catch OnFrameObserved already had). Four new tests cover: filtering, the slow-handler-does-not-block invariant (mirroring the existing raw-subscription FR-RAW-011 test), Dispose stopping delivery, and handler-exception routing + delivery continuing. Public API surface changed (new type in CanKit.Pro.RawCan) -- approval baseline updated in the same commit. Verified: full build + 414/414 tests green on net10.0. --- src/CanKit.Pro.RawCan/CanBusService.cs | 16 ++- .../CanBusServiceExtensions.cs | 104 +++++++++++++++ .../CanKit.Pro.RawCan.approved.txt | 2 + .../TestCases/RawCanSubscriptionTests.cs | 123 ++++++++++++++++++ 4 files changed, 243 insertions(+), 2 deletions(-) create mode 100644 src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs diff --git a/src/CanKit.Pro.RawCan/CanBusService.cs b/src/CanKit.Pro.RawCan/CanBusService.cs index 0881b27..06660bd 100644 --- a/src/CanKit.Pro.RawCan/CanBusService.cs +++ b/src/CanKit.Pro.RawCan/CanBusService.cs @@ -179,8 +179,7 @@ private void OnFrameObserved(object? sender, CanReceiveDataView e) } catch (Exception ex) { - try { BackgroundExceptionOccurred?.Invoke(this, ex); } - catch { /* a fault listener must not break dispatch either */ } + RaiseBackgroundException(ex); } } } @@ -188,6 +187,19 @@ private void OnFrameObserved(object? sender, CanReceiveDataView e) /// public event EventHandler? BackgroundExceptionOccurred; + /// + /// Routes through , + /// isolating a misbehaving listener from the caller. Internal (not part of + /// ) because events can only be raised from their declaring + /// type; exposed so can report a failing + /// callback handler through this same fault channel instead of a second, parallel one. + /// + internal void RaiseBackgroundException(Exception ex) + { + try { BackgroundExceptionOccurred?.Invoke(this, ex); } + catch { /* a fault listener must not break dispatch either */ } + } + /// public void Dispose() { diff --git a/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs new file mode 100644 index 0000000..139073b --- /dev/null +++ b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs @@ -0,0 +1,104 @@ +using System; +using System.Threading; +using System.Threading.Tasks; +using CanKit.Abstractions.API.Can.Definitions; + +namespace CanKit.Pro.RawCan +{ + /// + /// Callback-style convenience layer over + /// for callers who want "filter + handler" instead of driving the async-enumerable + /// stream themselves. + /// (对 的回调式便捷封装， + /// 面向只想要"过滤 + 处理函数"、不想自己驱动 异步流的调用方。) + /// + public static class CanBusServiceExtensions + { + /// + /// Registers a subscription and invokes for every frame it + /// accepts, on a dedicated background task. Disposing the returned handle stops delivery + /// and lets that task end. + /// (注册一路订阅，对其接收到的每一帧在专用后台任务上调用 ； + /// 释放返回的句柄会停止投递并使该任务结束。) + /// + /// + /// Built entirely on the existing pull API, so the same + /// per-subscription bounded, drop-oldest buffer applies (FR-RAW-011): a slow + /// only ever falls behind and drops its own oldest frames -- it + /// can never delay delivery to other subscriptions or to the bus's own + /// FrameObserved event, because the dispatch hot path never waits on a + /// subscriber's consumer. An exception thrown by is isolated + /// per frame -- delivery continues with the next frame -- and, when + /// is a , routed through + /// , the same fault channel every + /// other background failure in this service uses. + /// (完全基于现有的 拉取式 API 构建，因此同样适用逐订阅有界、 + /// 丢弃最旧的缓冲区（FR-RAW-011）：迟缓的 只会自己落后并丢弃自己最旧的帧—— + /// 因为分发热路径从不等待订阅方的消费者，它永远不会延迟向其他订阅或总线自身 FrameObserved + /// 事件的投递。 抛出的异常按帧隔离——投递会以下一帧继续——当 + /// 是 时，异常会经由 + /// 上抛，与本服务其余后台故障共用同一通道。) + /// + /// The service to subscribe on. + /// Invoked for each accepted frame, in arrival order. + /// Per-frame filter, or null to accept all frames. + /// + /// Bounded buffer capacity for the underlying subscription; null uses + /// . + /// + /// Disposing this stops the subscription and the background delivery task. + public static IDisposable Subscribe( + this ICanBusService service, + Action onNext, + Func? predicate = null, + int? bufferCapacity = null) + { + if (service is null) throw new ArgumentNullException(nameof(service)); + if (onNext is null) throw new ArgumentNullException(nameof(onNext)); + + var subscription = service.Subscribe(predicate, bufferCapacity); + var pumpTask = Task.Run(async () => + { + await foreach (var frame in subscription.Frames.ConfigureAwait(false)) + { + try + { + onNext(frame); + } + catch (Exception ex) + { + if (service is CanBusService concrete) + concrete.RaiseBackgroundException(ex); + } + } + }); + + return new CallbackSubscription(subscription, pumpTask); + } + + private sealed class CallbackSubscription : IDisposable + { + private readonly ISubscription _subscription; + private readonly Task _pumpTask; + private int _disposed; + + public CallbackSubscription(ISubscription subscription, Task pumpTask) + { + _subscription = subscription; + _pumpTask = pumpTask; + } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) return; // idempotent + // Completes the channel, which ends the pump task's `await foreach` gracefully + // (no exception -- see Subscription.Dispose/ReadAsync). + _subscription.Dispose(); + // Best-effort bounded join so a caller who disposes and then immediately tears + // down surrounding state doesn't race the last in-flight onNext call; matches the + // same dispose-teardown idiom used for background readers throughout this codebase. + try { _pumpTask.Wait(TimeSpan.FromSeconds(2)); } catch { /* best-effort */ } + } + } + } +} diff --git a/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt b/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt index 1af0208..6f0713f 100644 --- a/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt +++ b/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.RawCan.approved.txt @@ -10,6 +10,8 @@ class CanKit.Pro.RawCan.CanBusService method System.Void Dispose() prop CanKit.Abstractions.API.Can.ICanBus Bus {get} prop System.Int32 SubscriptionCount {get} +class CanKit.Pro.RawCan.CanBusServiceExtensions + method System.IDisposable Subscribe(CanKit.Pro.RawCan.ICanBusService service, System.Action onNext, System.Func predicate, System.Nullable bufferCapacity) struct CanKit.Pro.RawCan.CanIdFilter method CanKit.Pro.RawCan.CanIdFilter Mask(System.UInt32 accCode, System.UInt32 accMask, CanKit.Abstractions.API.Common.Definitions.CanFilterIDType idType) method CanKit.Pro.RawCan.CanIdFilter Range(System.UInt32 from, System.UInt32 to, CanKit.Abstractions.API.Common.Definitions.CanFilterIDType idType) diff --git a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs index 72e9273..e2d4fa8 100644 --- a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs @@ -113,6 +113,129 @@ public async Task Slow_Subscription_Does_Not_Block_Others_Or_The_Bus_Event() Volatile.Read(ref busEventCount).Should().Be(n); } + // Callback-style Subscribe(onNext, predicate): the handler is invoked for matching frames + // only, in arrival order. + [Fact] + public async Task Callback_Subscribe_Invokes_Handler_For_Matching_Frames_Only() + { + var session = NewSession(); + using var sender = Open(session, 0); + using var receiver = Open(session, 1); + using var service = new CanBusService(receiver); + + var received = new List(); + var lastReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var subscription = service.Subscribe( + frame => + { + lock (received) + { + received.Add(frame.ID); + if (received.Count >= 2) lastReceived.TrySetResult(true); + } + }, + predicate: f => f.ID == 0x123); + + sender.Transmit(CanFrame.Classic(0x123, new byte[] { 1 })); + sender.Transmit(CanFrame.Classic(0x456, new byte[] { 2 })); // filtered out + sender.Transmit(CanFrame.Classic(0x123, new byte[] { 3 })); + + await lastReceived.Task.WaitAsync(ShortTimeout); + lock (received) received.Should().Equal(0x123, 0x123); + } + + // Callback-style Subscribe must uphold the same FR-RAW-011 guarantee as the raw pull API: a + // slow/blocking onNext only ever falls behind its own subscription, never the bus event or a + // second, actively-draining subscription. + [Fact] + public async Task Callback_Subscribe_Slow_Handler_Does_Not_Block_Others_Or_The_Bus_Event() + { + var session = NewSession(); + using var sender = Open(session, 0); + using var receiver = Open(session, 1); + using var service = new CanBusService(receiver); + + var busEventCount = 0; + receiver.FrameObserved += (_, _) => Interlocked.Increment(ref busEventCount); + + var block = new SemaphoreSlim(0); // never released: the slow handler blocks forever + using var slow = service.Subscribe(_ => block.Wait(), bufferCapacity: 1); + + var fastCount = 0; + var allFastReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + const int n = 200; + using var fast = service.Subscribe(_ => + { + if (Interlocked.Increment(ref fastCount) >= n) allFastReceived.TrySetResult(true); + }, bufferCapacity: 512); + + for (var i = 0; i < n; i++) + sender.Transmit(CanFrame.Classic(0x300 + (i & 0x0F), new byte[] { (byte)i })); + + await allFastReceived.Task.WaitAsync(ShortTimeout); + Volatile.Read(ref busEventCount).Should().Be(n); + } + + // Disposing the callback handle stops delivery: no further onNext calls after Dispose returns. + [Fact] + public async Task Callback_Subscribe_Dispose_Stops_Delivery() + { + var session = NewSession(); + using var sender = Open(session, 0); + using var receiver = Open(session, 1); + using var service = new CanBusService(receiver); + + var count = 0; + var firstReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var subscription = service.Subscribe(_ => + { + Interlocked.Increment(ref count); + firstReceived.TrySetResult(true); + }); + + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 1 })); + await firstReceived.Task.WaitAsync(ShortTimeout); + + subscription.Dispose(); + var countAfterDispose = Volatile.Read(ref count); + + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 2 })); + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 3 })); + await Task.Delay(200); + + Volatile.Read(ref count).Should().Be(countAfterDispose); + } + + // A handler exception is isolated per frame -- delivery continues -- and surfaced via the + // service's existing fault channel, the same as a throwing predicate. + [Fact] + public async Task Callback_Subscribe_Handler_Exception_Is_Surfaced_And_Delivery_Continues() + { + var session = NewSession(); + using var sender = Open(session, 0); + using var receiver = Open(session, 1); + using var service = new CanBusService(receiver); + + var observed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + service.BackgroundExceptionOccurred += (_, ex) => observed.TrySetResult(ex); + + var secondReceived = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var calls = 0; + using var subscription = service.Subscribe(_ => + { + if (Interlocked.Increment(ref calls) == 1) + throw new InvalidOperationException("boom"); + secondReceived.TrySetResult(true); + }); + + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 1 })); + var ex = await observed.Task.WaitAsync(ShortTimeout); + ex.Should().BeOfType().Which.Message.Should().Be("boom"); + + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 2 })); + await secondReceived.Task.WaitAsync(ShortTimeout); + } + // FR-RAW-012: creating and disposing N subscriptions leaves no entries in the service registry. [Fact] public void Disposing_Subscriptions_Leaves_No_Registry_Entries() From be23113719b277a7dcf5acd246097fd0c033801d Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 9 Sep 2026 10:35:07 +0000 Subject: [PATCH 2/4] fix(rawcan): skip pump join when disposing callback Subscribe from onNext CallbackSubscription.Dispose waited on the pump task that invokes onNext, so disposing the handle from inside the callback deadlocked until the two-second timeout. Skip that join when Dispose runs on the pump, matching ProtocolActor. --- .../CanBusServiceExtensions.cs | 44 ++++++++++--------- .../TestCases/RawCanSubscriptionTests.cs | 23 ++++++++++ 2 files changed, 47 insertions(+), 20 deletions(-) diff --git a/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs index 139073b..1474e93 100644 --- a/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs +++ b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs @@ -56,36 +56,39 @@ public static IDisposable Subscribe( if (service is null) throw new ArgumentNullException(nameof(service)); if (onNext is null) throw new ArgumentNullException(nameof(onNext)); - var subscription = service.Subscribe(predicate, bufferCapacity); - var pumpTask = Task.Run(async () => - { - await foreach (var frame in subscription.Frames.ConfigureAwait(false)) - { - try - { - onNext(frame); - } - catch (Exception ex) - { - if (service is CanBusService concrete) - concrete.RaiseBackgroundException(ex); - } - } - }); - - return new CallbackSubscription(subscription, pumpTask); + return new CallbackSubscription(service.Subscribe(predicate, bufferCapacity), service, onNext); } private sealed class CallbackSubscription : IDisposable { private readonly ISubscription _subscription; private readonly Task _pumpTask; + private readonly AsyncLocal _isOnPump = new(); private int _disposed; - public CallbackSubscription(ISubscription subscription, Task pumpTask) + public CallbackSubscription(ISubscription subscription, ICanBusService service, Action onNext) { _subscription = subscription; - _pumpTask = pumpTask; + _pumpTask = Task.Run(async () => + { + // Marks the entire pump (including onNext) so Dispose can skip the join + // below when invoked from the callback itself -- waiting on this task from + // inside it would deadlock until the timeout. Same reentrancy guard as + // ProtocolActor.Dispose / _isOnLoop. + _isOnPump.Value = true; + await foreach (var frame in _subscription.Frames.ConfigureAwait(false)) + { + try + { + onNext(frame); + } + catch (Exception ex) + { + if (service is CanBusService concrete) + concrete.RaiseBackgroundException(ex); + } + } + }); } public void Dispose() @@ -94,6 +97,7 @@ public void Dispose() // Completes the channel, which ends the pump task's `await foreach` gracefully // (no exception -- see Subscription.Dispose/ReadAsync). _subscription.Dispose(); + if (_isOnPump.Value) return; // Best-effort bounded join so a caller who disposes and then immediately tears // down surrounding state doesn't race the last in-flight onNext call; matches the // same dispose-teardown idiom used for background readers throughout this codebase. diff --git a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs index e2d4fa8..89925bb 100644 --- a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs @@ -206,6 +206,29 @@ public async Task Callback_Subscribe_Dispose_Stops_Delivery() Volatile.Read(ref count).Should().Be(countAfterDispose); } + // Disposing the handle from inside onNext must not join the pump that is currently + // invoking that callback: a self-wait can only finish via the two-second timeout. + [Fact] + public async Task Callback_Subscribe_Dispose_From_Inside_Handler_Does_Not_Hang() + { + var session = NewSession(); + using var sender = Open(session, 0); + using var receiver = Open(session, 1); + using var service = new CanBusService(receiver); + + IDisposable? subscription = null; + var disposed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscription = service.Subscribe(_ => + { + subscription!.Dispose(); + disposed.TrySetResult(true); + }); + + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 1 })); + // The pump join timeout is 2s; completing well under that proves the self-wait was skipped. + await disposed.Task.WaitAsync(TimeSpan.FromMilliseconds(500)); + } + // A handler exception is isolated per frame -- delivery continues -- and surfaced via the // service's existing fault channel, the same as a throwing predicate. [Fact] From e44f3cccafc69ae37848c6c9de449a96640d8ad1 Mon Sep 17 00:00:00 2001 From: Dietmar Borgards <2646931+dborgards@users.noreply.github.com> Date: Wed, 9 Sep 2026 12:39:30 +0200 Subject: [PATCH 3/4] test(rawcan): dispose the SemaphoreSlim in the slow-handler test CodeQL flagged it (cs/local-not-disposed) on the PR. Genuine gap, not a by-design pattern like the rest of this session's CodeQL triage. --- tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs index 89925bb..56ae487 100644 --- a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs @@ -158,7 +158,7 @@ public async Task Callback_Subscribe_Slow_Handler_Does_Not_Block_Others_Or_The_B var busEventCount = 0; receiver.FrameObserved += (_, _) => Interlocked.Increment(ref busEventCount); - var block = new SemaphoreSlim(0); // never released: the slow handler blocks forever + using var block = new SemaphoreSlim(0); // never released: the slow handler blocks forever using var slow = service.Subscribe(_ => block.Wait(), bufferCapacity: 1); var fastCount = 0; From fa548741abf1fbdd948d95b9bef2ec53b4b72ee7 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Wed, 9 Sep 2026 10:47:23 +0000 Subject: [PATCH 4/4] fix(rawcan): stop callback pump after self-dispose Completing the subscription channel does not drop already-buffered frames, so disposing from onNext skipped the pump join and still delivered the remainder after Dispose returned. Stop the pump on the disposed flag instead. --- .../CanBusServiceExtensions.cs | 8 ++++ .../TestCases/RawCanSubscriptionTests.cs | 42 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs index 1474e93..d871c13 100644 --- a/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs +++ b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs @@ -78,6 +78,12 @@ public CallbackSubscription(ISubscription subscription, ICanBusService service, _isOnPump.Value = true; await foreach (var frame in _subscription.Frames.ConfigureAwait(false)) { + // Completing the channel writer (Subscription.Dispose) does not drop + // items already buffered. Without these checks a self-dispose from + // onNext would skip the pump join and still deliver every remaining + // queued frame after Dispose has returned. + if (Volatile.Read(ref _disposed) != 0) break; + try { onNext(frame); @@ -87,6 +93,8 @@ public CallbackSubscription(ISubscription subscription, ICanBusService service, if (service is CanBusService concrete) concrete.RaiseBackgroundException(ex); } + + if (Volatile.Read(ref _disposed) != 0) break; } }); } diff --git a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs index 56ae487..db783e3 100644 --- a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs @@ -229,6 +229,48 @@ public async Task Callback_Subscribe_Dispose_From_Inside_Handler_Does_Not_Hang() await disposed.Task.WaitAsync(TimeSpan.FromMilliseconds(500)); } + // Completing the channel writer does not drop items already queued. Disposing from inside + // onNext skips the pump join, so the pump must stop on the disposed flag rather than drain + // the remainder -- otherwise further onNext calls run after Dispose has returned. + [Fact] + public async Task Callback_Subscribe_Dispose_From_Inside_Handler_Does_Not_Deliver_Buffered_Frames() + { + var session = NewSession(); + using var sender = Open(session, 0); + using var receiver = Open(session, 1); + using var service = new CanBusService(receiver); + + IDisposable? subscription = null; + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var proceed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var disposed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var count = 0; + subscription = service.Subscribe(_ => + { + if (Interlocked.Increment(ref count) == 1) + { + entered.TrySetResult(true); + proceed.Task.GetAwaiter().GetResult(); + subscription!.Dispose(); + disposed.TrySetResult(true); + } + }); + + sender.Transmit(CanFrame.Classic(0x100, new byte[] { 1 })); + await entered.Task.WaitAsync(ShortTimeout); + + // Burst into the bounded buffer while onNext is blocked; these would otherwise be + // delivered after Dispose returns if the pump kept draining the completed channel. + for (var i = 0; i < 16; i++) + sender.Transmit(CanFrame.Classic(0x100, new byte[] { (byte)i })); + + proceed.TrySetResult(true); + await disposed.Task.WaitAsync(ShortTimeout); + await Task.Delay(200); + + Volatile.Read(ref count).Should().Be(1); + } + // A handler exception is isolated per frame -- delivery continues -- and surfaced via the // service's existing fault channel, the same as a throwing predicate. [Fact]