Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,10 +6,11 @@
## [Unreleased]

### 新增
- 暂无
- 心跳保活示例(demo 场景 4):周期心跳 + 接收空闲超时的完整范式——有心跳时连接稳定(零状态事件),停止心跳后空闲判定触发重连;README 活性探测小节同步补充代码片段。

### 修复
- 暂无
- 接收空闲超时在部分平台(Windows)会把取消中的接收折算成 0 字节完成,被误判为"对端正常关闭"(FIN)——现统一按会话故障处理(结果同为重连,但诊断语义正确;最小复现确认后修复)。
- demo 场景 3 强制重连后用 `WaitForConnectedAsync` 等待双方就绪存在竞态(旧会话瞬时仍为 Connected 会走"立即完成"快速路径)——改为轮询双方状态,连续 10 次运行无抖动。

## [2.2.0] - 2026-08-27

Expand Down
21 changes: 19 additions & 2 deletions README.en.md
Original file line number Diff line number Diff line change
Expand Up @@ -143,9 +143,26 @@ var conn = new StreamConnection<XDocument>(..., 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

Expand Down
20 changes: 18 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -143,9 +143,25 @@ var conn = new StreamConnection<XDocument>(..., 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`。

## 性能

Expand Down
118 changes: 109 additions & 9 deletions samples/StreamFrame.Demo/Program.cs
Original file line number Diff line number Diff line change
Expand Up @@ -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) =>
Expand All @@ -21,6 +23,7 @@
await RunXmlLengthPrefixScenarioAsync(cts.Token);
await RunStxEtxTextScenarioAsync(cts.Token);
await RunReconnectScenarioAsync(cts.Token);
await RunHeartbeatScenarioAsync(cts.Token);

Console.WriteLine("演示结束。");

Expand Down Expand Up @@ -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})");
Expand All @@ -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<string>(
new LengthPrefixFramer(), new Utf8TextCodec(),
IPAddress.Loopback, port, isActive: false, options);
var client = new StreamConnection<string>(
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<ConnectionState>();
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<ConnectionState>();
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();
}

// ---------------------------------------------------------------------------
// 辅助
// ---------------------------------------------------------------------------
Expand Down
10 changes: 9 additions & 1 deletion src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -600,7 +600,15 @@ private async Task<int> ReceiveWithIdleTimeoutAsync(Socket socket, Memory<byte>
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)
{
Expand Down
Loading