feat(rawcan): callback-style Subscribe(onNext, predicate)

src/CanKit.Pro.RawCan/CanBusService.cs

@@ -179,15 +179,27 @@ 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);
                 }
             }
         }
 
         /// <inheritdoc />
         public event EventHandler<Exception>? BackgroundExceptionOccurred;
 
+        /// <summary>
+        /// Routes <paramref name="ex"/> through <see cref="BackgroundExceptionOccurred"/>,
+        /// isolating a misbehaving listener from the caller. Internal (not part of
+        /// <see cref="ICanBusService"/>) because events can only be raised from their declaring
+        /// type; exposed so <see cref="CanBusServiceExtensions.Subscribe"/> can report a failing
+        /// callback handler through this same fault channel instead of a second, parallel one.
+        /// </summary>
+        internal void RaiseBackgroundException(Exception ex)
+        {
+            try { BackgroundExceptionOccurred?.Invoke(this, ex); }
+            catch { /* a fault listener must not break dispatch either */ }
+        }
+
         /// <inheritdoc />
         public void Dispose()
         {

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
+{
+    /// <summary>
+    /// Callback-style convenience layer over <see cref="ICanBusService.Subscribe(Func{CanFrameView,bool}?,int?)"/>
+    /// for callers who want "filter + handler" instead of driving the async-enumerable
+    /// <see cref="ISubscription.Frames"/> stream themselves.
+    /// (对 <see cref="ICanBusService.Subscribe(Func{CanFrameView,bool}?,int?)"/> 的回调式便捷封装，
+    /// 面向只想要"过滤 + 处理函数"、不想自己驱动 <see cref="ISubscription.Frames"/> 异步流的调用方。)
+    /// </summary>
+    public static class CanBusServiceExtensions
+    {
+        /// <summary>
+        /// Registers a subscription and invokes <paramref name="onNext"/> for every frame it
+        /// accepts, on a dedicated background task. Disposing the returned handle stops delivery
+        /// and lets that task end.
+        /// (注册一路订阅，对其接收到的每一帧在专用后台任务上调用 <paramref name="onNext"/>；
+        /// 释放返回的句柄会停止投递并使该任务结束。)
+        /// </summary>
+        /// <remarks>
+        /// Built entirely on the existing <see cref="ISubscription"/> pull API, so the same
+        /// per-subscription bounded, drop-oldest buffer applies (FR-RAW-011): a slow
+        /// <paramref name="onNext"/> only ever falls behind and drops its own oldest frames -- it
+        /// can never delay delivery to other subscriptions or to the bus's own
+        /// <c>FrameObserved</c> event, because the dispatch hot path never waits on a
+        /// subscriber's consumer. An exception thrown by <paramref name="onNext"/> is isolated
+        /// per frame -- delivery continues with the next frame -- and, when <paramref name="service"/>
+        /// is a <see cref="CanBusService"/>, routed through
+        /// <see cref="ICanBusService.BackgroundExceptionOccurred"/>, the same fault channel every
+        /// other background failure in this service uses.
+        /// (完全基于现有的 <see cref="ISubscription"/> 拉取式 API 构建，因此同样适用逐订阅有界、
+        /// 丢弃最旧的缓冲区（FR-RAW-011）：迟缓的 <paramref name="onNext"/> 只会自己落后并丢弃自己最旧的帧——
+        /// 因为分发热路径从不等待订阅方的消费者，它永远不会延迟向其他订阅或总线自身 <c>FrameObserved</c>
+        /// 事件的投递。<paramref name="onNext"/> 抛出的异常按帧隔离——投递会以下一帧继续——当
+        /// <paramref name="service"/> 是 <see cref="CanBusService"/> 时，异常会经由
+        /// <see cref="ICanBusService.BackgroundExceptionOccurred"/> 上抛，与本服务其余后台故障共用同一通道。)
+        /// </remarks>
+        /// <param name="service">The service to subscribe on.</param>
+        /// <param name="onNext">Invoked for each accepted frame, in arrival order.</param>
+        /// <param name="predicate">Per-frame filter, or null to accept all frames.</param>
+        /// <param name="bufferCapacity">
+        /// Bounded buffer capacity for the underlying subscription; null uses
+        /// <see cref="CanBusService.DefaultBufferCapacity"/>.
+        /// </param>
+        /// <returns>Disposing this stops the subscription and the background delivery task.</returns>
+        public static IDisposable Subscribe(
+            this ICanBusService service,
+            Action<CanFrameView> onNext,
+            Func<CanFrameView, bool>? 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<bool> _isOnPump = new();
+            private int _disposed;
+
+            public CallbackSubscription(ISubscription subscription, ICanBusService service, Action<CanFrameView> 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 */ }
+            }
+        }
+    }
+}

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<CanKit.Abstractions.API.Can.Definitions.CanFrameView> onNext, System.Func<CanKit.Abstractions.API.Can.Definitions.CanFrameView,System.Boolean> predicate, System.Nullable<System.Int32> 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)

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<int>();
+        var lastReceived = new TaskCompletionSource<bool>(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<bool>(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<bool>(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<bool>(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<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var proceed = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var disposed = new TaskCompletionSource<bool>(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<Exception>(TaskCreationOptions.RunContinuationsAsynchronously);
+        service.BackgroundExceptionOccurred += (_, ex) => observed.TrySetResult(ex);
+
+        var secondReceived = new TaskCompletionSource<bool>(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<InvalidOperationException>().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()