feat(CanBus): add queued TX bus and extension method

pkuyo · pkuyo · commit 95fa6299ded8 · 2025-10-30T22:32:00.000+08:00
Removed conditional compilation for NET8_0_OR_GREATER around GetFramesAsync in all bus adapters and ICanBus abstraction, making async frame streaming available for all targets. Added QueuedTxCanBus and related options to support queued CAN frame transmission with configurable backoff and queue policies.
diff --git a/src/adapters/CanKit.Adapter.ControlCAN/ControlCanBus.cs b/src/adapters/CanKit.Adapter.ControlCAN/ControlCanBus.cs
@@ -270,14 +270,12 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int
         return await _asyncRx.ReceiveBatchAsync(count, timeOut, cancellationToken).ConfigureAwait(false);
     }
 
-#if NET8_0_OR_GREATER
     public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
         await foreach (var item in _asyncRx.ReadAllAsync(cancellationToken))
             yield return item;
     }
-#endif
 
     public BusState BusState => BusState.Unknown;
 
diff --git a/src/adapters/CanKit.Adapter.Kvaser/KvaserBus.cs b/src/adapters/CanKit.Adapter.Kvaser/KvaserBus.cs
@@ -563,7 +563,6 @@ private void HandleBackgroundException(Exception ex)
         try { var snap = Volatile.Read(ref BackgroundExceptionOccurred); snap?.Invoke(this, ex); } catch { /*ignored*/ }
     }
 
-#if NET8_0_OR_GREATER
     public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
@@ -572,5 +571,4 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com
             yield return item;
         }
     }
-#endif
 }
diff --git a/src/adapters/CanKit.Adapter.PCAN/PcanBus.cs b/src/adapters/CanKit.Adapter.PCAN/PcanBus.cs
@@ -254,7 +254,7 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int
             .ConfigureAwait(false);
     }
 
-#if NET8_0_OR_GREATER
+
     public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
@@ -263,7 +263,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com
             yield return item;
         }
     }
-#endif
 
     public PcanBusRtConfigurator Options { get; }
 
diff --git a/src/adapters/CanKit.Adapter.SocketCAN/SocketCanBus.cs b/src/adapters/CanKit.Adapter.SocketCAN/SocketCanBus.cs
@@ -319,7 +319,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int
             .ConfigureAwait(false);
     }
 
-#if NET8_0_OR_GREATER
     public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
@@ -328,7 +327,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com
             yield return item;
         }
     }
-#endif
 
     public void ClearBuffer() => throw new NotSupportedException("SocketCAN does not support clear buffer");
 
diff --git a/src/adapters/CanKit.Adapter.Vector/VectorBus.cs b/src/adapters/CanKit.Adapter.Vector/VectorBus.cs
@@ -360,7 +360,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int
         return await _asyncRx.ReceiveBatchAsync(count, timeOut, cancellationToken).ConfigureAwait(false);
     }
 
-#if NET8_0_OR_GREATER
     public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
@@ -369,7 +368,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([EnumeratorCancella
             yield return frame;
         }
     }
-#endif
 
     public event EventHandler<CanReceiveData> FrameReceived
     {
diff --git a/src/adapters/CanKit.Adapter.Virtual/VirtualBus.cs b/src/adapters/CanKit.Adapter.Virtual/VirtualBus.cs
@@ -140,7 +140,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int
             .ConfigureAwait(false);
     }
 
-#if NET8_0_OR_GREATER
     public async IAsyncEnumerable<CanReceiveData> GetFramesAsync(
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
@@ -150,7 +149,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync(
             yield return item;
         }
     }
-#endif
 
     public VirtualBusRtConfigurator Options { get; }
 
diff --git a/src/adapters/CanKit.Adapter.ZLG/ZlgCanBus.cs b/src/adapters/CanKit.Adapter.ZLG/ZlgCanBus.cs
@@ -629,7 +629,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int
             return await _asyncRx.ReceiveBatchAsync(count, timeOut, cancellationToken).ConfigureAwait(false);
         }
 
-#if NET8_0_OR_GREATER
         public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
         {
             ThrowIfDisposed();
@@ -639,7 +638,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com
             }
 
         }
-#endif
 
         internal int GetAutoSendIndex(bool add = true)
         {
diff --git a/src/core/CanKit.Core/Abstractions/ICanBus.cs b/src/core/CanKit.Core/Abstractions/ICanBus.cs
@@ -167,15 +167,13 @@ public interface ICanBus : IDisposable
         /// <returns>Received frames (收到的帧集合)。当达到期望数量或超时/取消时返回。</returns>
         Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int timeOut = 0, CancellationToken cancellationToken = default);
 
-
-#if NET8_0_OR_GREATER
         /// <summary>
         /// Stream received frames asynchronously (异步流式获取接收帧)
         /// </summary>
         /// <param name="cancellationToken">Cancellation (取消令牌)</param>
         /// <returns>IAsyncEnumerable of frames</returns>
         IAsyncEnumerable<CanReceiveData> GetFramesAsync(CancellationToken cancellationToken = default);
-#endif
+
 
         /// <summary>
         /// Raised when a new CAN frame is received (接收到新 CAN 帧时触发)。
diff --git a/src/core/CanKit.Core/CanBus.cs b/src/core/CanKit.Core/CanBus.cs
@@ -147,6 +147,13 @@ public static TBus Open<TBus, TBusOptions, TInitCfg>(DeviceType deviceType,
         return Open<TBus, TBusOptions, TInitCfg>(device, typedChOptions, typedInitCfg);
     }
 
+
+    public static QueuedCanBus WithQueuedTx(this ICanBus inner, QueuedCanBusOptions? options = null)
+    {
+        options ??= new QueuedCanBusOptions();
+        return new QueuedCanBus(inner, options);
+    }
+
     private sealed class DeviceOwner(ICanDevice device) : IDisposable
     {
         private ICanDevice? _device = device;
diff --git a/src/core/CanKit.Core/CanKit.Core.csproj b/src/core/CanKit.Core/CanKit.Core.csproj
@@ -23,6 +23,7 @@
   <ItemGroup Condition="'$(TargetFramework)'=='netstandard2.0'">
 	<PackageReference Include="System.Memory" Version="4.5.4" />
     <PackageReference Include="System.Runtime.CompilerServices.Unsafe" Version="6.1.2" />
+    <PackageReference Include="System.Threading.Channels" Version="9.0.10" />
   </ItemGroup>
 
     
diff --git a/src/core/CanKit.Core/Utils/AsyncFramePipe.cs b/src/core/CanKit.Core/Utils/AsyncFramePipe.cs
@@ -5,14 +5,12 @@
 using System.Threading;
 using System.Threading.Tasks;
 using CanKit.Core.Definitions;
-#if NET8_0_OR_GREATER
 using System.Threading.Channels;
-#endif
+
 namespace CanKit.Core.Utils;
 
 public sealed class AsyncFramePipe
 {
-#if NET8_0_OR_GREATER
     private readonly Channel<CanReceiveData> _channel;
 
     private volatile TaskCompletionSource<Exception?> _exceptionPulse =
@@ -160,130 +158,4 @@ public void ExceptionOccured(Exception ex)
             new TaskCompletionSource<Exception?>(TaskCreationOptions.RunContinuationsAsynchronously));
         _ = old.TrySetResult(ex);
     }
-
-#else
-    private readonly ConcurrentQueue<CanReceiveData> _queue = new();
-    private readonly int? _capacity;
-    private readonly object _gate = new();
-    private List<TaskCompletionSource<bool>>? _waiters;
-
-    public AsyncFramePipe(int? capacity = null) => _capacity = capacity;
-
-    public void Publish(CanReceiveData frame)
-    {
-        if (_capacity.HasValue)
-        {
-            while (_queue.Count >= _capacity.Value && _queue.TryDequeue(out _))
-            {
-            }
-        }
-
-        _queue.Enqueue(frame);
-
-        List<TaskCompletionSource<bool>>? toWake = null;
-        lock (_gate)
-        {
-            if (_waiters is { Count: > 0 })
-            {
-                toWake = _waiters;
-                _waiters = null;
-            }
-        }
-
-        if (toWake != null)
-            foreach (var t in toWake)
-            {
-                _ = t.TrySetResult(true);
-            }
-    }
-    public void Clear()
-    {
-        while (_queue.TryDequeue(out _))
-        {
-        }
-    }
-    public async Task<IReadOnlyList<CanReceiveData>> ReceiveBatchAsync(
-        int count, int timeoutMs, CancellationToken cancellationToken)
-    {
-        if (count < 0) throw new ArgumentOutOfRangeException(nameof(count));
-        var list = new List<CanReceiveData>((int)Math.Max(1, Math.Min(count, 256)));
-
-        if (timeoutMs == 0)
-        {
-            while ((count == 0 || list.Count < count) && _queue.TryDequeue(out var item))
-                list.Add(item);
-            return list;
-        }
-
-        Stopwatch? sw = timeoutMs > 0 ? Stopwatch.StartNew() : null;
-
-        while (count == 0 || list.Count < count)
-        {
-            while ((count == 0 || list.Count < count) && _queue.TryDequeue(out var item))
-                list.Add(item);
-            if (count != 0 && list.Count >= count) break;
-
-            TaskCompletionSource<bool> tcs;
-            lock (_gate)
-            {
-                tcs = new(TaskCreationOptions.RunContinuationsAsynchronously);
-                _waiters ??= new List<TaskCompletionSource<bool>>();
-                _waiters.Add(tcs);
-            }
-
-            try
-            {
-                if (sw != null) // 有超时
-                {
-                    var remaining = Math.Max(0, timeoutMs - (int)sw.ElapsedMilliseconds);
-                    var delay = Task.Delay(remaining, cancellationToken);
-                    var completed = await Task.WhenAny(tcs.Task, delay).ConfigureAwait(false);
-                    if (completed == delay) break;
-
-                    await tcs.Task.ConfigureAwait(false);
-                }
-                else
-                {
-                    var cancelTcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-                    using (cancellationToken.Register(() => cancelTcs.TrySetResult(true)))
-                    {
-                        var completed = await Task.WhenAny(tcs.Task, cancelTcs.Task).ConfigureAwait(false);
-                        if (completed == cancelTcs.Task) break;
-
-                        await tcs.Task.ConfigureAwait(false);
-                    }
-                }
-            }
-            finally
-            {
-                lock (_gate)
-                {
-                    _waiters?.Remove(tcs);
-                }
-            }
-        }
-
-        return list;
-    }
-    public void ExceptionOccured(Exception ex)
-    {
-        List<TaskCompletionSource<bool>>? toWake = null;
-        lock (_gate)
-        {
-            if (_waiters is { Count: > 0 })
-            {
-                toWake = _waiters;
-                _waiters = null;
-            }
-        }
-        // set all current waiters to exception status
-        if (toWake != null)
-        {
-            foreach (var t in toWake)
-            {
-                _ = t.TrySetException(ex);
-            }
-        }
-    }
-#endif
 }
diff --git a/src/core/CanKit.Core/Utils/PreciseDelay.cs b/src/core/CanKit.Core/Utils/PreciseDelay.cs
diff --git a/src/core/CanKit.Core/Utils/QueuedTxCanBus.cs b/src/core/CanKit.Core/Utils/QueuedTxCanBus.cs

Original file line number	Diff line number	Diff line change
`@@ -270,14 +270,12 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int`
`270`	`270`	`return await _asyncRx.ReceiveBatchAsync(count, timeOut, cancellationToken).ConfigureAwait(false);`
`271`	`271`	`}`
`272`	`272`
`273`		`-#if NET8_0_OR_GREATER`
`274`	`273`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)`
`275`	`274`	`{`
`276`	`275`	`ThrowIfDisposed();`
`277`	`276`	`await foreach (var item in _asyncRx.ReadAllAsync(cancellationToken))`
`278`	`277`	`yield return item;`
`279`	`278`	`}`
`280`		`-#endif`
`281`	`279`
`282`	`280`	`public BusState BusState => BusState.Unknown;`
`283`	`281`
Original file line number	Diff line number	Diff line change
`@@ -563,7 +563,6 @@ private void HandleBackgroundException(Exception ex)`
`563`	`563`	`try { var snap = Volatile.Read(ref BackgroundExceptionOccurred); snap?.Invoke(this, ex); } catch { /ignored/ }`
`564`	`564`	`}`
`565`	`565`
`566`		`-#if NET8_0_OR_GREATER`
`567`	`566`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)`
`568`	`567`	`{`
`569`	`568`	`ThrowIfDisposed();`
`@@ -572,5 +571,4 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com`
`572`	`571`	`yield return item;`
`573`	`572`	`}`
`574`	`573`	`}`
`575`		`-#endif`
`576`	`574`	`}`
Original file line number	Diff line number	Diff line change
`@@ -254,7 +254,7 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int`
`254`	`254`	`.ConfigureAwait(false);`
`255`	`255`	`}`
`256`	`256`
`257`		`-#if NET8_0_OR_GREATER`
	`257`	`+`
`258`	`258`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)`
`259`	`259`	`{`
`260`	`260`	`ThrowIfDisposed();`
`@@ -263,7 +263,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com`
`263`	`263`	`yield return item;`
`264`	`264`	`}`
`265`	`265`	`}`
`266`		`-#endif`
`267`	`266`
`268`	`267`	`public PcanBusRtConfigurator Options { get; }`
`269`	`268`
Original file line number	Diff line number	Diff line change
`@@ -319,7 +319,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int`
`319`	`319`	`.ConfigureAwait(false);`
`320`	`320`	`}`
`321`	`321`
`322`		`-#if NET8_0_OR_GREATER`
`323`	`322`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)`
`324`	`323`	`{`
`325`	`324`	`ThrowIfDisposed();`
`@@ -328,7 +327,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com`
`328`	`327`	`yield return item;`
`329`	`328`	`}`
`330`	`329`	`}`
`331`		`-#endif`
`332`	`330`
`333`	`331`	`public void ClearBuffer() => throw new NotSupportedException("SocketCAN does not support clear buffer");`
`334`	`332`
Original file line number	Diff line number	Diff line change
`@@ -360,7 +360,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int`
`360`	`360`	`return await _asyncRx.ReceiveBatchAsync(count, timeOut, cancellationToken).ConfigureAwait(false);`
`361`	`361`	`}`
`362`	`362`
`363`		`-#if NET8_0_OR_GREATER`
`364`	`363`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)`
`365`	`364`	`{`
`366`	`365`	`ThrowIfDisposed();`
`@@ -369,7 +368,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([EnumeratorCancella`
`369`	`368`	`yield return frame;`
`370`	`369`	`}`
`371`	`370`	`}`
`372`		`-#endif`
`373`	`371`
`374`	`372`	`public event EventHandler<CanReceiveData> FrameReceived`
`375`	`373`	`{`
Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int`
`140`	`140`	`.ConfigureAwait(false);`
`141`	`141`	`}`
`142`	`142`
`143`		`-#if NET8_0_OR_GREATER`
`144`	`143`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync(`
`145`	`144`	`[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)`
`146`	`145`	`{`
`@@ -150,7 +149,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync(`
`150`	`149`	`yield return item;`
`151`	`150`	`}`
`152`	`151`	`}`
`153`		`-#endif`
`154`	`152`
`155`	`153`	`public VirtualBusRtConfigurator Options { get; }`
`156`	`154`
Original file line number	Diff line number	Diff line change
`@@ -629,7 +629,6 @@ public async Task<IReadOnlyList<CanReceiveData>> ReceiveAsync(int count = 1, int`
`629`	`629`	`return await _asyncRx.ReceiveBatchAsync(count, timeOut, cancellationToken).ConfigureAwait(false);`
`630`	`630`	`}`
`631`	`631`
`632`		`-#if NET8_0_OR_GREATER`
`633`	`632`	`public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)`
`634`	`633`	`{`
`635`	`634`	`ThrowIfDisposed();`
`@@ -639,7 +638,6 @@ public async IAsyncEnumerable<CanReceiveData> GetFramesAsync([System.Runtime.Com`
`639`	`638`	`}`
`640`	`639`
`641`	`640`	`}`
`642`		`-#endif`
`643`	`641`
`644`	`642`	`internal int GetAutoSendIndex(bool add = true)`
`645`	`643`	`{`
Original file line number	Diff line number	Diff line change
`@@ -147,6 +147,13 @@ public static TBus Open<TBus, TBusOptions, TInitCfg>(DeviceType deviceType,`
`147`	`147`	`return Open<TBus, TBusOptions, TInitCfg>(device, typedChOptions, typedInitCfg);`
`148`	`148`	`}`
`149`	`149`
	`150`	`+`
	`151`	`+ public static QueuedCanBus WithQueuedTx(this ICanBus inner, QueuedCanBusOptions? options = null)`
	`152`	`+ {`
	`153`	`+ options ??= new QueuedCanBusOptions();`
	`154`	`+ return new QueuedCanBus(inner, options);`
	`155`	`+ }`
	`156`	`+`
`150`	`157`	`private sealed class DeviceOwner(ICanDevice device) : IDisposable`
`151`	`158`	`{`
`152`	`159`	`private ICanDevice? _device = device;`