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..d871c13 --- /dev/null +++ b/src/CanKit.Pro.RawCan/CanBusServiceExtensions.cs @@ -0,0 +1,116 @@ +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)); + + 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, ICanBusService service, Action onNext) + { + _subscription = subscription; + _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)) + { + // 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); + } + catch (Exception ex) + { + if (service is CanBusService concrete) + concrete.RaiseBackgroundException(ex); + } + + if (Volatile.Read(ref _disposed) != 0) break; + } + }); + } + + 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(); + 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. + 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..db783e3 100644 --- a/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs +++ b/tests/CanKit.Pro.Tests/TestCases/RawCanSubscriptionTests.cs @@ -113,6 +113,194 @@ 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); + + 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; + 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); + } + + // 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)); + } + + // 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] + 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()