From a3f465f60a7bc6519172e183f95a08f4a4de4171 Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 27 Aug 2026 14:35:58 +0800 Subject: [PATCH] =?UTF-8?q?feat!:=20netstandard2.0=20=E5=A4=9A=E7=9B=AE?= =?UTF-8?q?=E6=A0=87=EF=BC=88=E5=85=BC=E5=AE=B9=E5=B1=82=20+=20net48=20?= =?UTF-8?q?=E5=85=A8=E9=87=8F=E6=B5=8B=E8=AF=95=20+=20=E8=B7=A8=E5=B9=B3?= =?UTF-8?q?=E5=8F=B0=20CI=20=E7=9F=A9=E9=98=B5=EF=BC=89=EF=BC=8C=E7=89=88?= =?UTF-8?q?=E6=9C=AC=202.1.0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .editorconfig | 8 ++ .github/workflows/ci.yml | 21 ++++- .github/workflows/release.yml | 44 ++++++++-- CHANGELOG.md | 22 ++++- README.en.md | 10 +++ README.md | 10 +++ docs/DESIGN.md | 9 ++ .../StreamFrame.Protocols.Xml.csproj | 9 +- src/StreamFrame/Codec/BufferWriterStream.cs | 8 +- .../Compatibility/NetStandard20Shims.cs | 69 +++++++++++++++ .../Connection/StreamConnection.cs | 58 ++++++++++-- src/StreamFrame/Framing/LengthPrefixFramer.cs | 2 +- src/StreamFrame/Framing/StxEtxFramer.cs | 88 ++++++++++++------- src/StreamFrame/StreamFrame.csproj | 15 +++- .../ConnectionLifetimeTests.cs | 4 +- .../ConnectionResilienceTests.cs | 24 ++--- .../ConnectionUsabilityTests.cs | 6 +- test/StreamFrame.Tests/FrameDecoderTests.cs | 16 ++-- .../LengthPrefixFramerTests.cs | 10 +-- .../SingleClientConnectionTests.cs | 4 + .../StreamFrame.Tests.csproj | 9 +- .../StreamingFrameEncodeTests.cs | 2 +- test/StreamFrame.Tests/StxEtxFramerTests.cs | 20 +++-- test/StreamFrame.Tests/TestCompat.cs | 52 +++++++++++ test/StreamFrame.Tests/TestCompat.net48.cs | 54 ++++++++++++ .../XmlDocumentCodecTests.cs | 2 +- 26 files changed, 473 insertions(+), 103 deletions(-) create mode 100644 src/StreamFrame/Compatibility/NetStandard20Shims.cs create mode 100644 test/StreamFrame.Tests/TestCompat.cs create mode 100644 test/StreamFrame.Tests/TestCompat.net48.cs diff --git a/.editorconfig b/.editorconfig index ffa30a7..1e1e55c 100644 --- a/.editorconfig +++ b/.editorconfig @@ -41,6 +41,14 @@ dotnet_diagnostic.MA0046.severity = none # 选项校验用 nameof(属性) 作为 ArgumentOutOfRangeException 参数名：校验对象是属性而非方法参数 dotnet_diagnostic.MA0015.severity = none +# 兼容 shim 层：一个文件聚合多个小类型是刻意设计（与 #if 块对应） +[src/StreamFrame/Compatibility/**.cs] +dotnet_diagnostic.MA0048.severity = none + +# 测试侧兼容层同理 +[test/StreamFrame.Tests/TestCompat*.cs] +dotnet_diagnostic.MA0048.severity = none + # 测试项目：xUnit 无同步上下文，ConfigureAwait 与比较器重载建议均为噪音 [test/**.cs] dotnet_diagnostic.MA0002.severity = none diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c59e7c6..3362785 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,7 +12,16 @@ permissions: jobs: build-test: - runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + include: + # 现代 .NET 在 linux 跑；net48（验证 netstandard2.0 资产）需要 Windows 运行时 + - os: ubuntu-latest + frameworks: "net8.0 net10.0" + - os: windows-latest + frameworks: "net48" + runs-on: ${{ matrix.os }} steps: - uses: actions/checkout@v7 @@ -26,5 +35,11 @@ jobs: - name: Build run: dotnet build StreamFrame.slnx -c Release - - name: Test - run: dotnet test StreamFrame.slnx -c Release --no-build + - name: Test (${{ matrix.frameworks }}) + shell: bash + env: + FRAMEWORKS: ${{ matrix.frameworks }} + run: | + for fw in $FRAMEWORKS; do + dotnet test StreamFrame.slnx -c Release --no-build -f $fw + done diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 4a7e28f..c1d1af0 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -16,7 +16,42 @@ permissions: id-token: write # OIDC Trusted Publishing 换取 nuget.org 短时 API key jobs: + # 测试矩阵：所有 OS×TFM 组合通过后才允许发布 + test: + strategy: + fail-fast: false + matrix: + include: + - os: ubuntu-latest + frameworks: "net8.0 net10.0" + - os: windows-latest + frameworks: "net48" + runs-on: ${{ matrix.os }} + steps: + - uses: actions/checkout@v7 + + - name: Set up .NET + uses: actions/setup-dotnet@v6 + with: + dotnet-version: | + 8.0.x + 10.0.x + + - name: Build + run: dotnet build StreamFrame.slnx -c Release + + - name: Test (${{ matrix.frameworks }}) + shell: bash + env: + FRAMEWORKS: ${{ matrix.frameworks }} + run: | + for fw in $FRAMEWORKS; do + dotnet test StreamFrame.slnx -c Release --no-build -f $fw + done + release: + needs: test + if: startsWith(github.ref, 'refs/tags/') runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 @@ -31,7 +66,6 @@ jobs: 10.0.x - name: Verify tag matches csproj versions - if: startsWith(github.ref, 'refs/tags/') run: | version="${GITHUB_REF_NAME#v}" for proj in src/StreamFrame/StreamFrame.csproj src/StreamFrame.Protocols.Xml/StreamFrame.Protocols.Xml.csproj; do @@ -56,7 +90,6 @@ jobs: # ---------- 以下步骤仅在 tag 推送时执行 ---------- - name: Extract changelog section for this version - if: startsWith(github.ref, 'refs/tags/') id: notes run: | version="${GITHUB_REF_NAME#v}" @@ -73,7 +106,7 @@ jobs: fi - name: Create Release (from CHANGELOG) - if: startsWith(github.ref, 'refs/tags/') && steps.notes.outputs.has_notes == 'true' + if: steps.notes.outputs.has_notes == 'true' uses: softprops/action-gh-release@v3 with: tag_name: ${{ github.ref_name }} @@ -84,7 +117,7 @@ jobs: artifacts/*.snupkg - name: Create Release (auto-generated notes) - if: startsWith(github.ref, 'refs/tags/') && steps.notes.outputs.has_notes != 'true' + if: steps.notes.outputs.has_notes != 'true' uses: softprops/action-gh-release@v3 with: tag_name: ${{ github.ref_name }} @@ -95,7 +128,6 @@ jobs: artifacts/*.snupkg - name: Check NUGET_USER variable - if: startsWith(github.ref, 'refs/tags/') run: | if [ -z "${NUGET_USER}" ]; then echo "::error::NUGET_USER 为空。Settings → Secrets and variables → Actions → Variables 添加 NUGET_USER（nuget.org 用户名，注意是 Variables 不是 Secrets）。" @@ -105,14 +137,12 @@ jobs: NUGET_USER: ${{ vars.NUGET_USER }} - name: NuGet Login (OIDC) - if: startsWith(github.ref, 'refs/tags/') id: login uses: NuGet/login@v1 with: user: ${{ vars.NUGET_USER }} - name: Push to nuget.org - if: startsWith(github.ref, 'refs/tags/') run: | dotnet nuget push "artifacts/*.nupkg" \ --api-key "${{ steps.login.outputs.NUGET_API_KEY }}" \ diff --git a/CHANGELOG.md b/CHANGELOG.md index 7832eed..b66f558 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,11 +5,24 @@ ## [Unreleased] -### 变更 -- CI Actions 升级到 Node 24 运行时：`actions/checkout@v7`、`actions/setup-dotnet@v6`、`softprops/action-gh-release@v3`（消除 Node 20 弃用告警）。 +### 新增 +- 暂无 + +### 修复 +- 暂无 + +## [2.1.0] - 2026-08-27 ### 新增 -- 多目标框架（第一阶段）：`net8.0;net10.0`——net10 为当前 LTS（支持至 2028-11），CI 双框架全量测试；net9+ 目标采用 `System.Threading.Lock`。 +- **多目标框架：`netstandard2.0;net8.0;net10.0`**——打开 .NET Framework 4.6.2+/Unity/Mono 受众（设备通讯/WMS 场景存量巨大）。ns2.0 资产经 net48 全量测试（真实 TCP 回环）验证；CI 矩阵 Ubuntu（net8/net10）+ Windows（net48）；发布流水线改为"测试矩阵全过 → 再发布"。兼容要点：PolySharp 语言 polyfill、Socket TaskExtensions 回退 + 可取消等待、KeepAlive 经 SIO_KEEPALIVE_VALS、StxEtx 定界器改手工扫描（SequenceReader 无 ns2.0 包资产）。net9+ 目标采用 `System.Threading.Lock`。 +- 帧定界基准测试 `bench/StreamFrame.Benchmarks`（BenchmarkDotNet）：LengthPrefix/StxEtx 的流式 vs 纯函数编码对比与切帧吞吐，为"流式零拷贝"说法提供可复现数据。 +- 项目 LOGO（`docs/logo/`，SVG 母版 + PNG），README 双语版接入展示、两个 NuGet 包启用包图标（随本版本发布生效）。 +- README 新增 CI 构建状态徽章与"支持框架"矩阵；main 分支保护（必须走 PR 且 CI 通过、禁 force push/删除、强制线性历史）。 +- 英文版 README（`README.en.md`，与中文版互链），便于国际用户检索与阅读。 + +### 变更 +- CI Actions 升级到 Node 24 运行时：`actions/checkout@v7`、`actions/setup-dotnet@v6`、`softprops/action-gh-release@v3`（消除 Node 20 弃用告警）。 +- `GetMessages`/发送 worker 改用 `WaitToReadAsync+TryRead` 等价循环（ns2.0 的 Channels 包无 `ReadAllAsync`；各目标行为一致）。 - 帧定界基准测试 `bench/StreamFrame.Benchmarks`（BenchmarkDotNet）：LengthPrefix/StxEtx 的流式 vs 纯函数编码对比与切帧吞吐，为"流式零拷贝"说法提供可复现数据。 - 项目 LOGO（`docs/logo/`，SVG 母版 + PNG），README 双语版接入展示、两个 NuGet 包启用包图标（随下个版本发布生效）。 - README 新增 CI 构建状态徽章；main 分支保护（必须走 PR 且 CI 通过、禁 force push/删除、强制线性历史）。 @@ -98,7 +111,8 @@ - 发布自动化：`release.yml`（tag 自动建 GitHub Release）、`publish-nuget.yml`（OIDC Trusted Publishing 推送 nuget.org）。 - 单元测试 25 个 + 三场景端到端 demo。 -[Unreleased]: https://github.com/CSJ608/StreamFrame/compare/v2.0.0...HEAD +[Unreleased]: https://github.com/CSJ608/StreamFrame/compare/v2.1.0...HEAD +[2.1.0]: https://github.com/CSJ608/StreamFrame/compare/v2.0.0...v2.1.0 [2.0.0]: https://github.com/CSJ608/StreamFrame/compare/v1.2.0...v2.0.0 [1.2.0]: https://github.com/CSJ608/StreamFrame/compare/v1.1.0...v1.2.0 [1.1.0]: https://github.com/CSJ608/StreamFrame/compare/v1.0.1...v1.1.0 diff --git a/README.en.md b/README.en.md index 542a9e8..dc63cd8 100644 --- a/README.en.md +++ b/README.en.md @@ -154,6 +154,16 @@ Measured with BenchmarkDotNet (details and how to reproduce in [bench/README.md] - **Streaming encode**: halves per-frame heap allocation (single vs. double buffer); 25–35% faster for small payloads, on par for large ones; - **Frame decoding**: `LengthPrefixFramer` ≈10 ns/frame vs `StxEtxFramer` ≈1.1 µs/frame (byte-by-byte scanning) — prefer length-prefix for high-throughput scenarios. +## Supported frameworks + +| Package target | Runtime | +|---|---| +| `net10.0` (recommended, LTS until 2028-11) | .NET 10 | +| `net8.0` (LTS until 2026-11) | .NET 8 | +| `netstandard2.0` | .NET Framework 4.6.2+, Unity, Mono, ... | + +The `netstandard2.0` asset is validated by the **full net48 test suite** (real TCP loopback); TCP KeepAlive parameters are set via `SIO_KEEPALIVE_VALS` on .NET Framework. CI runs all tests on Ubuntu (net8/net10) and Windows (net48). + ## Dependencies - [System.IO.Pipelines](https://www.nuget.org/packages/System.IO.Pipelines) diff --git a/README.md b/README.md index 7926a94..b96e73b 100644 --- a/README.md +++ b/README.md @@ -154,6 +154,16 @@ BenchmarkDotNet 实测（详见 [bench/README.md](bench/README.md)，可本地 - **流式编码**：每帧堆分配减半（单缓冲 vs 双缓冲），小负载耗时快 25–35%，大负载持平； - **切帧吞吐**：`LengthPrefixFramer` ≈10 ns/帧，`StxEtxFramer` ≈1.1 µs/帧（逐字节扫描）——高吞吐场景优先长度前缀。 +## 支持框架 + +| 包目标 | 运行环境 | +|---|---| +| `net10.0`（推荐，LTS 至 2028-11） | .NET 10 | +| `net8.0`（LTS 至 2026-11） | .NET 8 | +| `netstandard2.0` | .NET Framework 4.6.2+、Unity、Mono 等 | + +`netstandard2.0` 资产经 **net48 全量测试套件**（真实 TCP 回环）验证；TCP KeepAlive 参数在 .NET Framework 上通过 `SIO_KEEPALIVE_VALS` 设置。CI 在 Ubuntu（net8/net10）与 Windows（net48）双平台运行全部测试。 + ## 依赖 - [System.IO.Pipelines](https://www.nuget.org/packages/System.IO.Pipelines) diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 52032cf..94cc0c7 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -142,6 +142,15 @@ socket --ReceiveAsync--> pipe.Writer --[Pipe]--> FrameDecoder.RunAsync **原始字节与诊断（1.2.0）**：`RawBytesReceived/Sent` 全量输出（含被定界器丢弃的噪声字节），发送侧按实际写出分片回调；两者与 `ConnectionChanged`、`FrameError` 的用户回调异常均被隔离，不反噬会话。`IFrameDiscardReporting`（可选接口，内置两个 codec 实现）让定界器精确上报重同步丢弃的字节，经 `FrameError(DiscardedByResync)` 交上层调试。 +### netstandard2.0 兼容层（2.1.0） + +多目标 `netstandard2.0;net8.0;net10.0`： +- **PolySharp**（PrivateAssets）：Range/Index/init 等语言级 polyfill，不进包。 +- **`Compatibility/NetStandard20Shims.cs`**：`Task.WaitAsync` 等价物；Socket 的 Memory 收发回退到 `SocketTaskExtensions`——System.Net.Sockets 4.3.0 没有带 CancellationToken 的重载，用 3 参重载 + "可取消等待"包装（netfx 本就无法中止已提交的 socket I/O，取消仅放弃等待）；TCP KeepAlive 参数经 `SIO_KEEPALIVE_VALS` 设置（netfx 无 TcpKeepAliveTime/Interval 选项名）。 +- **StxEtx 定界器改为手工扫描**：`SequenceReader` 从未发布 netstandard2.0 包资产。单段缓冲走 Span 快路径（零分配），多段租借临时数组；所有 TFM 共用同一实现（`ReadOnlySequence.FirstSpan` 同理改用 `First.Span`）。 +- ns2.0 目标额外引用 `System.Threading.Channels` + `Microsoft.Bcl.AsyncInterfaces`（现代 .NET 内置于共享框架，NU1510 要求移除显式引用）。 +- **验证矩阵**：测试项目多目标 `net8.0;net10.0;net48`；CI 在 Ubuntu 跑 net8/net10、Windows 跑 net48（真实 .NET Framework 运行时加载 ns2.0 资产执行全量测试）。 + ### 不内置（上移驱动层） - 心跳 / 控制消息（泛型下框架不知消息形态） diff --git a/src/StreamFrame.Protocols.Xml/StreamFrame.Protocols.Xml.csproj b/src/StreamFrame.Protocols.Xml/StreamFrame.Protocols.Xml.csproj index 71bff1b..3b30a10 100644 --- a/src/StreamFrame.Protocols.Xml/StreamFrame.Protocols.Xml.csproj +++ b/src/StreamFrame.Protocols.Xml/StreamFrame.Protocols.Xml.csproj @@ -10,6 +10,10 @@ + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + runtime; build; native; contentfiles; analyzers; buildtransitive all @@ -17,15 +21,16 @@ - net8.0;net10.0 + netstandard2.0;net8.0;net10.0 enable enable true + latest StreamFrame.Protocols.Xml - 2.0.0 + 2.1.0 StreamFrame 的 XML 报文示例驱动：XmlDocumentCodec（XmlSerializer / XmlWriter 编解码）。 Marci MIT diff --git a/src/StreamFrame/Codec/BufferWriterStream.cs b/src/StreamFrame/Codec/BufferWriterStream.cs index 31aa19d..252676a 100644 --- a/src/StreamFrame/Codec/BufferWriterStream.cs +++ b/src/StreamFrame/Codec/BufferWriterStream.cs @@ -35,9 +35,15 @@ public override void SetLength(long value) => throw new NotSupportedException(); public override void Write(byte[] buffer, int offset, int count) - => Write(buffer.AsSpan(offset, count)); + => WriteCore(buffer.AsSpan(offset, count)); +#if !NETSTANDARD2_0 + // Stream 的 Span 虚方法重载自 netstandard2.1 起才有；ns2.0 走上面的 byte[] 重载（同样直写 IBufferWriter） public override void Write(ReadOnlySpan buffer) + => WriteCore(buffer); +#endif + + private void WriteCore(ReadOnlySpan buffer) { while (!buffer.IsEmpty) { diff --git a/src/StreamFrame/Compatibility/NetStandard20Shims.cs b/src/StreamFrame/Compatibility/NetStandard20Shims.cs new file mode 100644 index 0000000..a6d3ced --- /dev/null +++ b/src/StreamFrame/Compatibility/NetStandard20Shims.cs @@ -0,0 +1,69 @@ +#if NETSTANDARD2_0 +using System; +using System.Net.Sockets; +using System.Runtime.InteropServices; +using System.Threading; +using System.Threading.Tasks; + +namespace System.Threading.Tasks +{ + internal static class TaskWaitAsyncCompat + { + /// Task.WaitAsync(CancellationToken) 的 netstandard2.0 等价物（net5+ 内置）。 + public static async Task WaitAsync(this Task task, CancellationToken cancellationToken) + { + var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var registration = cancellationToken.Register( + static state => ((TaskCompletionSource)state!).TrySetResult(true), + cancelled, + useSynchronizationContext: false); + + var completed = await Task.WhenAny(task, cancelled.Task).ConfigureAwait(false); + if (completed != task) + throw new OperationCanceledException(cancellationToken); + + await task.ConfigureAwait(false); + } + + /// + /// 旧 SocketTaskExtensions（System.Net.Sockets 4.3.0）没有带 CancellationToken 的重载： + /// 用 3 参重载发起操作，等待阶段可被取消（netfx 本就无法真正中止已提交的 socket I/O， + /// 取消后操作结果被丢弃，socket 随后关闭）。 + /// + public static async Task WithCancellation(this Task task, CancellationToken cancellationToken) + { + var cancelled = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var registration = cancellationToken.Register( + static state => ((TaskCompletionSource)state!).TrySetResult(true), + cancelled, + useSynchronizationContext: false); + + var completed = await Task.WhenAny(task, cancelled.Task).ConfigureAwait(false); + if (completed != task) + throw new OperationCanceledException(cancellationToken); + + return await task.ConfigureAwait(false); + } + } +} + +namespace System.Net.Sockets +{ + internal static class SocketMemoryCompat + { + public static Task ReceiveAsync( + this Socket socket, Memory buffer, SocketFlags socketFlags, CancellationToken cancellationToken) + => MemoryMarshal.TryGetArray((ReadOnlyMemory)buffer, out var segment) + ? SocketTaskExtensions.ReceiveAsync(socket, segment, socketFlags).WithCancellation(cancellationToken) + : Task.FromException(new ArgumentException( + "netstandard2.0 目标要求接收缓冲为数组支撑的 Memory。", nameof(buffer))); + + public static Task SendAsync( + this Socket socket, ReadOnlyMemory buffer, SocketFlags socketFlags, CancellationToken cancellationToken) + => MemoryMarshal.TryGetArray(buffer, out var segment) + ? SocketTaskExtensions.SendAsync(socket, segment, socketFlags).WithCancellation(cancellationToken) + : Task.FromException(new ArgumentException( + "netstandard2.0 目标要求发送缓冲为数组支撑的 Memory。", nameof(buffer))); + } +} +#endif diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index dd606ca..2798251 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -87,7 +87,7 @@ public string? RemoteIpAddress private CancellationTokenRegistration _lifetimeRegistration; /// 等待连接就绪的挂起等待者；进入 Connected 时完成并清空，Dispose 时取消。 - private TaskCompletionSource? _whenConnected; + private TaskCompletionSource? _whenConnected; /// /// 创建一条连接。 @@ -221,7 +221,12 @@ private async Task ConnectAsync(CancellationToken ct) var socket = CreateTcpSocket(); try { +#if NETSTANDARD2_0 + // ns2.0 的 SocketTaskExtensions 无带 ct 的重载：IPAddress[] 形式直连（无 DNS）+ 可取消等待 + await SocketTaskExtensions.ConnectAsync(socket, new[] { IpAddress }, Port).WaitAsync(ct).ConfigureAwait(false); +#else await socket.ConnectAsync(IpAddress, Port, ct).ConfigureAwait(false); +#endif } catch { @@ -247,7 +252,13 @@ private async Task AcceptAsync(CancellationToken ct) try { +#if NETSTANDARD2_0 + var listener = _server!; + // FromAsync 无法取消；停机时 listener 被释放，挂起的 accept 以异常收尾 + var socket = await Task.Factory.FromAsync(listener.BeginAccept, listener.EndAccept, null).ConfigureAwait(false); +#else var socket = await _server!.AcceptAsync(ct).ConfigureAwait(false); +#endif ConfigureSocket(socket); // 单客户端模式：accept 到第一个客户端后关闭监听 socket， @@ -297,8 +308,27 @@ private void ConfigureSocket(Socket socket) return; socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true); +#if NETSTANDARD2_0 + // netfx 没有TcpKeepAliveTime/Interval 选项名：Windows 上用 SIO_KEEPALIVE_VALS 设置等价参数。 + // netstandard2.0 资产实际运行环境主要是 .NET Framework（仅 Windows）；失败只记日志， + // 保留系统默认 KeepAlive 参数（2 小时）兜底。 + try + { + const int SioKeepAliveValues = -1744830460; + var values = new byte[12]; + BitConverter.GetBytes(1).CopyTo(values, 0); // onOff = 1 + BitConverter.GetBytes(_options.KeepAliveTimeMs).CopyTo(values, 4); // 首次探测前静默 ms + BitConverter.GetBytes(_options.KeepAliveIntervalMs).CopyTo(values, 8); // 探测间隔 ms + socket.IOControl(SioKeepAliveValues, values, null); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "SIO_KEEPALIVE_VALS 设置失败（非 Windows 平台？），回退系统默认 KeepAlive 参数。"); + } +#else socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, _options.KeepAliveTimeMs); socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, _options.KeepAliveIntervalMs); +#endif } /// 立即进入重连流程。 @@ -356,7 +386,7 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE if (newState == ConnectionState.Connected) { // 唤醒所有 WaitForConnectedAsync 等待者；下次离开 Connected 再等时重新创建 - Interlocked.Exchange(ref _whenConnected, null)?.TrySetResult(); + Interlocked.Exchange(ref _whenConnected, null)?.TrySetResult(true); } switch (newState) @@ -561,9 +591,11 @@ private async Task SendWorkerAsync(CancellationToken ct) { // 发送失败（socket 故障/会话失效）不在此吞掉：上抛给会话守护 → 断线重连， // 未发送的消息留在连接级队列，由重连后的新 worker 继续发送。 - await foreach (var message in _sendQueue.Reader.ReadAllAsync(ct).ConfigureAwait(false)) + var reader = _sendQueue.Reader; + while (await reader.WaitToReadAsync(ct).ConfigureAwait(false)) { - await SendFramedAsync(message, ct).ConfigureAwait(false); + while (reader.TryRead(out var message)) + await SendFramedAsync(message, ct).ConfigureAwait(false); } } @@ -664,7 +696,7 @@ public Task WaitForConnectedAsync(CancellationToken ct = default) } /// 获取或创建"等待 Connected"的完成源；Connected 时完成并置空，可重复使用。 - private TaskCompletionSource GetOrCreateWhenConnected() + private TaskCompletionSource GetOrCreateWhenConnected() { while (true) { @@ -672,19 +704,27 @@ private TaskCompletionSource GetOrCreateWhenConnected() if (existing is not null) return existing; - var created = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var created = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); if (Interlocked.CompareExchange(ref _whenConnected, created, null) is null) return created; } } - public IAsyncEnumerable GetMessages(CancellationToken ct = default) - => _messageRelay.Reader.ReadAllAsync(ct); + public async IAsyncEnumerable GetMessages( + [EnumeratorCancellation] CancellationToken ct = default) + { + var reader = _messageRelay.Reader; + while (await reader.WaitToReadAsync(ct).ConfigureAwait(false)) + { + while (reader.TryRead(out var message)) + yield return message; + } + } public ValueTask DisposeAsync() { Shutdown(); - return ValueTask.CompletedTask; + return new ValueTask(); } /// diff --git a/src/StreamFrame/Framing/LengthPrefixFramer.cs b/src/StreamFrame/Framing/LengthPrefixFramer.cs index 846777b..ad24dd9 100644 --- a/src/StreamFrame/Framing/LengthPrefixFramer.cs +++ b/src/StreamFrame/Framing/LengthPrefixFramer.cs @@ -84,7 +84,7 @@ private static int ReadLengthPrefix(in ReadOnlySequence buffer) { var header = buffer.Slice(0, LengthPrefixSize); if (header.IsSingleSegment) - return BinaryPrimitives.ReadInt32BigEndian(header.FirstSpan); + return BinaryPrimitives.ReadInt32BigEndian(header.First.Span); Span tmp = stackalloc byte[LengthPrefixSize]; header.CopyTo(tmp); diff --git a/src/StreamFrame/Framing/StxEtxFramer.cs b/src/StreamFrame/Framing/StxEtxFramer.cs index 6b71b76..2f2a161 100644 --- a/src/StreamFrame/Framing/StxEtxFramer.cs +++ b/src/StreamFrame/Framing/StxEtxFramer.cs @@ -85,46 +85,68 @@ public bool TryDecodeFrame(ref ReadOnlySequence buffer, out ReadOnlySequen payload = default; var original = buffer; - var reader = new SequenceReader(buffer); - - long lastStxOffset = -1; - long lastStxPayloadStart = -1; + // 手工扫描（不依赖 SequenceReader——它从未发布 netstandard2.0 包资产）： + // 单段缓冲直接扫 Span（零分配）；多段缓冲租借临时数组拼连续后扫。 + byte[]? rented = null; + ReadOnlySpan span; + if (buffer.IsSingleSegment) + { + span = buffer.First.Span; // FirstSpan（netcore 专属）不可用；单段时 First.Span 等价 + } + else + { + rented = ArrayPool.Shared.Rent((int)buffer.Length); + buffer.CopyTo(rented); + span = rented.AsSpan(0, (int)buffer.Length); + } - while (reader.TryRead(out var b)) + try { - if (b == STX) - { - lastStxOffset = reader.Consumed - 1; // STX 位置 - lastStxPayloadStart = reader.Consumed; // 负载起始位置 - } - else if (b == ETX) - { - if (lastStxPayloadStart < 0) - continue; // 孤立 ETX，跳过 + long lastStxOffset = -1; + long lastStxPayloadStart = -1; - var payloadLength = reader.Consumed - 1 - lastStxPayloadStart; - if (payloadLength > MaxPayloadBytes) + for (var i = 0; i < span.Length; i++) + { + var b = span[i]; + if (b == STX) { - // 超长帧：丢弃当前部分帧并继续扫描 - lastStxPayloadStart = -1; - lastStxOffset = -1; - continue; + lastStxOffset = i; // STX 位置 + lastStxPayloadStart = i + 1; // 负载起始位置 + } + else if (b == ETX) + { + if (lastStxPayloadStart < 0) + continue; // 孤立 ETX，跳过 + + var payloadLength = i - lastStxPayloadStart; + if (payloadLength > MaxPayloadBytes) + { + // 超长帧：丢弃当前部分帧并继续扫描 + lastStxPayloadStart = -1; + lastStxOffset = -1; + continue; + } + + payload = buffer.Slice(lastStxPayloadStart, payloadLength); + buffer = buffer.Slice(i + 1); + discarded = original.Slice(0, lastStxOffset); // 保留点之前的噪声/被中止半帧 + return true; } - - payload = buffer.Slice(lastStxPayloadStart, payloadLength); - buffer = buffer.Slice(reader.Consumed); - discarded = original.Slice(0, lastStxOffset); // 保留点之前的噪声/被中止半帧 - return true; } - } - // 缓冲区耗尽但未收到完整帧：保留自最后一个 STX 起的余量等待后续数据。 - // 保留点之前的字节（杂散噪声、被更新的 STX 中止的旧半帧）即为本次丢弃。 - discarded = original.Slice(0, lastStxOffset >= 0 ? lastStxOffset : original.Length); - buffer = lastStxOffset >= 0 - ? buffer.Slice(lastStxOffset) - : buffer.Slice(buffer.End); + // 缓冲区耗尽但未收到完整帧：保留自最后一个 STX 起的余量等待后续数据。 + // 保留点之前的字节（杂散噪声、被更新的 STX 中止的旧半帧）即为本次丢弃。 + discarded = original.Slice(0, lastStxOffset >= 0 ? lastStxOffset : original.Length); + buffer = lastStxOffset >= 0 + ? buffer.Slice(lastStxOffset) + : buffer.Slice(buffer.End); - return false; + return false; + } + finally + { + if (rented is not null) + ArrayPool.Shared.Return(rented); + } } } diff --git a/src/StreamFrame/StreamFrame.csproj b/src/StreamFrame/StreamFrame.csproj index 8f68da2..32c1550 100644 --- a/src/StreamFrame/StreamFrame.csproj +++ b/src/StreamFrame/StreamFrame.csproj @@ -1,15 +1,17 @@ - net8.0;net10.0 + netstandard2.0;net8.0;net10.0 enable enable true + + latest StreamFrame - 2.0.0 + 2.1.0 通用 socket 通讯框架：帧定界（IFramer）与帧内编解码（ICodec<TMessage>）插件化，支持长度前缀 / STX-ETX 定界与 IPv4/IPv6 双栈，可选流式零拷贝发送。 Marci MIT @@ -30,10 +32,19 @@ all + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + diff --git a/test/StreamFrame.Tests/ConnectionLifetimeTests.cs b/test/StreamFrame.Tests/ConnectionLifetimeTests.cs index a6982c1..e90dfb0 100644 --- a/test/StreamFrame.Tests/ConnectionLifetimeTests.cs +++ b/test/StreamFrame.Tests/ConnectionLifetimeTests.cs @@ -34,8 +34,8 @@ private static async Task WaitForStateAsync( ConnectionState expected, int timeoutMs = 5000) { - var deadline = Environment.TickCount64 + timeoutMs; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) { if (connection.State == expected) return; diff --git a/test/StreamFrame.Tests/ConnectionResilienceTests.cs b/test/StreamFrame.Tests/ConnectionResilienceTests.cs index f3736b4..669887b 100644 --- a/test/StreamFrame.Tests/ConnectionResilienceTests.cs +++ b/test/StreamFrame.Tests/ConnectionResilienceTests.cs @@ -37,8 +37,8 @@ private static async Task WaitForStateAsync( Func predicate, int timeoutMs = 5000) { - var deadline = Environment.TickCount64 + timeoutMs; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) { if (predicate(connection.State)) return; @@ -93,8 +93,8 @@ public async Task DisconnectThenReconnect_MessagesStillDelivered() await client1.GetStream().WriteAsync(Frame("before-disconnect")); } - var deadline = Environment.TickCount64 + 5000; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + 5000; + while (TestClock.TickCount64 < deadline) { lock (received) { @@ -118,8 +118,8 @@ public async Task DisconnectThenReconnect_MessagesStillDelivered() await client2.GetStream().WriteAsync(Frame("after-reconnect")); } - deadline = Environment.TickCount64 + 5000; - while (Environment.TickCount64 < deadline) + deadline = TestClock.TickCount64 + 5000; + while (TestClock.TickCount64 < deadline) { lock (received) { @@ -186,8 +186,8 @@ public async Task DecodeError_DisconnectPolicy_ReconnectsAndRecovers() await client2.GetStream().WriteAsync(Frame("recovered")); } - var deadline = Environment.TickCount64 + 5000; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + 5000; + while (TestClock.TickCount64 < deadline) { lock (received) { @@ -239,8 +239,8 @@ public async Task DecodeError_SkipFramePolicy_KeepsConnectionAndDeliversNext() var bytes = Frame("poison").Concat(Frame("good")).ToArray(); await client.GetStream().WriteAsync(bytes); - var deadline = Environment.TickCount64 + 5000; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + 5000; + while (TestClock.TickCount64 < deadline) { lock (received) { @@ -299,8 +299,8 @@ public async Task RawBytesHook_Throws_SessionSurvives() await client.GetStream().WriteAsync(Frame("first")); await client.GetStream().WriteAsync(Frame("second")); - var deadline = Environment.TickCount64 + 5000; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + 5000; + while (TestClock.TickCount64 < deadline) { lock (received) { diff --git a/test/StreamFrame.Tests/ConnectionUsabilityTests.cs b/test/StreamFrame.Tests/ConnectionUsabilityTests.cs index 844e072..c684c92 100644 --- a/test/StreamFrame.Tests/ConnectionUsabilityTests.cs +++ b/test/StreamFrame.Tests/ConnectionUsabilityTests.cs @@ -171,13 +171,13 @@ public async Task BoundedReceiveQueue_SlowConsumer_DeliversAllInOrder() await server.WaitForConnectedAsync().WaitAsync(TimeSpan.FromSeconds(5)); // 一次性灌入 5 帧（容量 2）：解码必须暂停等待消费、随后按序全部送达 - var buffer = new System.Buffers.ArrayBufferWriter(); + var buffer = new TestWrittenBufferWriter(); foreach (var i in Enumerable.Range(1, 5)) new LengthPrefixFramer().EncodeFrame(Encoding.UTF8.GetBytes($"msg-{i}"), buffer); await client.GetStream().WriteAsync(buffer.WrittenSpan.ToArray()); - var deadline = Environment.TickCount64 + 5000; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + 5000; + while (TestClock.TickCount64 < deadline) { lock (received) { diff --git a/test/StreamFrame.Tests/FrameDecoderTests.cs b/test/StreamFrame.Tests/FrameDecoderTests.cs index 731d5bb..6293d7e 100644 --- a/test/StreamFrame.Tests/FrameDecoderTests.cs +++ b/test/StreamFrame.Tests/FrameDecoderTests.cs @@ -16,7 +16,9 @@ private static IEnumerable Chunk(byte[] bytes, int chunkSize) for (var i = 0; i < bytes.Length; i += chunkSize) { var length = Math.Min(chunkSize, bytes.Length - i); - yield return bytes[i..(i + length)]; + var chunk = new byte[length]; + Array.Copy(bytes, i, chunk, 0, length); + yield return chunk; } } @@ -74,8 +76,8 @@ private static async Task WriteAllAsync(DecoderHarness h, byte[] bytes, int chun private static async Task WaitForMessagesAsync(DecoderHarness h, int expected, int timeoutMs = 5000) { - var deadline = Environment.TickCount64 + timeoutMs; - while (Environment.TickCount64 < deadline) + var deadline = TestClock.TickCount64 + timeoutMs; + while (TestClock.TickCount64 < deadline) { lock (h.Messages) { @@ -98,7 +100,7 @@ public async Task LengthPrefix_GluedFrames_AllDecoded() var p2 = "second-message-with-more-bytes"; var p3 = "third"; - var all = new ArrayBufferWriter(); + var all = new TestWrittenBufferWriter(); framing.EncodeFrame(Encoding.UTF8.GetBytes(p1), all); framing.EncodeFrame(Encoding.UTF8.GetBytes(p2), all); framing.EncodeFrame(Encoding.UTF8.GetBytes(p3), all); @@ -125,7 +127,7 @@ public async Task LengthPrefix_ChunkedFrames_HandlesHalfAndGluedPackets() var p2 = "B".PadRight(40, 'C'); var p3 = "hello world"; - var all = new ArrayBufferWriter(); + var all = new TestWrittenBufferWriter(); framing.EncodeFrame(Encoding.UTF8.GetBytes(p1), all); framing.EncodeFrame(Encoding.UTF8.GetBytes(p2), all); framing.EncodeFrame(Encoding.UTF8.GetBytes(p3), all); @@ -150,7 +152,7 @@ public async Task StxEtx_Chunked_AllDecoded() var p1 = "STX-frame-one"; var p2 = "STX-frame-two-with-longer-payload"; - var all = new ArrayBufferWriter(); + var all = new TestWrittenBufferWriter(); framing.EncodeFrame(Encoding.UTF8.GetBytes(p1), all); framing.EncodeFrame(Encoding.UTF8.GetBytes(p2), all); @@ -278,7 +280,7 @@ public async Task DecodeError_SkipFramePolicy_ContinuesWithNextFrame() var bad = Encoding.UTF8.GetBytes("bad"); var good = Encoding.UTF8.GetBytes("good"); - var all = new ArrayBufferWriter(); + var all = new TestWrittenBufferWriter(); framing.EncodeFrame(bad, all); // ThrowingDecodeCodec 只对 "bad" 抛 framing.EncodeFrame(good, all); await WriteAllAsync(h, all.WrittenSpan.ToArray()); diff --git a/test/StreamFrame.Tests/LengthPrefixFramerTests.cs b/test/StreamFrame.Tests/LengthPrefixFramerTests.cs index ac93efb..af63975 100644 --- a/test/StreamFrame.Tests/LengthPrefixFramerTests.cs +++ b/test/StreamFrame.Tests/LengthPrefixFramerTests.cs @@ -7,7 +7,7 @@ public class LengthPrefixFramerTests { private static (byte[] frame, ReadOnlySequence buffer) EncodeSingle(IFramer codec, byte[] payload) { - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); return (writer.WrittenSpan.ToArray(), new ReadOnlySequence(writer.WrittenSpan.ToArray())); } @@ -18,14 +18,14 @@ public void Encode_WritesBigEndianLengthHeader() var codec = new LengthPrefixFramer(); var payload = Encoding.UTF8.GetBytes("hello"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); var bytes = writer.WrittenSpan.ToArray(); Assert.Equal(9, bytes.Length); // 大端 0x00 0x00 0x00 0x05 - Assert.Equal(new byte[] { 0x00, 0x00, 0x00, 0x05 }, bytes[..4]); - Assert.Equal(payload, bytes[4..]); + Assert.Equal(new byte[] { 0x00, 0x00, 0x00, 0x05 }, bytes.Take(4).ToArray()); + Assert.Equal(payload, bytes.Skip(4).ToArray()); } [Fact] @@ -60,7 +60,7 @@ public void Decode_GluedFrames_ExtractsAll() var p1 = Encoding.UTF8.GetBytes("first"); var p2 = Encoding.UTF8.GetBytes("second-message"); - var all = new ArrayBufferWriter(); + var all = new TestWrittenBufferWriter(); codec.EncodeFrame(p1, all); codec.EncodeFrame(p2, all); var buffer = new ReadOnlySequence(all.WrittenSpan.ToArray()); diff --git a/test/StreamFrame.Tests/SingleClientConnectionTests.cs b/test/StreamFrame.Tests/SingleClientConnectionTests.cs index 2181b25..3736b85 100644 --- a/test/StreamFrame.Tests/SingleClientConnectionTests.cs +++ b/test/StreamFrame.Tests/SingleClientConnectionTests.cs @@ -40,7 +40,11 @@ private static async Task TryTcpConnectAsync(int port, int timeoutMs = 150 using var timeout = new CancellationTokenSource(timeoutMs); try { +#if NET48 + await client.ConnectAsync(IPAddress.Loopback.ToString(), port); // netfx 无 (IPAddress,int,ct) 重载；回环连接瞬时完成 +#else await client.ConnectAsync(IPAddress.Loopback, port, timeout.Token); +#endif return client.Connected; } catch (Exception) when (timeout.IsCancellationRequested) diff --git a/test/StreamFrame.Tests/StreamFrame.Tests.csproj b/test/StreamFrame.Tests/StreamFrame.Tests.csproj index 7e4b086..13a831b 100644 --- a/test/StreamFrame.Tests/StreamFrame.Tests.csproj +++ b/test/StreamFrame.Tests/StreamFrame.Tests.csproj @@ -1,9 +1,10 @@ - net8.0;net10.0 + net8.0;net10.0;net48 enable enable + latest false true @@ -11,6 +12,12 @@ + + runtime; build; native; contentfiles; analyzers; build_transitive + all + + + runtime; build; native; contentfiles; analyzers; buildtransitive all diff --git a/test/StreamFrame.Tests/StreamingFrameEncodeTests.cs b/test/StreamFrame.Tests/StreamingFrameEncodeTests.cs index 2f11050..e56b48c 100644 --- a/test/StreamFrame.Tests/StreamingFrameEncodeTests.cs +++ b/test/StreamFrame.Tests/StreamingFrameEncodeTests.cs @@ -23,7 +23,7 @@ private static byte[] EncodeStreaming(IStreamingFramer framing, byte[] payload) /// private static byte[] EncodePlain(IFramer framing, byte[] payload) { - var buffer = new ArrayBufferWriter(); + var buffer = new TestWrittenBufferWriter(); framing.EncodeFrame(payload, buffer); return buffer.WrittenSpan.ToArray(); } diff --git a/test/StreamFrame.Tests/StxEtxFramerTests.cs b/test/StreamFrame.Tests/StxEtxFramerTests.cs index 97cab87..5c7330f 100644 --- a/test/StreamFrame.Tests/StxEtxFramerTests.cs +++ b/test/StreamFrame.Tests/StxEtxFramerTests.cs @@ -14,13 +14,15 @@ public void Encode_WrapsPayload() var codec = new StxEtxFramer(); var payload = Encoding.UTF8.GetBytes("hello"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); var bytes = writer.WrittenSpan.ToArray(); Assert.Equal(STX, bytes[0]); - Assert.Equal(ETX, bytes[^1]); - Assert.Equal(payload, bytes[1..^1]); + Assert.Equal(ETX, bytes[bytes.Length - 1]); + var inner = new byte[bytes.Length - 2]; + Array.Copy(bytes, 1, inner, 0, inner.Length); + Assert.Equal(payload, inner); } [Fact] @@ -30,7 +32,7 @@ public void Encode_RejectsPayloadContainingStxOrEtx() // 负载含 0x02，plain 模式必须拒绝 Assert.Throws(() => { - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(new byte[] { 0x41, 0x02, 0x42 }, writer); }); } @@ -41,7 +43,7 @@ public void Decode_ExtractsPayload() var codec = new StxEtxFramer(); var payload = Encoding.UTF8.GetBytes("payload data"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); var buffer = new ReadOnlySequence(writer.WrittenSpan.ToArray()); @@ -56,7 +58,7 @@ public void Decode_HalfPacket_ReturnsFalse_AndKeepsFromStx() var codec = new StxEtxFramer(); var payload = Encoding.UTF8.GetBytes("hello"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); var buffer = new ReadOnlySequence(writer.WrittenSpan.ToArray()); @@ -72,7 +74,7 @@ public void Decode_LoneEtx_IsSkippedAsNoise() { var codec = new StxEtxFramer(); var payload = Encoding.UTF8.GetBytes("hi"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); // 前缀加孤立 ETX 噪声，再接正常帧 @@ -91,7 +93,7 @@ public void Decode_NewStx_DiscardsPreviousHalfPacket() { var codec = new StxEtxFramer(); var payload = Encoding.UTF8.GetBytes("complete"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.EncodeFrame(payload, writer); var full = writer.WrittenSpan.ToArray(); @@ -117,7 +119,7 @@ public void Decode_OverlongFrame_IsDropped() // 后面接一个正常短帧 var shortPayload = Encoding.UTF8.GetBytes("ok"); - var shortWriter = new ArrayBufferWriter(); + var shortWriter = new TestWrittenBufferWriter(); codec.EncodeFrame(shortPayload, shortWriter); bytes.AddRange(shortWriter.WrittenSpan.ToArray()); diff --git a/test/StreamFrame.Tests/TestCompat.cs b/test/StreamFrame.Tests/TestCompat.cs new file mode 100644 index 0000000..47c22cb --- /dev/null +++ b/test/StreamFrame.Tests/TestCompat.cs @@ -0,0 +1,52 @@ +using System.Buffers; +using System.IO; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace StreamFrame.Tests; + +/// net48 缺失 API 的测试侧 polyfill（net8/net10 走运行时内置）。 +internal static class TestClock +{ + public static long TickCount64 => +#if NET48 + Environment.TickCount; // netfx 仅有 int 版（24.9 天回绕，测试时长内无碍） +#else + Environment.TickCount64; +#endif +} + +#if NET48 +internal static class TestTaskWaitAsyncCompat +{ + public static async Task WaitAsync(this Task task, TimeSpan timeout) + { + var completed = await Task.WhenAny(task, Task.Delay(timeout)).ConfigureAwait(false); + if (completed != task) + throw new TimeoutException($"等待超过 {timeout}。"); + await task.ConfigureAwait(false); + } + + public static async Task WaitAsync(this Task task, TimeSpan timeout, CancellationToken cancellationToken) + { + var delay = Task.Delay(timeout, cancellationToken); + var completed = await Task.WhenAny(task, delay).ConfigureAwait(false); + if (completed != task) + { + delay.Ignore(); + if (cancellationToken.IsCancellationRequested) + throw new OperationCanceledException(cancellationToken); + throw new TimeoutException($"等待超过 {timeout}。"); + } + await task.ConfigureAwait(false); + } +} + +internal static class TestTaskExtensions +{ + public static void Ignore(this Task _) + { + } +} +#endif diff --git a/test/StreamFrame.Tests/TestCompat.net48.cs b/test/StreamFrame.Tests/TestCompat.net48.cs new file mode 100644 index 0000000..5a0fe19 --- /dev/null +++ b/test/StreamFrame.Tests/TestCompat.net48.cs @@ -0,0 +1,54 @@ +#if NET48 +using System.Buffers; +using System.IO; +using System.Text; +using System.Threading.Tasks; + +// 块式命名空间：一个文件多个命名空间（文件级命名空间仅限单个） +namespace System.Text +{ + internal static class EncodingSequenceCompat + { + public static string GetString(this Encoding encoding, in ReadOnlySequence bytes) + => bytes.IsSingleSegment + ? encoding.GetString(bytes.First.Span.ToArray()) + : encoding.GetString(bytes.ToArray()); + + public static string GetString(this Encoding encoding, ReadOnlySpan bytes) + => encoding.GetString(bytes.ToArray()); + } +} + +namespace System.IO +{ + internal static class TestStreamCompat + { + // netfx 的 Stream 没有单参（ReadOnlyMemory）的 WriteAsync 重载 + public static Task WriteAsync(this Stream stream, byte[] buffer) + => stream.WriteAsync(buffer, 0, buffer.Length); + } +} +#endif + +#if NET48 +namespace System.Threading.Channels +{ + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + + internal static class TestChannelCompat + { + // netfx 的 Channels 包没有 ReadAllAsync（netcore 为实例方法，优先于本扩展） + public static async IAsyncEnumerable ReadAllAsync( + this ChannelReader reader, [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default) + { + while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) + { + while (reader.TryRead(out var item)) + yield return item; + } + } + } +} +#endif diff --git a/test/StreamFrame.Tests/XmlDocumentCodecTests.cs b/test/StreamFrame.Tests/XmlDocumentCodecTests.cs index f2a1f02..a6db032 100644 --- a/test/StreamFrame.Tests/XmlDocumentCodecTests.cs +++ b/test/StreamFrame.Tests/XmlDocumentCodecTests.cs @@ -13,7 +13,7 @@ public void EncodeThenDecode_RoundTripsDocument() var codec = new XmlDocumentCodec(); var doc = XDocument.Parse("42DeviceA"); - var writer = new ArrayBufferWriter(); + var writer = new TestWrittenBufferWriter(); codec.Encode(doc, writer); // 编码结果应包含 XML 内容