From 7804370c4402128e728764f9b07f7647b1dc361a Mon Sep 17 00:00:00 2001 From: Marci Date: Thu, 27 Aug 2026 23:22:42 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=BF=83=E8=B7=B3=E4=BF=9D=E6=B4=BB?= =?UTF-8?q?=E7=A4=BA=E4=BE=8B=EF=BC=88demo=20=E5=9C=BA=E6=99=AF=204=20+=20?= =?UTF-8?q?README=20=E8=8C=83=E5=BC=8F=EF=BC=89=EF=BC=9B=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=E7=A9=BA=E9=97=B2=E8=B6=85=E6=97=B6=200=20=E5=AD=97=E8=8A=82?= =?UTF-8?q?=E5=AE=8C=E6=88=90=E8=AF=AF=E5=88=A4=E4=B8=8E=20demo=20?= =?UTF-8?q?=E5=9C=BA=E6=99=AF=203=20=E7=AB=9E=E6=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 5 +- README.en.md | 21 +++- README.md | 20 ++- samples/StreamFrame.Demo/Program.cs | 118 ++++++++++++++++-- .../Connection/StreamConnection.cs | 10 +- 5 files changed, 158 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35f262b..f6f074a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,10 +6,11 @@ ## [Unreleased] ### 新增 -- 暂无 +- 心跳保活示例（demo 场景 4）：周期心跳 + 接收空闲超时的完整范式——有心跳时连接稳定（零状态事件），停止心跳后空闲判定触发重连；README 活性探测小节同步补充代码片段。 ### 修复 -- 暂无 +- 接收空闲超时在部分平台（Windows）会把取消中的接收折算成 0 字节完成，被误判为"对端正常关闭"（FIN）——现统一按会话故障处理（结果同为重连，但诊断语义正确；最小复现确认后修复）。 +- demo 场景 3 强制重连后用 `WaitForConnectedAsync` 等待双方就绪存在竞态（旧会话瞬时仍为 Connected 会走"立即完成"快速路径）——改为轮询双方状态，连续 10 次运行无抖动。 ## [2.2.0] - 2026-08-27 diff --git a/README.en.md b/README.en.md index ef6b652..e910439 100644 --- a/README.en.md +++ b/README.en.md @@ -143,9 +143,26 @@ var conn = new StreamConnection(..., logger: loggerFactory.CreateLogg Without one, no logging happens (zero-dependency usage). -### Liveness tips +### Liveness & heartbeat pattern -For production, enable `TcpKeepAlive = true`; for protocols with periodic traffic, also consider `ReceiveIdleTimeoutMs` (e.g., 3× the heartbeat period) as a second safety net against half-open connections. +For production, enable `TcpKeepAlive = true`; an application-level heartbeat combined with `ReceiveIdleTimeoutMs` (3× the heartbeat period, tolerating 1-2 lost beats) is an even stronger combo. The framework does not ship a heartbeat (the message shape belongs to your protocol) — the pattern: + +```csharp +var options = new StreamConnectionOptions { ReceiveIdleTimeoutMs = 1500 }; + +// One side beats periodically; the other echoes PONG on any message (bidirectional +// traffic resets both idle timers). Full runnable sample: demo scenario 4. +_ = Task.Run(async () => +{ + while (!cts.IsCancellationRequested) + { + await client.SendAsync("PING", cts.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), cts.Token); // ~1/3 of the timeout + } +}); +``` + +When the peer dies silently (power loss / unplugged cable, no FIN/RST), the idle timeout kills the session and auto-reconnect kicks in; if the peer is unreachable, reconnection keeps failing and the state stays in `Connecting`/`Retry`. ## Performance diff --git a/README.md b/README.md index 237eea9..6543010 100644 --- a/README.md +++ b/README.md @@ -143,9 +143,25 @@ var conn = new StreamConnection(..., logger: loggerFactory.CreateLogg 不传则无日志输出（零依赖可用）。 -### 活性探测建议 +### 活性探测与心跳范式 -生产环境建议开启 `TcpKeepAlive = true`；有周期性报文的协议可再加 `ReceiveIdleTimeoutMs`（如心跳周期的 3 倍），双保险兜底半开连接。 +生产环境建议开启 `TcpKeepAlive = true`；应用层心跳配合 `ReceiveIdleTimeoutMs`（取心跳周期的 3 倍，容忍偶尔丢 1-2 次）是更强的组合——框架不内置心跳（消息形态由协议决定），范式如下，完整可运行示例见 demo 场景 4： + +```csharp +var options = new StreamConnectionOptions { ReceiveIdleTimeoutMs = 1500 }; + +// 一侧周期发心跳；另一侧收到任何消息回 PONG（双向都有字节即可重置双方空闲计时） +_ = Task.Run(async () => +{ + while (!cts.IsCancellationRequested) + { + await client.SendAsync("PING", cts.Token); + await Task.Delay(TimeSpan.FromMilliseconds(500), cts.Token); // 周期 ≈ 超时的 1/3 + } +}); +``` + +对端"猝死"（断电/拔线，无 FIN/RST）时静默超限 → 会话判定死亡 → 自动重连；对端不可达时重连持续失败，状态停留在 `Connecting`/`Retry`。 ## 性能 diff --git a/samples/StreamFrame.Demo/Program.cs b/samples/StreamFrame.Demo/Program.cs index 7bf53d4..66b494a 100644 --- a/samples/StreamFrame.Demo/Program.cs +++ b/samples/StreamFrame.Demo/Program.cs @@ -5,10 +5,12 @@ using StreamFrame; using StreamFrame.Protocols.Xml; -// 本 demo 演示 StreamFrame 的三种帧/编解码组合： +// 本 demo 演示 StreamFrame 的四种典型用法： // 1) XML 消息 + 4 字节长度前缀帧（对应 SamSung 风格） // 2) 纯文本消息 + STX/ETX 包裹帧 -// 两种场景共用同一套连接核心，仅 framing 与 codec 不同。 +// 3) 断线自动重连 +// 4) 心跳保活 + 接收空闲超时（半开连接探测） +// 所有场景共用同一套连接核心，仅 framing / codec / 选项不同。 var cts = new CancellationTokenSource(); Console.CancelKeyPress += (_, e) => @@ -21,6 +23,7 @@ await RunXmlLengthPrefixScenarioAsync(cts.Token); await RunStxEtxTextScenarioAsync(cts.Token); await RunReconnectScenarioAsync(cts.Token); +await RunHeartbeatScenarioAsync(cts.Token); Console.WriteLine("演示结束。"); @@ -250,14 +253,13 @@ int ConnectingCount() Console.WriteLine("\n主动端强制重连，模拟链路中断…"); client.Reconnect(); - try - { - await Task.WhenAll( - client.WaitForConnectedAsync(ct).WaitAsync(TimeSpan.FromSeconds(8), ct), - server.WaitForConnectedAsync(ct).WaitAsync(TimeSpan.FromSeconds(8), ct)); - } - catch (TimeoutException) + // 注意不能用 WaitForConnectedAsync 等"重连"：发起强制重连的瞬间旧会话仍是 Connected， + // 该调用会走"已连接立即完成"的快速路径。这里轮询双方状态直到都回到 Connected。 + var reconnectDeadline = DateTime.UtcNow.AddSeconds(15); + while (!(client.State == ConnectionState.Connected && server.State == ConnectionState.Connected) + && DateTime.UtcNow < reconnectDeadline && !ct.IsCancellationRequested) { + await Task.Delay(50, ct); } Assert(client.State == ConnectionState.Connected && server.State == ConnectionState.Connected, $"重连后双方应回到 Connected（client={client.State}, server={server.State}）"); @@ -272,6 +274,104 @@ await Task.WhenAll( await client.DisposeAsync(); } +// --------------------------------------------------------------------------- +// 场景 4：心跳保活 + 接收空闲超时 +// --------------------------------------------------------------------------- +// 心跳范式（协议无关，框架不内置心跳——消息形态由驱动决定）： +// 1) 一侧周期性发送心跳报文（周期取空闲超时的 1/3 左右） +// 2) 双方开启 ReceiveIdleTimeoutMs（= 3× 心跳周期，容忍丢 1-2 次） +// 3) 对端回 PONG（或任何业务流量），双向都有字节即可重置空闲计时 +// 4) 对端“猝死”（断电/拔线，无 FIN/RST）时，静默超时判定连接死亡并自动重连 +static async Task RunHeartbeatScenarioAsync(CancellationToken ct) +{ + const int port = 5400; + Console.WriteLine("\n=== 场景 4：心跳保活 + 接收空闲超时 ==="); + + var heartbeatInterval = TimeSpan.FromMilliseconds(500); + var options = new StreamConnectionOptions + { + ReceiveIdleTimeoutMs = 1500, // = 3× 心跳周期：容忍偶尔丢 1-2 次心跳 + ConnectRetryDelayMs = 500, + AcceptRetryDelayMs = 500, + }; + + var server = new StreamConnection( + new LengthPrefixFramer(), new Utf8TextCodec(), + IPAddress.Loopback, port, isActive: false, options); + var client = new StreamConnection( + new LengthPrefixFramer(), new Utf8TextCodec(), + IPAddress.Loopback, port, isActive: true, options); + + // 服务端：收到任何消息回 PONG（维持双向流量，双方空闲计时都被重置） + _ = Task.Run(async () => + { + await foreach (var message in server.GetMessages(ct)) + await server.SendAsync($"PONG:{message}", ct); + }, ct); + + // 客户端：周期心跳 + var heartbeatCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + var heartbeatTask = Task.Run(async () => + { + try + { + while (!heartbeatCts.IsCancellationRequested) + { + await client.SendAsync("PING", heartbeatCts.Token); + Console.WriteLine("[Cli4] PING"); + await Task.Delay(heartbeatInterval, heartbeatCts.Token); + } + } + catch (OperationCanceledException) + { + } + }, ct); + + client.Start(ct); + server.Start(ct); + await Task.WhenAll(client.WaitForConnectedAsync(ct), server.WaitForConnectedAsync(ct)); + + // 心跳正常运行 2.2 秒（≈ 4 个周期）：双方保持 Connected，且零状态事件（连接稳定） + var stableStates = new List(); + client.ConnectionChanged += (_, s) => { lock (stableStates) stableStates.Add(s); }; + await Task.Delay(2200, ct); + Assert(client.State == ConnectionState.Connected && server.State == ConnectionState.Connected, + $"心跳期间双方应保持 Connected（client={client.State}, server={server.State}）"); + lock (stableStates) + Assert(stableStates.Count == 0, $"心跳期间不应有任何状态事件，实际 {stableStates.Count} 次"); + Console.WriteLine("心跳保活正常 ✓（连接稳定，零状态事件）"); + + // 模拟对端“猝死”：停止心跳且完全静默（连接仍在，只是没有字节） + heartbeatCts.Cancel(); + await heartbeatTask; + Console.WriteLine("已静默，等待空闲超时…"); + + // 静默超过 ReceiveIdleTimeoutMs 后，接收侧判定会话死亡并进入重连： + // 双方进程都活着时重连会立刻成功，状态表现为“翻动”（Retry/Connecting/Connected 循环） + // 而非停在断开——观察 ConnectionChanged 事件即可证明空闲检测生效。 + // （真实半开场景——对端断电/不可达——重连会持续失败，状态停留在 Connecting/Retry） + var flapStates = new List(); + client.ConnectionChanged += (_, s) => { lock (flapStates) flapStates.Add(s); }; + var deadline = DateTime.UtcNow.AddSeconds(6); + while (DateTime.UtcNow < deadline && !ct.IsCancellationRequested) + { + lock (flapStates) + { + if (flapStates.Count >= 2) + break; + } + await Task.Delay(100, ct); + } + lock (flapStates) + Assert(flapStates.Count >= 2, + $"静默后应触发空闲判定与重连（至少 2 次状态事件），实际 {flapStates.Count} 次：[{string.Join(", ", flapStates)}]"); + Console.WriteLine($"空闲超时生效 ✓（触发的状态序列：[{string.Join(", ", flapStates)}]，进入重连循环）"); + + Console.WriteLine("场景 4 通过 ✓"); + await server.DisposeAsync(); + await client.DisposeAsync(); +} + // --------------------------------------------------------------------------- // 辅助 // --------------------------------------------------------------------------- diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs index b90153d..7b23115 100644 --- a/src/StreamFrame/Connection/StreamConnection.cs +++ b/src/StreamFrame/Connection/StreamConnection.cs @@ -600,7 +600,15 @@ private async Task ReceiveWithIdleTimeoutAsync(Socket socket, Memory linked.CancelAfter(_options.ReceiveIdleTimeoutMs); try { - return await socket.ReceiveAsync(memory, SocketFlags.None, linked.Token).ConfigureAwait(false); + var count = await socket.ReceiveAsync(memory, SocketFlags.None, linked.Token).ConfigureAwait(false); + if (count == 0 && linked.IsCancellationRequested) + { + // 部分平台（Windows）把取消中的接收折算成 0 字节完成而非抛取消异常—— + // 不加区分会误判为对端正常关闭（FIN），语义应为空闲超时会话故障 + throw new SessionFaultException($"接收空闲超过 {_options.ReceiveIdleTimeoutMs}ms，判定连接死亡。"); + } + + return count; } catch (OperationCanceledException) when (!ct.IsCancellationRequested) {