From a17e7db2579742fdfe28683ffa67a77b681c0650 Mon Sep 17 00:00:00 2001 From: Marci Date: Fri, 28 Aug 2026 22:45:06 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8D=87=E7=BA=A7=20StreamFrame=202.3.?= =?UTF-8?q?0=20=E4=BC=9A=E8=AF=9D=E9=80=82=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 14 +- Directory.Packages.props | 2 +- docs/ARCHITECTURE.md | 21 +- docs/COMPATIBILITY.md | 14 +- docs/ROADMAP.md | 8 +- docs/STREAMFRAME-ADAPTER.md | 137 +++---- .../Hsms/Connection/HsmsConnectionOptions.cs | 18 +- .../HsmsStreamConnectionOptionsAdapter.cs | 1 + src/SecsFrame/Transport/HsmsT8Monitor.cs | 163 --------- src/SecsFrame/Transport/HsmsTransportFrame.cs | 8 - .../Transport/HsmsTransportOptions.cs | 20 ++ .../Transport/HsmsTransportSessionContext.cs | 65 ---- .../HsmsTransportSessionExpiredException.cs | 7 +- .../Transport/SessionBoundHsmsFrameCodec.cs | 29 -- .../Transport/StreamFrameHsmsTransport.cs | 276 +++++++------- .../HsmsConnectionOptionsTests.cs | 18 + ...HsmsStreamConnectionOptionsAdapterTests.cs | 27 +- test/SecsFrame.Tests/HsmsT8MonitorTests.cs | 157 -------- .../SessionBoundHsmsFrameCodecTests.cs | 49 --- .../StreamFrameHsmsTransportTests.cs | 337 ++++++++++-------- 20 files changed, 510 insertions(+), 861 deletions(-) delete mode 100644 src/SecsFrame/Transport/HsmsT8Monitor.cs delete mode 100644 src/SecsFrame/Transport/HsmsTransportFrame.cs delete mode 100644 src/SecsFrame/Transport/HsmsTransportSessionContext.cs delete mode 100644 src/SecsFrame/Transport/SessionBoundHsmsFrameCodec.cs delete mode 100644 test/SecsFrame.Tests/HsmsT8MonitorTests.cs delete mode 100644 test/SecsFrame.Tests/SessionBoundHsmsFrameCodecTests.cs diff --git a/CHANGELOG.md b/CHANGELOG.md index 368496a..5ea69d6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,9 +12,13 @@ - 增加全部 Item 格式、长度边界、嵌套 List、分段输入、非法输入和往返协议向量测试。 - 增加动态 SecsMessage、HSMS Data Message 信封与可选单根 Item 的 Payload 编解码。 - 增加完整线上帧、无 Body、空 List、分段输入、非法 SType/PType 和消息往返测试。 -- 增加 StreamFrame 内部会话适配器，提供单调 Session ID、会话失效通知和旧会话发送拒绝。 -- 增加基于实际 Socket 写出分片的整帧发送确认，以及可替换计时器驱动的未完成帧超时。 -- 增加发送竞态、迟到接收、超时重置和 StreamFrame 真实 TCP 回环测试。 +- 升级官方 StreamFrame 2.3.0，采用原生单调 Session ID、会话失效通知、 + 会话绑定发送和未完成帧超时，删除 #38/#39 对应的临时回调适配。 +- 发送任务由 StreamFrame 在整帧写入本机 Socket 后完成，旧会话消息不会 + 跨重连重放；T8 FrameError 映射为保留 transport Session ID 的诊断。 +- T8 配置改为要求可由正整毫秒精确表示且不超过 int.MaxValue + 毫秒；亚毫秒、截断和溢出输入在公共选项边界立即拒绝。 +- 增加发送竞态、迟到接收、原生 T8 映射和 StreamFrame 真实 TCP 回环测试。 - 增加 Active/Passive 共用的内部 HSMS-SS Select/Separate 状态机， 分离 TCP Connected、Selecting 与 Selected 状态。 - 增加从整帧实际写出后启动的 T6、连接建立后启动的 T7、System Bytes @@ -40,8 +44,8 @@ Active/Passive 双端真实 TCP 数据事务测试。 - 增加显式 HSMS T5/T8 运输配置；Active 连接把 T5 无损映射为固定 StreamFrame 重试间隔并关闭指数退避，Passive 监听重试保持独立。 -- 增加按接收进展与会话代次隔离的 T8 监视、专用超时异常，以及部分长度 - 头、部分 Payload、陈旧回调、替换会话和真实 TCP 超时测试。 +- 增加 StreamFrame 原生接收进展 T8、专用超时异常及关闭原因关联，覆盖 + 部分长度头、替换会话和真实 TCP 超时测试。 - 增加公共 HsmsConnection 与不可变 HsmsConnectionOptions，组合动态消息收发、Selected 等待、 单消费者事件流、入站一次性回复和控制命令。 diff --git a/Directory.Packages.props b/Directory.Packages.props index e8e3141..f48ff7f 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -9,7 +9,7 @@ - + diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 40d0d41..37ef696 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -53,18 +53,17 @@ HSMS 会话切换后，旧会话的控制请求、数据事务和等待中的应 IHsmsTransport 使用单一事件流按 Session ID 报告会话打开、 帧到达和会话关闭。StreamFrameHsmsTransport 已在内部实现： -- 每次 TCP Connected 生成单调递增、不可复用的 Session ID； -- 收到的帧在 codec 解码时绑定 Session ID，迟到消费不会被标成新会话； -- 发送信封在编码前再次校验 Session ID，旧会话消息不会上线； -- 同一时间只向 StreamFrame 提交一帧，并根据 RawBytesSent 的实际分片 - 累计确认，整帧写完后才完成发送； +- 直接采用 StreamFrame 2.3.0 的单调 TCP Session ID； +- 通过会话感知接收流保留帧的原始 Session ID，迟到消费不会被标成新会话； +- 会话绑定发送在 StreamFrame 原生 FIFO 中校验 Session ID，旧会话消息 + 不会重放，整帧写入本机 Socket 后才完成； - Active 连接把显式 T5 映射为固定 StreamFrame 重试间隔，关闭指数退避； -- 原始接收字节独立跟踪长度前缀和剩余 Payload，仅有未完成帧时运行 T8， - 并用代次标识隔离接收进展与替换会话的陈旧回调。 +- 显式 T8 映射为 StreamFrame 原生未完成帧超时，并把对应 FrameError + 转换为带 transport Session ID 的 HSMS 关闭原因。 -这些能力隔离当前 StreamFrame 2.2.0 尚未提供的高级语义，不进入公共 -API。#38 已合入上游但尚未发布，#39 正在实现；正式包可用后只替换 -IHsmsTransport 实现。详细失效模式与替换条件见 +这些能力仍隔离在 IHsmsTransport 后，不进入公共 API。此前 +跟踪的 StreamFrame #38/#39 已在 2.3.0 完成迁移；SecsFrame 只保留 +协议关闭原因和异常边界转换。详细失效模式与验证证据见 [STREAMFRAME-ADAPTER.md](STREAMFRAME-ADAPTER.md)。 ### 单线程会话状态机 @@ -136,7 +135,7 @@ HSMS 头 Session ID 与 System Bytes。没有 W-Bit 的出站消息在写出后 组合层。它要求显式网络、协议 Session ID 和 T3/T5/T6/T7/T8 配置，公开 动态消息发送、一次性入站回复、控制命令、Selected 等待和单消费者事件 流。公共层不暴露 transport Session generation、StreamFrame -StreamConnectionOptions 或 #38/#39 的回调适配。 +StreamConnectionOptions 或 StreamFrame 的会话感知接口。 连接内部独占消费事务事件，并分别更新状态等待信号与公共事件 channel， 因此 readiness 等待不会与业务消息消费者竞争。详细生命周期和取消语义 diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index e953608..1eb60e8 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -17,19 +17,19 @@ 这些条目只是调查清单，不代表当前实现已经支持。 -## 上游能力适配 +## 上游能力集成 -StreamFrame #38（未完成帧超时）已经合入上游 main，但尚未 -进入正式 NuGet 版本；#39（会话感知发送确认与消息上下文）正在实现。 -SecsFrame 当前固定使用 StreamFrame 2.2.0，并通过内部 -StreamFrameHsmsTransport 补足对应语义。 +StreamFrame #38（未完成帧超时）与 #39（会话感知发送确认及消息上下文） +已经随官方 StreamFrame 2.3.0 发布。SecsFrame 固定该 NuGet 版本，并由 +内部 StreamFrameHsmsTransport 使用原生会话 API；此前基于 +原始字节回调、发送信封和自建 Session ID 的临时实现已经删除。 这不是设备兼容模式： - 不放宽协议输入，也不接受标准严格模式本应拒绝的数据； - 不提供默认关闭的厂商开关； -- 不暴露为公共 API，不形成长期兼容承诺； -- 上游能力可用并完成等价测试后，直接替换内部实现。 +- 不暴露 StreamFrame 类型为公共 API，不形成跨层兼容承诺； +- SecsFrame 仅保留 HSMS 关闭原因、异常和事件边界转换。 具体机制、测试证据和已知代价见 [STREAMFRAME-ADAPTER.md](STREAMFRAME-ADAPTER.md)。 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 08a0469..ea4aaff 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -10,12 +10,12 @@ - [x] 完整 SECS-II Item 类型、嵌套 List 和严格二进制编解码。 - [x] 动态 SecsMessage、可选单根 Item 与 HSMS Data Message Payload 编解码。 -- [x] StreamFrame #38/#39 的内部会话绑定、写出确认和未完成帧超时适配。 +- [x] 迁移到 StreamFrame 2.3.0 原生会话绑定、写出确认和未完成帧超时。 - [x] Active/Passive 共用状态机的 Select/Select Response、Separate、 T6/T7、System Bytes 关联和会话失效处理。 - [x] Linktest、Reject、Deselect 控制消息与对应 T6 分支。 - [x] Selected 数据发送、Primary/Secondary 关联、T3 与入站同事务回复。 -- [x] 显式 T5 固定重连节流、T8 接收进展监视和跨会话旧回调隔离。 +- [x] 显式 T5 固定重连节流、原生 T8 映射和跨会话故障隔离。 - [x] 最小公共连接外观、显式 T3/T5/T6/T7/T8 配置、动态消息事件与 确定的取消/释放语义。 - [x] 使用 nuget.org 官方 Secs4Net 3.1.0 的 Active/Passive、Select、 @@ -25,8 +25,8 @@ - [ ] 使用授权 E37/E37.1 核对 T5/T8 默认值与精确启停边界。 - [x] Host、Equipment 双角色端点及两种 Active/Passive 拓扑的双向回环。 -阶段 1 的工程能力已经形成完整基础链路。下一切片进入 GEM 通用行为，同时 -继续跟踪 StreamFrame #38/#39 的正式包发布并准备等价迁移。任何标准默认值 +阶段 1 的工程能力已经形成完整基础链路，StreamFrame #38/#39 的正式包 +迁移与等价测试已经完成。后续切片继续扩展 GEM 通用行为。任何标准默认值 和完整合规结论都依赖团队合法获得的 E37/E37.1 及相关 GEM 标准副本与 一致性测试。 diff --git a/docs/STREAMFRAME-ADAPTER.md b/docs/STREAMFRAME-ADAPTER.md index dcaada8..d16b029 100644 --- a/docs/STREAMFRAME-ADAPTER.md +++ b/docs/STREAMFRAME-ADAPTER.md @@ -1,91 +1,70 @@ -# StreamFrame 内部适配 +# StreamFrame 会话适配 -## 背景 +## 上游基线 -SecsFrame 当前依赖 StreamFrame 2.2.0。以下上游能力仍在跟踪： +SecsFrame 固定依赖官方 [StreamFrame 2.3.0](https://www.nuget.org/packages/StreamFrame/2.3.0)。 +该版本已经发布并解决此前跟踪的两项能力缺口： -- [StreamFrame #38](https://github.com/CSJ608/StreamFrame/issues/38)： - 未完成帧超时已经合入上游 main，但尚未进入正式包； -- [StreamFrame #39](https://github.com/CSJ608/StreamFrame/issues/39)： - 会话感知的发送确认与消息上下文正在实现。 +- [#38](https://github.com/CSJ608/StreamFrame/issues/38) / [PR #41](https://github.com/CSJ608/StreamFrame/pull/41)： + 未完成帧超时与 IncompleteFrameTimeout 诊断； +- [#39](https://github.com/CSJ608/StreamFrame/issues/39) / [PR #42](https://github.com/CSJ608/StreamFrame/pull/42)： + 会话编号、会话绑定发送、写出确认和带会话归属的接收消息。 -StreamFrame 现有 SendAsync 在消息进入连接级队列后完成，队列 -会跨 TCP 重连保留；GetMessages 也是跨重连的稳定消息流。 -这些语义适合通用业务连接，但不足以支撑 HSMS 的严格会话边界。因此 -SecsFrame 在内部 IHsmsTransport 后提供临时适配，公共 API -不依赖 StreamFrame 的队列或原始字节回调。 +SecsFrame 不再维护基于原始字节回调的 T8 监视器、发送确认计数器、会话 +信封 codec 或自建 TCP Session ID。公共 API 和内部 +IHsmsTransport 契约保持不变，上层 HSMS 状态机不直接依赖 +StreamFrame 类型。 -## 会话事件 +## 会话边界 -内部传输输出同一条有序事件流： +StreamFrameHsmsTransport 只消费 +ISessionAwareStreamConnection<HsmsFrame>： -- SessionOpened：TCP Connected 后分配单调递增 Session ID； -- FrameReceived：携带 codec 解码时绑定的 Session ID； -- SessionClosed：会话失效，超时时附带原因。 +- CurrentSessionId 作为 TCP 会话的单调编号； +- GetSessionMessages 保留帧的原始会话归属，迟到消费不会被 + 标记为新会话； +- SendInSessionAsync 把帧绑定到指定会话，旧会话帧不会在 + 重连后重放； +- 发送任务仅在整帧写入本机 Socket 后完成，T3/T6 因而仍从明确的写出 + 完成点启动。 -接收帧不是在业务消费时读取“当前会话”，因此旧解码任务的迟到结果仍 -保留旧 Session ID。状态机可以按 ID 丢弃已关闭会话的事件。 -状态机还可以通过 TryCloseSession 只关闭匹配的当前 Session -ID，并把协议错误或 T6/T7 超时原因带入 SessionClosed。 +GetMessages 与 GetSessionMessages 是同一接收通道 +的竞争消费视图，适配器只使用后者。发送中途失败时，远端是否收到部分字节 +仍是未知状态；上层必须依靠事务关联、超时和业务幂等恢复，不能假设远端 +一定没有收到。 -## 发送确认与不重放 +StreamFrame 的 SessionExpiredException 在内部边界转换为 +HsmsTransportSessionExpiredException，并保留原异常。协议 +主动关闭或 T8 关闭时，适配器按 TCP Session ID 保留首个关闭原因，使 +SessionClosed 与尚未完成的发送观察到同一个协议异常。 -适配器的发送路径一次只允许一个待确认帧： - -1. 入队前确认调用方 Session ID 仍是当前会话； -2. 信封进入 StreamFrame codec 时再次确认，关闭会话的信封编码失败； -3. 根据 RawBytesSent 每次实际 Socket 写出的字节数累计； -4. 四字节长度前缀、十字节头和 Body 全部写完后完成发送任务； -5. 会话先关闭时，以明确的会话失效异常完成等待任务。 - -因此会话层 T6 与事务层 T3 都从发送任务成功完成后启动。调用取消若发生 -在进入 StreamFrame 队列之前会取消发送；一旦成功入队，适配器等待明确 -的写完或会话失效结果，避免向调用方返回含糊状态。 - -旧信封即使因极窄竞态留在 StreamFrame 的跨重连队列中，也会在新会话 -编码前失败，绝不会写入 Socket。当前 fail-closed 代价是该编码失败会让 -StreamFrame 再重建一次连接；上游 #39 提供原生会话队列后可消除此代价。 - -## T5 连接重试 +## T5 与 T8 内部 HsmsTransportOptions 要求调用方显式给出 T5 与 T8， -不内置未经授权标准核对的默认值。Active 连接通过独立选项副本把 T5 -无损转换为 StreamFrame 的整毫秒连接重试间隔，并把最大退避设置为零， -使连续失败保持固定间隔。原 StreamConnectionOptions 不会 -被修改，其余队列、缓冲、KeepAlive 和解码选项逐项保留。 - -Passive 端的监听失败重试不是当前适配器定义的 T5 行为，因此不会覆盖 -AcceptRetryDelayMs 或最大退避设置。T5 必须为正值、可由 -整毫秒精确表示，并且不超过 StreamFrame 2.2.0 的 int -毫秒范围；适配器拒绝截断、舍入和溢出。 - -## T8 接收超时 - -适配器直接观察 StreamFrame 的原始接收分片，并独立跟踪 HSMS 四字节 -大端长度前缀与剩余 Payload： - -- 从未收到字节或完整帧后的空闲期不计时； -- 收到部分长度头或部分 Payload 后启动计时； -- 每次收到后续字节重新计时； -- 当前帧完整且没有下一帧残片时停止计时； -- 超时触发当前连接重建，并用携带 transport Session ID 的 - HsmsT8TimeoutException 关闭会话。 - -每次接收进展都创建带代次标识的新计时器。旧计时器即使已经排队、在 -Dispose 后仍执行回调，也不能使后续进展或替换 TCP 会话超时。测试使用 -手动计时器覆盖部分长度头、部分 Payload、进度重置、完整帧后空闲、连续 -完整帧、下一帧残片和跨会话陈旧回调；真实 TCP 回环验证分片收包与实际 -T8 到期关闭。 - -以上 T5/T8 是当前工程基线。具体默认值、连接失败分类、T5 起止点、T8 -对每个网络分片还是协议字符的精确解释，以及标准结论，仍须依据团队合法 -获得的 SEMI E37/E37.1 版本及一致性测试核对。 - -## 替换条件 - -当 #38/#39 都发布稳定 NuGet API 后： - -1. 运行现有会话、发送确认、迟到消息和超时测试向量验证语义等价； -2. 用原生 Session ID、发送完成任务和未完成帧超时替换内部回调适配； -3. 删除旧信封 fail-closed 路径； -4. 保持 IHsmsTransport 及其上层状态机契约不变。 +不内置未经授权标准核对的默认值。Active 连接把 T5 映射为固定的 +ConnectRetryDelayMs 并关闭最大退避；Passive 监听重试配置 +保持调用方设置。原 StreamConnectionOptions 始终复制后再 +适配，不会被原地修改。 + +T8 映射为 StreamFrame 的 IncompleteFrameTimeoutMs。因此 +T5/T8 都必须为正值、能由整毫秒精确表示且不超过 int.MaxValue +毫秒；不允许截断、舍入或溢出。调用方在源 StreamFrame 选项中设置的 +IncompleteFrameTimeoutMs 会被显式 HSMS T8 覆盖。 + +StreamFrame 只在已有未完成帧时运行该超时：空闲连接和完整帧后的静默不 +触发；接收进展重置计时；到期先发布 +FrameErrorKind.IncompleteFrameTimeout，再拆除对应会话。 +适配器把该诊断转换为携带 transport Session ID 的 +HsmsT8TimeoutException，不再次请求重连。 + +## 验证边界 + +适配器单元测试覆盖原生 Session ID、迟到消息、发送完成、并发入队、会话 +失效、显式关闭原因和 T8 诊断映射。真实 TCP 回环测试覆盖分片帧收发与 +实际 T8 到期；完整套件继续覆盖 T3/T6 从发送完成点启动以及旧会话不 +重放。 + +这些测试证明当前工程契约与 StreamFrame 2.3.0 的互操作行为，不构成 +SEMI 合规声明。T5/T8 的标准默认值、精确启停边界和异常恢复仍须依据团队 +合法获得的 SEMI E37/E37.1 版本及一致性测试核对。仓库不得提交标准正文、 +表格或 Schema。 diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs index 497851b..d25d608 100644 --- a/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs +++ b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs @@ -17,7 +17,7 @@ public sealed class HsmsConnectionOptions /// The Active connection retry interval. /// The control transaction reply timeout. /// The selection timeout after TCP connection. - /// The incomplete-message receive timeout. + /// The positive whole-millisecond incomplete-message receive timeout. public HsmsConnectionOptions( IPAddress ipAddress, int port, @@ -59,7 +59,7 @@ public HsmsConnectionOptions( ValidateT5(t5, nameof(t5)); ValidatePositive(t6, nameof(t6), "T6"); ValidatePositive(t7, nameof(t7), "T7"); - ValidatePositive(t8, nameof(t8), "T8"); + ValidateStreamFrameMilliseconds(t8, nameof(t8), "T8"); Port = port; ConnectionMode = connectionMode; @@ -116,13 +116,21 @@ private static void ValidateT5( TimeSpan value, string parameterName) { - ValidatePositive(value, parameterName, "T5"); + ValidateStreamFrameMilliseconds(value, parameterName, "T5"); + } + + private static void ValidateStreamFrameMilliseconds( + TimeSpan value, + string parameterName, + string timerName) + { + ValidatePositive(value, parameterName, timerName); if (value.Ticks % TimeSpan.TicksPerMillisecond != 0) { throw new ArgumentOutOfRangeException( parameterName, value, - "T5 must be representable as a whole number of milliseconds."); + $"{timerName} must be representable as a whole number of milliseconds."); } var milliseconds = value.Ticks / TimeSpan.TicksPerMillisecond; @@ -131,7 +139,7 @@ private static void ValidateT5( throw new ArgumentOutOfRangeException( parameterName, value, - "T5 exceeds the supported connection retry-delay range."); + $"{timerName} exceeds the supported StreamFrame millisecond range."); } } } diff --git a/src/SecsFrame/Transport/HsmsStreamConnectionOptionsAdapter.cs b/src/SecsFrame/Transport/HsmsStreamConnectionOptionsAdapter.cs index 15cb630..30012fc 100644 --- a/src/SecsFrame/Transport/HsmsStreamConnectionOptionsAdapter.cs +++ b/src/SecsFrame/Transport/HsmsStreamConnectionOptionsAdapter.cs @@ -31,6 +31,7 @@ public static StreamConnectionOptions Create( KeepAliveIntervalMs = source.KeepAliveIntervalMs, ReceiveQueueCapacity = source.ReceiveQueueCapacity, ReceiveIdleTimeoutMs = source.ReceiveIdleTimeoutMs, + IncompleteFrameTimeoutMs = hsmsOptions.T8Milliseconds, }; if (isActive) diff --git a/src/SecsFrame/Transport/HsmsT8Monitor.cs b/src/SecsFrame/Transport/HsmsT8Monitor.cs deleted file mode 100644 index 8b77dfd..0000000 --- a/src/SecsFrame/Transport/HsmsT8Monitor.cs +++ /dev/null @@ -1,163 +0,0 @@ -namespace SecsFrame; - -internal sealed class HsmsT8Monitor : IDisposable -{ -#if NET9_0_OR_GREATER - private readonly Lock _gate = new(); -#else - private readonly object _gate = new(); -#endif - private readonly TimeSpan _timeout; - private readonly Action _onTimeout; - private readonly IHsmsTransportTimerFactory _timerFactory; - private IHsmsTransportTimer? _timer; - private uint _declaredPayloadLength; - private long _remainingPayloadBytes; - private long _timerGeneration; - private int _prefixBytesRead; - private bool _disposed; - - public HsmsT8Monitor( - TimeSpan timeout, - Action onTimeout, - IHsmsTransportTimerFactory? timerFactory = null) - { - if (timeout <= TimeSpan.Zero) - { - throw new ArgumentOutOfRangeException( - nameof(timeout), - timeout, - "T8 must be positive."); - } - - _timeout = timeout; - _onTimeout = - onTimeout ?? throw new ArgumentNullException(nameof(onTimeout)); - _timerFactory = - timerFactory ?? SystemHsmsTransportTimerFactory.Instance; - } - - public void Observe(ReadOnlySpan bytes) - { - if (bytes.IsEmpty) - return; - - lock (_gate) - { - if (_disposed) - return; - - ObserveFrames(bytes); - if (IsIncomplete) - RestartTimer(); - else - StopTimer(); - } - } - - public void Reset() - { - lock (_gate) - { - if (_disposed) - return; - - ResetFrameState(); - StopTimer(); - } - } - - public void Dispose() - { - lock (_gate) - { - if (_disposed) - return; - - _disposed = true; - ResetFrameState(); - StopTimer(); - } - } - - private bool IsIncomplete => _prefixBytesRead > 0; - - private void ObserveFrames(ReadOnlySpan bytes) - { - var offset = 0; - while (offset < bytes.Length) - { - while (_prefixBytesRead < HsmsFramer.LengthPrefixSize && - offset < bytes.Length) - { - _declaredPayloadLength = - (_declaredPayloadLength << 8) | bytes[offset++]; - _prefixBytesRead++; - } - - if (_prefixBytesRead < HsmsFramer.LengthPrefixSize) - break; - - if (_remainingPayloadBytes == 0) - _remainingPayloadBytes = _declaredPayloadLength; - - if (_remainingPayloadBytes == 0) - { - ResetFrameState(); - continue; - } - - var available = bytes.Length - offset; - var consumed = (int)Math.Min( - _remainingPayloadBytes, - available); - _remainingPayloadBytes -= consumed; - offset += consumed; - - if (_remainingPayloadBytes == 0) - ResetFrameState(); - } - } - - private void RestartTimer() - { - StopTimer(); - var generation = _timerGeneration; - _timer = _timerFactory.Create( - () => OnTimeout(generation)); - _timer.Change(_timeout); - } - - private void StopTimer() - { - _timerGeneration++; - _timer?.Dispose(); - _timer = null; - } - - private void OnTimeout(long generation) - { - var notify = false; - lock (_gate) - { - if (!_disposed && - generation == _timerGeneration && - IsIncomplete) - { - ResetFrameState(); - StopTimer(); - notify = true; - } - } - - if (notify) - _onTimeout(); - } - - private void ResetFrameState() - { - _declaredPayloadLength = 0; - _remainingPayloadBytes = 0; - _prefixBytesRead = 0; - } -} diff --git a/src/SecsFrame/Transport/HsmsTransportFrame.cs b/src/SecsFrame/Transport/HsmsTransportFrame.cs deleted file mode 100644 index f2fa340..0000000 --- a/src/SecsFrame/Transport/HsmsTransportFrame.cs +++ /dev/null @@ -1,8 +0,0 @@ -using System.Runtime.InteropServices; - -namespace SecsFrame; - -[StructLayout(LayoutKind.Auto)] -internal readonly record struct HsmsTransportFrame( - HsmsTransportSessionId SessionId, - HsmsFrame Frame); diff --git a/src/SecsFrame/Transport/HsmsTransportOptions.cs b/src/SecsFrame/Transport/HsmsTransportOptions.cs index d9218a7..46c2ec9 100644 --- a/src/SecsFrame/Transport/HsmsTransportOptions.cs +++ b/src/SecsFrame/Transport/HsmsTransportOptions.cs @@ -37,9 +37,27 @@ public HsmsTransportOptions(TimeSpan t5, TimeSpan t8) "T8 must be positive."); } + if (t8.Ticks % TimeSpan.TicksPerMillisecond != 0) + { + throw new ArgumentOutOfRangeException( + nameof(t8), + t8, + "T8 must be representable as a whole number of milliseconds."); + } + + var t8Milliseconds = t8.Ticks / TimeSpan.TicksPerMillisecond; + if (t8Milliseconds > int.MaxValue) + { + throw new ArgumentOutOfRangeException( + nameof(t8), + t8, + "T8 exceeds the StreamFrame incomplete-frame timeout range."); + } + T5 = t5; T8 = t8; T5Milliseconds = (int)t5Milliseconds; + T8Milliseconds = (int)t8Milliseconds; } public TimeSpan T5 { get; } @@ -47,4 +65,6 @@ public HsmsTransportOptions(TimeSpan t5, TimeSpan t8) public TimeSpan T8 { get; } internal int T5Milliseconds { get; } + + internal int T8Milliseconds { get; } } diff --git a/src/SecsFrame/Transport/HsmsTransportSessionContext.cs b/src/SecsFrame/Transport/HsmsTransportSessionContext.cs deleted file mode 100644 index be120d4..0000000 --- a/src/SecsFrame/Transport/HsmsTransportSessionContext.cs +++ /dev/null @@ -1,65 +0,0 @@ -namespace SecsFrame; - -internal sealed class HsmsTransportSessionContext -{ -#if NET9_0_OR_GREATER - private readonly Lock _gate = new(); -#else - private readonly object _gate = new(); -#endif - private long _lastSessionId; - private HsmsTransportSessionId _current; - - public HsmsTransportSessionId Open() - { - lock (_gate) - { - if (_current.IsValid) - throw new InvalidOperationException($"Transport session {_current.Value} is already active."); - if (_lastSessionId == long.MaxValue) - throw new InvalidOperationException("The transport session identifier space is exhausted."); - - _current = new HsmsTransportSessionId(++_lastSessionId); - return _current; - } - } - - public bool TryClose(out HsmsTransportSessionId sessionId) - { - lock (_gate) - { - sessionId = _current; - if (!sessionId.IsValid) - return false; - - _current = default; - return true; - } - } - - public bool IsCurrent(HsmsTransportSessionId sessionId) - { - lock (_gate) - return sessionId.IsValid && sessionId == _current; - } - - public bool TryGetCurrent(out HsmsTransportSessionId sessionId) - { - lock (_gate) - { - sessionId = _current; - return sessionId.IsValid; - } - } - - public HsmsTransportSessionId GetCurrent() - { - lock (_gate) - { - if (!_current.IsValid) - throw new HsmsTransportSessionExpiredException(default, "No transport session is active."); - - return _current; - } - } -} diff --git a/src/SecsFrame/Transport/HsmsTransportSessionExpiredException.cs b/src/SecsFrame/Transport/HsmsTransportSessionExpiredException.cs index c6976a4..ecdce77 100644 --- a/src/SecsFrame/Transport/HsmsTransportSessionExpiredException.cs +++ b/src/SecsFrame/Transport/HsmsTransportSessionExpiredException.cs @@ -6,8 +6,11 @@ internal sealed class HsmsTransportSessionExpiredException : IOException { public HsmsTransportSessionExpiredException( HsmsTransportSessionId sessionId, - string? message = null) - : base(message ?? $"Transport session {sessionId.Value} is no longer active.") + string? message = null, + Exception? innerException = null) + : base( + message ?? $"Transport session {sessionId.Value} is no longer active.", + innerException) { SessionId = sessionId; } diff --git a/src/SecsFrame/Transport/SessionBoundHsmsFrameCodec.cs b/src/SecsFrame/Transport/SessionBoundHsmsFrameCodec.cs deleted file mode 100644 index 9250da3..0000000 --- a/src/SecsFrame/Transport/SessionBoundHsmsFrameCodec.cs +++ /dev/null @@ -1,29 +0,0 @@ -using System.Buffers; -using StreamFrame; - -namespace SecsFrame; - -internal sealed class SessionBoundHsmsFrameCodec : ICodec -{ - private readonly HsmsTransportSessionContext _sessionContext; - private readonly HsmsFrameCodec _frameCodec = new(); - - public SessionBoundHsmsFrameCodec(HsmsTransportSessionContext sessionContext) - { - _sessionContext = sessionContext ?? throw new ArgumentNullException(nameof(sessionContext)); - } - - public HsmsTransportFrame Decode(in ReadOnlySequence frame, CancellationToken ct = default) - => new(_sessionContext.GetCurrent(), _frameCodec.Decode(frame, ct)); - - public void Encode( - HsmsTransportFrame message, - IBufferWriter writer, - CancellationToken ct = default) - { - if (!_sessionContext.IsCurrent(message.SessionId)) - throw new HsmsTransportSessionExpiredException(message.SessionId); - - _frameCodec.Encode(message.Frame, writer, ct); - } -} diff --git a/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs b/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs index 2633c91..e419ac7 100644 --- a/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs +++ b/src/SecsFrame/Transport/StreamFrameHsmsTransport.cs @@ -1,5 +1,6 @@ using System.Net; using System.Runtime.CompilerServices; +using System.Runtime.ExceptionServices; using System.Threading.Channels; using StreamFrame; @@ -7,47 +8,32 @@ namespace SecsFrame; internal sealed class StreamFrameHsmsTransport : IHsmsTransport { - private readonly IStreamConnection _connection; - private readonly HsmsTransportSessionContext _sessionContext; - private readonly HsmsT8Monitor _t8Monitor; + private readonly ISessionAwareStreamConnection _connection; private readonly Channel _events; - private readonly SemaphoreSlim _sendGate = new(1, 1); #if NET9_0_OR_GREATER - private readonly Lock _pendingGate = new(); - private readonly Lock _closeReasonGate = new(); + private readonly Lock _sessionGate = new(); #else - private readonly object _pendingGate = new(); - private readonly object _closeReasonGate = new(); + private readonly object _sessionGate = new(); #endif - private PendingSend? _pendingSend; - private Exception? _nextCloseReason; + private readonly Dictionary _sessions = new(); private Task? _messagePump; + private long _currentSessionId; private int _started; private int _disposed; internal StreamFrameHsmsTransport( - IStreamConnection connection, - HsmsTransportSessionContext sessionContext, - HsmsTransportOptions options, - IHsmsTransportTimerFactory? timerFactory) + ISessionAwareStreamConnection connection) { _connection = connection ?? throw new ArgumentNullException(nameof(connection)); - _sessionContext = sessionContext ?? throw new ArgumentNullException(nameof(sessionContext)); - options = options ?? throw new ArgumentNullException(nameof(options)); _events = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false, AllowSynchronousContinuations = false, }); - _t8Monitor = new HsmsT8Monitor( - options.T8, - OnT8Timeout, - timerFactory); _connection.ConnectionChanged += OnConnectionChanged; - _connection.RawBytesReceived = OnRawBytesReceived; - _connection.RawBytesSent = OnRawBytesSent; + _connection.FrameError += OnFrameError; } public static StreamFrameHsmsTransport Create( @@ -61,23 +47,18 @@ public static StreamFrameHsmsTransport Create( if (hsmsOptions is null) throw new ArgumentNullException(nameof(hsmsOptions)); - var sessionContext = new HsmsTransportSessionContext(); var adaptedOptions = HsmsStreamConnectionOptionsAdapter.Create( isActive, hsmsOptions, connectionOptions); - var connection = new StreamConnection( + var connection = new StreamConnection( framer ?? new HsmsFramer(), - new SessionBoundHsmsFrameCodec(sessionContext), + new HsmsFrameCodec(), ipAddress, port, isActive, adaptedOptions); - return new StreamFrameHsmsTransport( - connection, - sessionContext, - hsmsOptions, - timerFactory: null); + return new StreamFrameHsmsTransport(connection); } public void Start(CancellationToken cancellationToken) @@ -117,40 +98,30 @@ public async ValueTask SendAsync( if (frame is null) throw new ArgumentNullException(nameof(frame)); ThrowIfDisposed(); - EnsureCurrentSession(sessionId); + RegisterPendingSend(sessionId); - await _sendGate.WaitAsync(cancellationToken).ConfigureAwait(false); - PendingSend? pending = null; try { - ThrowIfDisposed(); - EnsureCurrentSession(sessionId); - - pending = new PendingSend(GetWireLength(frame)); - lock (_pendingGate) - { - EnsureCurrentSession(sessionId); - _pendingSend = pending; - } - - await _connection.SendAsync( - new HsmsTransportFrame(sessionId, frame), + await _connection.SendInSessionAsync( + sessionId.Value, + frame, cancellationToken).ConfigureAwait(false); - - await pending.Completion.Task.ConfigureAwait(false); } - finally + catch (SessionExpiredException ex) { - if (pending is not null) + var closeReason = GetCloseReason(sessionId); + if (closeReason is not null) { - lock (_pendingGate) - { - if (ReferenceEquals(_pendingSend, pending)) - _pendingSend = null; - } + ExceptionDispatchInfo.Capture(closeReason).Throw(); } - _sendGate.Release(); + throw new HsmsTransportSessionExpiredException( + sessionId, + innerException: ex); + } + finally + { + CompletePendingSend(sessionId); } } @@ -159,7 +130,6 @@ public async ValueTask DisposeAsync() if (Interlocked.Exchange(ref _disposed, 1) != 0) return; - FailPending(new ObjectDisposedException(nameof(StreamFrameHsmsTransport))); await _connection.DisposeAsync().ConfigureAwait(false); if (_messagePump is not null) @@ -174,9 +144,7 @@ public async ValueTask DisposeAsync() } _connection.ConnectionChanged -= OnConnectionChanged; - _connection.RawBytesReceived = null; - _connection.RawBytesSent = null; - _t8Monitor.Dispose(); + _connection.FrameError -= OnFrameError; _events.Writer.TryComplete(); } @@ -185,11 +153,22 @@ public bool TryCloseSession( Exception? error = null) { ThrowIfDisposed(); - if (!_sessionContext.IsCurrent(sessionId)) - return false; - lock (_closeReasonGate) - _nextCloseReason = error; + Exception? previousReason; + lock (_sessionGate) + { + if (_currentSessionId != sessionId.Value || + _connection.CurrentSessionId != sessionId.Value || + !_sessions.TryGetValue(sessionId.Value, out var context) || + context.IsClosed) + { + return false; + } + + previousReason = context.CloseReason; + if (error is not null && context.CloseReason is null) + context.CloseReason = error; + } try { @@ -198,22 +177,26 @@ public bool TryCloseSession( } catch { - lock (_closeReasonGate) - _nextCloseReason = null; + if (error is not null) + RestoreCloseReason(sessionId, error, previousReason); throw; } } private async Task PumpMessagesAsync() { - var messages = _connection.GetMessages(CancellationToken.None).GetAsyncEnumerator(); + var messages = _connection + .GetSessionMessages(CancellationToken.None) + .GetAsyncEnumerator(); try { while (await messages.MoveNextAsync().ConfigureAwait(false)) { var message = messages.Current; if (!_events.Writer.TryWrite( - HsmsTransportEvent.FrameReceived(message.SessionId, message.Frame))) + HsmsTransportEvent.FrameReceived( + new HsmsTransportSessionId(message.SessionId), + message.Message))) { break; } @@ -235,93 +218,127 @@ private void OnConnectionChanged(object? sender, ConnectionState state) { if (state == ConnectionState.Connected) { - _t8Monitor.Reset(); - var sessionId = _sessionContext.Open(); - _events.Writer.TryWrite(HsmsTransportEvent.SessionOpened(sessionId)); + var openedSessionId = _connection.CurrentSessionId; + if (openedSessionId <= 0) + { + _events.Writer.TryComplete( + new InvalidOperationException( + "StreamFrame published Connected without a valid session identifier.")); + return; + } + + lock (_sessionGate) + { + _sessions.Add(openedSessionId, new SessionContext()); + Volatile.Write(ref _currentSessionId, openedSessionId); + } + + _events.Writer.TryWrite( + HsmsTransportEvent.SessionOpened( + new HsmsTransportSessionId(openedSessionId))); return; } - if (!_sessionContext.TryClose(out var closedSessionId)) + var closedSessionValue = Interlocked.Exchange(ref _currentSessionId, 0); + if (closedSessionValue <= 0) return; - _t8Monitor.Reset(); - var exception = TakeCloseReason(); - FailPending( - exception ?? new HsmsTransportSessionExpiredException(closedSessionId)); - _events.Writer.TryWrite(HsmsTransportEvent.SessionClosed(closedSessionId, exception)); - } - - private void OnRawBytesReceived(ReadOnlyMemory bytes) - => _t8Monitor.Observe(bytes.Span); - - private void OnRawBytesSent(ReadOnlyMemory bytes) - { - PendingSend? completed = null; - lock (_pendingGate) + Exception? closeReason = null; + lock (_sessionGate) { - var pending = _pendingSend; - if (pending is null || pending.Completion.Task.IsCompleted) - return; - - pending.RemainingBytes -= bytes.Length; - if (pending.RemainingBytes == 0) - completed = pending; - else if (pending.RemainingBytes < 0) + if (_sessions.TryGetValue(closedSessionValue, out var context)) { - pending.Completion.TrySetException( - new InvalidOperationException("StreamFrame reported more sent bytes than the pending HSMS frame contains.")); + context.IsClosed = true; + closeReason = context.CloseReason; + RemoveSessionIfComplete(closedSessionValue, context); } } - completed?.Completion.TrySetResult(true); + _events.Writer.TryWrite( + HsmsTransportEvent.SessionClosed( + new HsmsTransportSessionId(closedSessionValue), + closeReason)); } - private void OnT8Timeout() + private void OnFrameError(object? sender, FrameErrorEventArgs args) { - if (Volatile.Read(ref _disposed) != 0) + if (args.Kind != FrameErrorKind.IncompleteFrameTimeout) return; - try + var sessionValue = Volatile.Read(ref _currentSessionId); + if (sessionValue <= 0) + return; + + lock (_sessionGate) { - if (_sessionContext.TryGetCurrent(out var sessionId)) + if (_currentSessionId == sessionValue && + _sessions.TryGetValue(sessionValue, out var context) && + !context.IsClosed && + context.CloseReason is null) { - TryCloseSession( - sessionId, - new HsmsT8TimeoutException(sessionId)); + context.CloseReason = new HsmsT8TimeoutException( + new HsmsTransportSessionId(sessionValue)); } } - catch (ObjectDisposedException) + } + + private void RegisterPendingSend(HsmsTransportSessionId sessionId) + { + lock (_sessionGate) { + if (_currentSessionId != sessionId.Value || + _connection.CurrentSessionId != sessionId.Value || + !_sessions.TryGetValue(sessionId.Value, out var context) || + context.IsClosed) + { + throw new HsmsTransportSessionExpiredException(sessionId); + } + + context.PendingSends++; } - catch (Exception ex) + } + + private void CompletePendingSend(HsmsTransportSessionId sessionId) + { + lock (_sessionGate) { - FailPending(ex); + if (!_sessions.TryGetValue(sessionId.Value, out var context)) + return; + + context.PendingSends--; + RemoveSessionIfComplete(sessionId.Value, context); } } - private void FailPending(Exception exception) + private Exception? GetCloseReason(HsmsTransportSessionId sessionId) { - PendingSend? pending; - lock (_pendingGate) - pending = _pendingSend; - - pending?.Completion.TrySetException(exception); + lock (_sessionGate) + { + return _sessions.TryGetValue(sessionId.Value, out var context) + ? context.CloseReason + : null; + } } - private Exception? TakeCloseReason() + private void RestoreCloseReason( + HsmsTransportSessionId sessionId, + Exception attemptedReason, + Exception? previousReason) { - lock (_closeReasonGate) + lock (_sessionGate) { - var exception = _nextCloseReason; - _nextCloseReason = null; - return exception; + if (_sessions.TryGetValue(sessionId.Value, out var context) && + ReferenceEquals(context.CloseReason, attemptedReason)) + { + context.CloseReason = previousReason; + } } } - private void EnsureCurrentSession(HsmsTransportSessionId sessionId) + private void RemoveSessionIfComplete(long sessionId, SessionContext context) { - if (!_sessionContext.IsCurrent(sessionId)) - throw new HsmsTransportSessionExpiredException(sessionId); + if (context.IsClosed && context.PendingSends == 0) + _sessions.Remove(sessionId); } private void ThrowIfDisposed() @@ -330,23 +347,12 @@ private void ThrowIfDisposed() throw new ObjectDisposedException(nameof(StreamFrameHsmsTransport)); } - private static long GetWireLength(HsmsFrame frame) - => checked( - (long)HsmsFramer.LengthPrefixSize + - HsmsMessageHeader.EncodedSize + - frame.Body.Length); - - private sealed class PendingSend + private sealed class SessionContext { - public PendingSend(long remainingBytes) - { - RemainingBytes = remainingBytes; - Completion = new TaskCompletionSource( - TaskCreationOptions.RunContinuationsAsynchronously); - } + public int PendingSends { get; set; } - public long RemainingBytes { get; set; } + public bool IsClosed { get; set; } - public TaskCompletionSource Completion { get; } + public Exception? CloseReason { get; set; } } } diff --git a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs index 5504926..839f0a3 100644 --- a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs +++ b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs @@ -101,6 +101,24 @@ public void T5_beyond_connection_retry_range_is_rejected() (long)int.MaxValue + 1))); } + [Fact] + public void Submillisecond_T8_is_rejected_instead_of_rounded() + { + Assert.Throws( + () => Create( + t8: TimeSpan.FromTicks( + TimeSpan.TicksPerMillisecond + 1))); + } + + [Fact] + public void T8_beyond_StreamFrame_range_is_rejected() + { + Assert.Throws( + () => Create( + t8: TimeSpan.FromMilliseconds( + (long)int.MaxValue + 1))); + } + private static HsmsConnectionOptions Create( IPAddress? ipAddress = null, int port = 5000, diff --git a/test/SecsFrame.Tests/HsmsStreamConnectionOptionsAdapterTests.cs b/test/SecsFrame.Tests/HsmsStreamConnectionOptionsAdapterTests.cs index 5f21928..38ee6ec 100644 --- a/test/SecsFrame.Tests/HsmsStreamConnectionOptionsAdapterTests.cs +++ b/test/SecsFrame.Tests/HsmsStreamConnectionOptionsAdapterTests.cs @@ -20,10 +20,12 @@ public void Active_options_apply_T5_disable_backoff_and_preserve_source() Assert.NotSame(source, adapted); Assert.Equal(1250, adapted.ConnectRetryDelayMs); Assert.Equal(0, adapted.MaxRetryDelayMs); + Assert.Equal(5000, adapted.IncompleteFrameTimeoutMs); Assert.Equal(source.AcceptRetryDelayMs, adapted.AcceptRetryDelayMs); AssertOtherOptionsEqual(source, adapted); Assert.Equal(111, source.ConnectRetryDelayMs); Assert.Equal(333, source.MaxRetryDelayMs); + Assert.Equal(888, source.IncompleteFrameTimeoutMs); } [Fact] @@ -43,6 +45,7 @@ public void Passive_options_preserve_StreamFrame_retry_configuration() Assert.Equal(source.ConnectRetryDelayMs, adapted.ConnectRetryDelayMs); Assert.Equal(source.AcceptRetryDelayMs, adapted.AcceptRetryDelayMs); Assert.Equal(source.MaxRetryDelayMs, adapted.MaxRetryDelayMs); + Assert.Equal(5000, adapted.IncompleteFrameTimeoutMs); AssertOtherOptionsEqual(source, adapted); } @@ -51,11 +54,12 @@ public void T5_at_StreamFrame_maximum_is_lossless() { var options = new HsmsTransportOptions( TimeSpan.FromMilliseconds(int.MaxValue), - TimeSpan.FromTicks(1)); + TimeSpan.FromMilliseconds(1)); Assert.Equal(TimeSpan.FromMilliseconds(int.MaxValue), options.T5); Assert.Equal(int.MaxValue, options.T5Milliseconds); - Assert.Equal(TimeSpan.FromTicks(1), options.T8); + Assert.Equal(TimeSpan.FromMilliseconds(1), options.T8); + Assert.Equal(1, options.T8Milliseconds); } [Fact] @@ -98,6 +102,24 @@ public void Nonpositive_T8_is_rejected() TimeSpan.FromTicks(-1))); } + [Fact] + public void Submillisecond_T8_is_rejected_instead_of_rounded() + { + Assert.Throws( + () => new HsmsTransportOptions( + TimeSpan.FromSeconds(1), + TimeSpan.FromTicks(TimeSpan.TicksPerMillisecond + 1))); + } + + [Fact] + public void T8_beyond_StreamFrame_range_is_rejected() + { + Assert.Throws( + () => new HsmsTransportOptions( + TimeSpan.FromSeconds(1), + TimeSpan.FromMilliseconds((long)int.MaxValue + 1))); + } + private static StreamConnectionOptions CreateSource() => new() { @@ -115,6 +137,7 @@ private static StreamConnectionOptions CreateSource() KeepAliveIntervalMs = 555, ReceiveQueueCapacity = 6, ReceiveIdleTimeoutMs = 777, + IncompleteFrameTimeoutMs = 888, }; private static void AssertOtherOptionsEqual( diff --git a/test/SecsFrame.Tests/HsmsT8MonitorTests.cs b/test/SecsFrame.Tests/HsmsT8MonitorTests.cs deleted file mode 100644 index c851ccd..0000000 --- a/test/SecsFrame.Tests/HsmsT8MonitorTests.cs +++ /dev/null @@ -1,157 +0,0 @@ -namespace SecsFrame.Tests; - -public sealed class HsmsT8MonitorTests -{ - private static readonly TimeSpan T8 = TimeSpan.FromSeconds(5); - - [Fact] - public void Idle_connection_does_not_create_a_timer() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - var timeoutCount = 0; - using var monitor = new HsmsT8Monitor( - T8, - () => timeoutCount++, - timerFactory); - - Assert.Empty(timerFactory.Timers); - Assert.Equal(0, timeoutCount); - } - - [Fact] - public void Partial_length_prefix_arms_T8() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - var timeoutCount = 0; - using var monitor = new HsmsT8Monitor( - T8, - () => timeoutCount++, - timerFactory); - - monitor.Observe(new byte[] { 0x00, 0x00 }); - var timer = Assert.Single(timerFactory.Timers); - Assert.Equal(T8, timer.DueTime); - timer.Fire(); - - Assert.Equal(1, timeoutCount); - Assert.True(timer.IsDisposed); - } - - [Fact] - public void Partial_payload_progress_replaces_T8_and_ignores_queued_callback() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - var timeoutCount = 0; - using var monitor = new HsmsT8Monitor( - T8, - () => timeoutCount++, - timerFactory); - - monitor.Observe(new byte[] { 0x00, 0x00, 0x00, 0x0A, 0x01 }); - var first = Assert.Single(timerFactory.Timers); - monitor.Observe(new byte[] { 0x02, 0x03 }); - var second = timerFactory.Timer!; - - Assert.NotSame(first, second); - Assert.True(first.IsDisposed); - Assert.True(second.IsArmed); - Assert.Equal(T8, second.DueTime); - first.ForceFire(); - Assert.Equal(0, timeoutCount); - second.Fire(); - Assert.Equal(1, timeoutCount); - } - - [Fact] - public void Complete_frame_stops_T8_during_later_idle_period() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - var timeoutCount = 0; - using var monitor = new HsmsT8Monitor( - T8, - () => timeoutCount++, - timerFactory); - var wireFrame = CreateHeaderOnlyWireFrame(); - - monitor.Observe(wireFrame.AsSpan(0, 5)); - var timer = Assert.Single(timerFactory.Timers); - monitor.Observe(wireFrame.AsSpan(5)); - timer.ForceFire(); - - Assert.True(timer.IsDisposed); - Assert.Equal(0, timeoutCount); - } - - [Fact] - public void Complete_frame_followed_by_partial_next_frame_remains_armed() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - var timeoutCount = 0; - using var monitor = new HsmsT8Monitor( - T8, - () => timeoutCount++, - timerFactory); - var first = CreateHeaderOnlyWireFrame(); - var bytes = new byte[first.Length + 1]; - first.CopyTo(bytes, 0); - bytes[^1] = 0; - - monitor.Observe(bytes); - timerFactory.Timer!.Fire(); - - Assert.Equal(1, timeoutCount); - } - - [Fact] - public void Reset_isolates_replacement_frame_from_queued_callback() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - var timeoutCount = 0; - using var monitor = new HsmsT8Monitor( - T8, - () => timeoutCount++, - timerFactory); - - monitor.Observe(new byte[] { 0x00 }); - var previousSessionTimer = timerFactory.Timer!; - monitor.Reset(); - monitor.Observe(new byte[] { 0x00, 0x00 }); - var currentSessionTimer = timerFactory.Timer!; - previousSessionTimer.ForceFire(); - - Assert.NotSame(previousSessionTimer, currentSessionTimer); - Assert.Equal(0, timeoutCount); - Assert.True(currentSessionTimer.IsArmed); - currentSessionTimer.Fire(); - Assert.Equal(1, timeoutCount); - } - - [Fact] - public void Multiple_complete_frames_do_not_create_a_timer() - { - var timerFactory = new ManualHsmsTransportTimerFactory(); - using var monitor = new HsmsT8Monitor(T8, () => { }, timerFactory); - var frame = CreateHeaderOnlyWireFrame(); - var bytes = new byte[frame.Length * 2]; - frame.CopyTo(bytes, 0); - frame.CopyTo(bytes, frame.Length); - - monitor.Observe(bytes); - - Assert.Empty(timerFactory.Timers); - } - - [Fact] - public void Nonpositive_T8_is_rejected() - { - Assert.Throws( - () => new HsmsT8Monitor(TimeSpan.Zero, () => { })); - } - - private static byte[] CreateHeaderOnlyWireFrame() - => new byte[] - { - 0x00, 0x00, 0x00, 0x0A, - 0x00, 0x01, 0x01, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, - }; -} diff --git a/test/SecsFrame.Tests/SessionBoundHsmsFrameCodecTests.cs b/test/SecsFrame.Tests/SessionBoundHsmsFrameCodecTests.cs deleted file mode 100644 index 372cb91..0000000 --- a/test/SecsFrame.Tests/SessionBoundHsmsFrameCodecTests.cs +++ /dev/null @@ -1,49 +0,0 @@ -using System.Buffers; - -namespace SecsFrame.Tests; - -public sealed class SessionBoundHsmsFrameCodecTests -{ - [Fact] - public void Decode_stamps_the_current_transport_session() - { - var context = new HsmsTransportSessionContext(); - var sessionId = context.Open(); - var codec = new SessionBoundHsmsFrameCodec(context); - var bytes = new byte[HsmsMessageHeader.EncodedSize]; - HsmsMessageHeader.CreateData(1, 1, 1, false, 1).Encode(bytes); - - var decoded = codec.Decode(new ReadOnlySequence(bytes)); - - Assert.Equal(sessionId, decoded.SessionId); - Assert.Equal(1, decoded.Frame.Header.Stream); - } - - [Fact] - public void Encode_rejects_an_envelope_from_a_closed_session() - { - var context = new HsmsTransportSessionContext(); - var oldSessionId = context.Open(); - Assert.True(context.TryClose(out _)); - context.Open(); - var codec = new SessionBoundHsmsFrameCodec(context); - var envelope = new HsmsTransportFrame( - oldSessionId, - new HsmsFrame(HsmsMessageHeader.CreateData(1, 1, 1, false, 1))); - - var exception = Assert.Throws( - () => codec.Encode(envelope, new TestBufferWriter())); - - Assert.Equal(oldSessionId, exception.SessionId); - } - - [Fact] - public void Decode_without_an_active_session_is_rejected() - { - var codec = new SessionBoundHsmsFrameCodec(new HsmsTransportSessionContext()); - var bytes = new byte[HsmsMessageHeader.EncodedSize]; - - Assert.Throws( - () => codec.Decode(new ReadOnlySequence(bytes))); - } -} diff --git a/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs b/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs index a34b3ef..f980bc9 100644 --- a/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs +++ b/test/SecsFrame.Tests/StreamFrameHsmsTransportTests.cs @@ -11,11 +11,10 @@ public sealed class StreamFrameHsmsTransportTests private static readonly TimeSpan T8 = TimeSpan.FromSeconds(5); [Fact] - public async Task Sessions_open_and_close_with_monotonic_identifiers() + public async Task Sessions_use_StreamFrame_monotonic_identifiers() { var connection = new FakeStreamConnection(); - var timerFactory = new ManualHsmsTransportTimerFactory(); - await using var transport = CreateTransport(connection, timerFactory); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -33,15 +32,15 @@ public async Task Sessions_open_and_close_with_monotonic_identifiers() Assert.Equal(opened1.SessionId, closed1.SessionId); Assert.Equal(HsmsTransportEventKind.SessionOpened, opened2.Kind); Assert.True(opened2.SessionId.Value > opened1.SessionId.Value); + Assert.Equal(connection.CurrentSessionId, opened2.SessionId.Value); await events.DisposeAsync().ConfigureAwait(true); } [Fact] - public async Task Late_received_frame_keeps_its_original_session_identifier() + public async Task Late_received_frame_keeps_its_native_session_identifier() { var connection = new FakeStreamConnection(); - var timerFactory = new ManualHsmsTransportTimerFactory(); - await using var transport = CreateTransport(connection, timerFactory); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -49,7 +48,7 @@ public async Task Late_received_frame_keeps_its_original_session_identifier() var originalSession = (await NextAsync(events)).SessionId; var frame = CreateFrame(); - connection.Emit(new HsmsTransportFrame(originalSession, frame)); + connection.Emit(originalSession.Value, frame); connection.RaiseState(ConnectionState.Retry); connection.RaiseState(ConnectionState.Connected); @@ -68,10 +67,10 @@ public async Task Late_received_frame_keeps_its_original_session_identifier() } [Fact] - public async Task Send_completes_only_after_the_entire_wire_frame_is_reported() + public async Task Send_completes_only_after_native_socket_write_confirmation() { var connection = new FakeStreamConnection(); - await using var transport = CreateTransport(connection, new ManualHsmsTransportTimerFactory()); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -81,20 +80,20 @@ public async Task Send_completes_only_after_the_entire_wire_frame_is_reported() var send = transport.SendAsync(sessionId, frame, cancellation.Token).AsTask(); - Assert.Single(connection.Sent); + Assert.Equal(1, connection.SentCount); + Assert.Equal(sessionId.Value, connection.GetSentSessionId(0)); + Assert.Equal(0, connection.OrdinarySendCount); Assert.False(send.IsCompleted); - connection.ReportSent(new byte[16]); - Assert.False(send.IsCompleted); - connection.ReportSent(new byte[1]); + connection.CompleteSend(0); await send; await events.DisposeAsync().ConfigureAwait(true); } [Fact] - public async Task Concurrent_sends_are_serialized_until_each_write_confirmation() + public async Task Concurrent_sends_delegate_to_native_session_queue() { var connection = new FakeStreamConnection(); - await using var transport = CreateTransport(connection, new ManualHsmsTransportTimerFactory()); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -105,13 +104,15 @@ public async Task Concurrent_sends_are_serialized_until_each_write_confirmation( var first = transport.SendAsync(sessionId, frame, cancellation.Token).AsTask(); var second = transport.SendAsync(sessionId, frame, cancellation.Token).AsTask(); - Assert.Single(connection.Sent); - connection.ReportSent(new byte[14]); + Assert.Equal(2, connection.SentCount); + Assert.False(first.IsCompleted); + Assert.False(second.IsCompleted); + connection.CompleteSend(0); await first; - await WaitUntilAsync(() => connection.Sent.Count == 2); Assert.False(second.IsCompleted); - connection.ReportSent(new byte[14]); + connection.CompleteSend(1); await second; + Assert.Equal(0, connection.OrdinarySendCount); await events.DisposeAsync().ConfigureAwait(true); } @@ -119,7 +120,7 @@ public async Task Concurrent_sends_are_serialized_until_each_write_confirmation( public async Task Closing_session_fails_pending_send_and_prevents_reuse() { var connection = new FakeStreamConnection(); - await using var transport = CreateTransport(connection, new ManualHsmsTransportTimerFactory()); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -132,21 +133,21 @@ public async Task Closing_session_fails_pending_send_and_prevents_reuse() var exception = await Assert.ThrowsAsync( async () => await pending.ConfigureAwait(true)).ConfigureAwait(true); Assert.Equal(sessionId, exception.SessionId); + Assert.IsType(exception.InnerException); await Assert.ThrowsAsync( async () => await transport.SendAsync( sessionId, CreateFrame(), cancellation.Token).ConfigureAwait(true)).ConfigureAwait(true); - Assert.Single(connection.Sent); + Assert.Equal(1, connection.SentCount); await events.DisposeAsync().ConfigureAwait(true); } [Fact] - public async Task T8_timeout_reconnects_and_reports_close_reason() + public async Task Native_incomplete_frame_timeout_reports_HSMS_T8_reason() { var connection = new FakeStreamConnection(); - var timerFactory = new ManualHsmsTransportTimerFactory(); - await using var transport = CreateTransport(connection, timerFactory); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -157,8 +158,9 @@ public async Task T8_timeout_reconnects_and_reports_close_reason() CreateFrame(), cancellation.Token).AsTask(); - connection.ReportReceived(new byte[] { 0x00, 0x00 }); - timerFactory.Timer!.Fire(); + connection.RaiseFrameError(FrameErrorKind.IncompleteFrameTimeout); + Assert.Equal(0, connection.ReconnectCount); + connection.Reconnect(); var closed = await NextAsync(events); var sendException = await Assert.ThrowsAsync( async () => await pendingSend.ConfigureAwait(true)).ConfigureAwait(true); @@ -173,62 +175,52 @@ public async Task T8_timeout_reconnects_and_reports_close_reason() } [Fact] - public async Task Queued_T8_callback_from_previous_session_does_not_close_current_session() + public async Task Other_native_frame_errors_do_not_become_T8_close_reasons() { var connection = new FakeStreamConnection(); - var timerFactory = new ManualHsmsTransportTimerFactory(); - await using var transport = CreateTransport(connection, timerFactory); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); connection.RaiseState(ConnectionState.Connected); - await NextAsync(events); + var opened = await NextAsync(events); - connection.ReportReceived(new byte[] { 0x00 }); - var previousSessionTimer = timerFactory.Timer!; + connection.RaiseFrameError(FrameErrorKind.DecodeFailed); connection.RaiseState(ConnectionState.Retry); - await NextAsync(events); - connection.RaiseState(ConnectionState.Connecting); - connection.RaiseState(ConnectionState.Connected); - var currentSessionId = (await NextAsync(events)).SessionId; - connection.ReportReceived(new byte[] { 0x00, 0x00 }); - var currentSessionTimer = timerFactory.Timer!; - - previousSessionTimer.ForceFire(); - Assert.Equal(0, connection.ReconnectCount); - Assert.True(currentSessionTimer.IsArmed); - - currentSessionTimer.Fire(); var closed = await NextAsync(events); - Assert.Equal(1, connection.ReconnectCount); - Assert.Equal(currentSessionId, closed.SessionId); - var timeout = Assert.IsType(closed.Error); - Assert.Equal(currentSessionId, timeout.SessionId); + + Assert.Equal(opened.SessionId, closed.SessionId); + Assert.Null(closed.Error); await events.DisposeAsync().ConfigureAwait(true); } [Fact] - public async Task Explicit_close_reconnects_current_session_and_reports_reason() + public async Task Explicit_close_reconnects_current_session_and_preserves_reason() { var connection = new FakeStreamConnection(); - await using var transport = CreateTransport( - connection, - new ManualHsmsTransportTimerFactory()); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); connection.RaiseState(ConnectionState.Connected); var opened = await NextAsync(events); + var pendingSend = transport.SendAsync( + opened.SessionId, + CreateFrame(), + cancellation.Token).AsTask(); var reason = new HsmsProtocolException("Invalid control message."); var closedCurrent = transport.TryCloseSession(opened.SessionId, reason); var closed = await NextAsync(events); + var sendException = await Assert.ThrowsAsync( + async () => await pendingSend.ConfigureAwait(true)).ConfigureAwait(true); Assert.True(closedCurrent); Assert.Equal(1, connection.ReconnectCount); Assert.Equal(HsmsTransportEventKind.SessionClosed, closed.Kind); Assert.Equal(opened.SessionId, closed.SessionId); Assert.Same(reason, closed.Error); + Assert.Same(reason, sendException); await events.DisposeAsync().ConfigureAwait(true); } @@ -236,9 +228,7 @@ public async Task Explicit_close_reconnects_current_session_and_reports_reason() public async Task Explicit_close_for_expired_session_does_not_reconnect_current_session() { var connection = new FakeStreamConnection(); - await using var transport = CreateTransport( - connection, - new ManualHsmsTransportTimerFactory()); + await using var transport = CreateTransport(connection); using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); @@ -260,36 +250,11 @@ public async Task Explicit_close_for_expired_session_does_not_reconnect_current_ currentSessionId, CreateFrame(), cancellation.Token).AsTask(); - connection.ReportSent(new byte[14]); + connection.CompleteSend(0); await currentSend.ConfigureAwait(true); await events.DisposeAsync().ConfigureAwait(true); } - [Fact] - public async Task Idle_or_complete_frames_do_not_arm_T8_or_reconnect() - { - var connection = new FakeStreamConnection(); - var timerFactory = new ManualHsmsTransportTimerFactory(); - await using var transport = CreateTransport(connection, timerFactory); - using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(5)); - var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); - transport.Start(cancellation.Token); - connection.RaiseState(ConnectionState.Connected); - await NextAsync(events); - - Assert.Empty(timerFactory.Timers); - connection.ReportReceived( - new byte[] - { - 0x00, 0x00, 0x00, 0x0A, - 0x00, 0x01, 0x01, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, - }); - - Assert.Empty(timerFactory.Timers); - Assert.Equal(0, connection.ReconnectCount); - await events.DisposeAsync().ConfigureAwait(true); - } - [Fact] public async Task Real_streamframe_connection_sends_and_receives_fragmented_frame() { @@ -298,13 +263,15 @@ public async Task Real_streamframe_connection_sends_and_receives_fragmented_fram IPAddress.Loopback, port, isActive: false, - new HsmsTransportOptions(T5, T8), + new HsmsTransportOptions( + T5, + TimeSpan.FromMilliseconds(1000)), new StreamConnectionOptions { AcceptRetryDelayMs = 10, ConnectRetryDelayMs = 10, }); - using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(10)); + using var cancellation = new CancellationTokenSource(TimeSpan.FromSeconds(20)); var events = transport.GetEventsAsync(cancellation.Token).GetAsyncEnumerator(); transport.Start(cancellation.Token); using var client = new System.Net.Sockets.TcpClient(); @@ -312,6 +279,8 @@ public async Task Real_streamframe_connection_sends_and_receives_fragmented_fram var opened = await NextAsync(events); var frame = CreateFrame(); + await Task.Delay(TimeSpan.FromMilliseconds(1200), cancellation.Token); + var send = transport.SendAsync(opened.SessionId, frame, cancellation.Token).AsTask(); var sentBytes = await ReadExactlyAsync( client.GetStream(), @@ -328,28 +297,25 @@ public async Task Real_streamframe_connection_sends_and_receives_fragmented_fram sentBytes); var stream = client.GetStream(); - await stream.WriteAsync( - sentBytes, - 0, - 2, - cancellation.Token); - await Task.Delay(TimeSpan.FromMilliseconds(20), cancellation.Token); - await stream.WriteAsync( + await WriteFragmentedWithProgressAsync( + stream, sentBytes, - 2, - 3, - cancellation.Token); - await Task.Delay(TimeSpan.FromMilliseconds(20), cancellation.Token); - await stream.WriteAsync( - sentBytes, - 5, - sentBytes.Length - 5, cancellation.Token); var received = await NextAsync(events); Assert.Equal(HsmsTransportEventKind.FrameReceived, received.Kind); Assert.Equal(opened.SessionId, received.SessionId); Assert.Equal(frame.Header, received.Frame!.Header); + + await Task.Delay(TimeSpan.FromMilliseconds(1200), cancellation.Token); + await stream.WriteAsync( + sentBytes, + 0, + sentBytes.Length, + cancellation.Token); + var receivedAfterIdle = await NextAsync(events); + Assert.Equal(HsmsTransportEventKind.FrameReceived, receivedAfterIdle.Kind); + Assert.Equal(opened.SessionId, receivedAfterIdle.SessionId); await events.DisposeAsync().ConfigureAwait(true); } @@ -391,13 +357,8 @@ await client.GetStream().WriteAsync( } private static StreamFrameHsmsTransport CreateTransport( - FakeStreamConnection connection, - ManualHsmsTransportTimerFactory timerFactory) - => new( - connection, - new HsmsTransportSessionContext(), - new HsmsTransportOptions(T5, T8), - timerFactory); + FakeStreamConnection connection) + => new(connection); private static int GetFreePort() { @@ -431,6 +392,22 @@ private static async Task ReadExactlyAsync( return bytes; } + private static async Task WriteFragmentedWithProgressAsync( + Stream stream, + byte[] bytes, + CancellationToken cancellationToken) + { + await stream.WriteAsync(bytes, 0, 2, cancellationToken).ConfigureAwait(true); + await Task.Delay(TimeSpan.FromMilliseconds(600), cancellationToken).ConfigureAwait(true); + await stream.WriteAsync(bytes, 2, 3, cancellationToken).ConfigureAwait(true); + await Task.Delay(TimeSpan.FromMilliseconds(600), cancellationToken).ConfigureAwait(true); + await stream.WriteAsync( + bytes, + 5, + bytes.Length - 5, + cancellationToken).ConfigureAwait(true); + } + private static HsmsFrame CreateFrame(ReadOnlyMemory body = default) => new( HsmsMessageHeader.CreateData(1, 1, 1, false, 1), @@ -443,23 +420,17 @@ private static async Task NextAsync( return events.Current; } - private static async Task WaitUntilAsync(Func predicate) + private sealed class FakeStreamConnection : ISessionAwareStreamConnection { - for (var index = 0; index < 100; index++) - { - if (predicate()) - return; - - await Task.Yield(); - } - - Assert.True(predicate()); - } - - private sealed class FakeStreamConnection : IStreamConnection - { - private readonly Channel _messages = - Channel.CreateUnbounded(); + private readonly Channel> _messages = + Channel.CreateUnbounded>(); +#if NET9_0_OR_GREATER + private readonly Lock _gate = new(); +#else + private readonly object _gate = new(); +#endif + private readonly List _sent = new(); + private long _sessionCounter; public ConnectionState State { get; private set; } = ConnectionState.Connecting; @@ -471,21 +442,28 @@ private sealed class FakeStreamConnection : IStreamConnection IPAddress.Loopback.ToString(); - public event EventHandler? ConnectionChanged; + public long CurrentSessionId { get; private set; } + + public int ReconnectCount { get; private set; } - public event EventHandler? FrameError + public int OrdinarySendCount { get; private set; } + + public int SentCount { - add { } - remove { } + get + { + lock (_gate) + return _sent.Count; + } } - public Action>? RawBytesReceived { get; set; } + public event EventHandler? ConnectionChanged; - public Action>? RawBytesSent { get; set; } + public event EventHandler? FrameError; - public List Sent { get; } = new(); + public Action>? RawBytesReceived { get; set; } - public int ReconnectCount { get; private set; } + public Action>? RawBytesSent { get; set; } public void Start(CancellationToken ct) { @@ -500,14 +478,49 @@ public void Reconnect() public Task WaitForConnectedAsync(CancellationToken ct = default) => Task.CompletedTask; - public Task SendAsync(HsmsTransportFrame message, CancellationToken ct = default) + public Task SendAsync(HsmsFrame message, CancellationToken ct = default) { ct.ThrowIfCancellationRequested(); - Sent.Add(message); + OrdinarySendCount++; return Task.CompletedTask; } - public async IAsyncEnumerable GetMessages( + public Task SendInSessionAsync( + long sessionId, + HsmsFrame message, + CancellationToken ct = default) + { + if (ct.IsCancellationRequested) + return Task.FromCanceled(ct); + + lock (_gate) + { + if (CurrentSessionId != sessionId) + { + return Task.FromException( + new SessionExpiredException( + sessionId, + $"Session {sessionId} expired.")); + } + + var completion = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + _sent.Add(new PendingSessionSend(sessionId, message, completion)); + return completion.Task; + } + } + + public async IAsyncEnumerable GetMessages( + [EnumeratorCancellation] CancellationToken ct = default) + { + while (await _messages.Reader.WaitToReadAsync(ct).ConfigureAwait(false)) + { + while (_messages.Reader.TryRead(out var message)) + yield return message.Message; + } + } + + public async IAsyncEnumerable> GetSessionMessages( [EnumeratorCancellation] CancellationToken ct = default) { while (await _messages.Reader.WaitToReadAsync(ct).ConfigureAwait(false)) @@ -526,17 +539,63 @@ public ValueTask DisposeAsync() public void RaiseState(ConnectionState state) { - State = state; + long expiredSessionId = 0; + lock (_gate) + { + if (state == ConnectionState.Connected && + State != ConnectionState.Connected) + { + CurrentSessionId = checked(++_sessionCounter); + } + else if (state != ConnectionState.Connected && + State == ConnectionState.Connected) + { + expiredSessionId = CurrentSessionId; + CurrentSessionId = 0; + } + + State = state; + if (expiredSessionId > 0) + { + foreach (var pending in _sent) + { + if (pending.SessionId == expiredSessionId) + { + pending.Completion.TrySetException( + new SessionExpiredException( + expiredSessionId, + $"Session {expiredSessionId} expired.")); + } + } + } + } + ConnectionChanged?.Invoke(this, state); } - public void Emit(HsmsTransportFrame frame) - => _messages.Writer.TryWrite(frame); + public void RaiseFrameError(FrameErrorKind kind) + => FrameError?.Invoke( + this, + new FrameErrorEventArgs(kind, ReadOnlyMemory.Empty)); - public void ReportReceived(ReadOnlyMemory bytes) - => RawBytesReceived?.Invoke(bytes); + public void Emit(long sessionId, HsmsFrame frame) + => _messages.Writer.TryWrite(new SessionMessage(sessionId, frame)); + + public long GetSentSessionId(int index) + { + lock (_gate) + return _sent[index].SessionId; + } + + public void CompleteSend(int index) + { + lock (_gate) + _sent[index].Completion.TrySetResult(true); + } - public void ReportSent(ReadOnlyMemory bytes) - => RawBytesSent?.Invoke(bytes); + private sealed record PendingSessionSend( + long SessionId, + HsmsFrame Message, + TaskCompletionSource Completion); } }