diff --git a/CHANGELOG.md b/CHANGELOG.md index 5b91c5f..e396ee2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,5 +42,10 @@ StreamFrame 重试间隔并关闭指数退避，Passive 监听重试保持独立。 - 增加按接收进展与会话代次隔离的 T8 监视、专用超时异常，以及部分长度 头、部分 Payload、陈旧回调、替换会话和真实 TCP 超时测试。 +- 增加公共 HsmsConnection 与不可变 + HsmsConnectionOptions，组合动态消息收发、Selected 等待、 + 单消费者事件流、入站一次性回复和控制命令。 +- 增加公共 API 选项边界、生命周期、等待/事务取消和 Active/Passive + 双端真实 TCP 往返测试；入站回复令牌绑定原连接与原运输会话。 [Unreleased]: https://github.com/CSJ608/SecsFrame/compare/main...HEAD diff --git a/README.md b/README.md index 5d26cc5..c5dbfc4 100644 --- a/README.md +++ b/README.md @@ -28,13 +28,16 @@ 共用的内部 HSMS-SS 状态机已经覆盖 TCP 会话、Select、Linktest、Reject、 Deselect、Separate、T6/T7 和 Selected 数据门控。内部数据事务 actor 进一步提供实际写出后启动的 T3、复合事务键关联、入站回复和会话失效 -隔离。Item 用法与边界见 +隔离。公共 HsmsConnection 将这些内部层组合成显式计时 +配置、动态消息收发、状态等待和单消费者事件流。Item 用法与边界见 [docs/SECS-II-ITEMS.md](docs/SECS-II-ITEMS.md)，消息集成见 [docs/SECS-MESSAGES.md](docs/SECS-MESSAGES.md)，传输适配边界见 [docs/STREAMFRAME-ADAPTER.md](docs/STREAMFRAME-ADAPTER.md)，状态机边界见 [docs/HSMS-SESSION-STATE-MACHINE.md](docs/HSMS-SESSION-STATE-MACHINE.md)， 数据事务边界见 [docs/HSMS-DATA-TRANSACTIONS.md](docs/HSMS-DATA-TRANSACTIONS.md)， +公共连接用法见 +[docs/HSMS-CONNECTION.md](docs/HSMS-CONNECTION.md)， 整体路线图见 [docs/ROADMAP.md](docs/ROADMAP.md)。 diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 31a65d0..93b541d 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -120,6 +120,18 @@ HSMS 头 Session ID 与 System Bytes。没有 W-Bit 的出站消息在写出后 和测试边界见 [HSMS-DATA-TRANSACTIONS.md](HSMS-DATA-TRANSACTIONS.md)。 +### 公共连接外观 + +HsmsConnection 是运输、会话和数据事务 actor 之上的薄公共 +组合层。它要求显式网络、协议 Session ID 和 T3/T5/T6/T7/T8 配置，公开 +动态消息发送、一次性入站回复、控制命令、Selected 等待和单消费者事件 +流。公共层不暴露 transport Session generation、StreamFrame +StreamConnectionOptions 或 #38/#39 的回调适配。 + +连接内部独占消费事务事件，并分别更新状态等待信号与公共事件 channel， +因此 readiness 等待不会与业务消息消费者竞争。详细生命周期和取消语义 +见 [HSMS-CONNECTION.md](HSMS-CONNECTION.md)。 + ### 严格与兼容分离 默认按标准拒绝畸形帧、非法 Item 和无效状态转换。对现场设备的已知偏差通过命名明确的兼容选项启用，并记录来源、风险和测试向量。 diff --git a/docs/COMPATIBILITY.md b/docs/COMPATIBILITY.md index 6c5c09b..6c7fc49 100644 --- a/docs/COMPATIBILITY.md +++ b/docs/COMPATIBILITY.md @@ -32,3 +32,13 @@ StreamFrame #38（未完成帧超时）和 #39（会话感知发送确认与消 具体机制、测试证据和已知代价见 [STREAMFRAME-ADAPTER.md](STREAMFRAME-ADAPTER.md)。 + +## 跨实现测试依赖 + +计划中的 secs4net 互操作夹具只能使用 nuget.org 发布的官方 NuGet 包， +并在 Directory.Packages.props 中固定明确版本。不得使用 +本地源码 checkout、ProjectReference、本地 feed 或自行修改的程序集作为 +互操作对端。源码仓库只读参考不进入 SecsFrame 的还原、构建或测试图。 + +夹具和文档必须记录官方包 ID、版本、运行框架、测试方向与通过/失败结果， +避免把本地差异误判为上游互操作能力。 diff --git a/docs/HSMS-CONNECTION.md b/docs/HSMS-CONNECTION.md new file mode 100644 index 0000000..bfe8192 --- /dev/null +++ b/docs/HSMS-CONNECTION.md @@ -0,0 +1,104 @@ +# 公共 HSMS 连接 API + +## 范围 + +HsmsConnection 是当前基础层的最小公共入口。它在内部组合 +StreamFrame 运输适配、HSMS-SS 会话 actor 和数据事务 actor，调用方不需要 +接触 transport Session generation、发送确认回调或内部 channel。 + +当前公共能力包括： + +- Active 或 Passive TCP 连接以及 Select 会话建立； +- 等待 Selected 状态和读取当前状态； +- 发送任意动态 SecsMessage，无需消息目录或预生成类型； +- W-Bit Primary 的 T3 关联与 Secondary 返回； +- 接收入站 Primary 或未匹配消息，并使用不可伪造令牌回复一次； +- Linktest、Deselect 和 Separate 控制命令； +- 状态、数据、未消费控制消息和解码失败的单消费者异步事件流。 + +Host/Equipment 是后续业务能力角色，不由 Active/Passive 推断。 + +## 显式配置 + +HsmsConnectionOptions 构造时必须给出： + +| 配置 | 当前用途 | +|---|---| +| IP 地址、端口 | Active 的远端地址或 Passive 的本地监听地址 | +| ConnectionMode | 只表示哪一端建立 TCP 连接 | +| SessionId | 出站 Data Message 使用的协议 Session ID | +| T3 | W-Bit Primary 写完后等待 Secondary | +| T5 | Active TCP 连接失败后的固定重试间隔 | +| T6 | Select、Linktest、Deselect 控制响应等待 | +| T7 | TCP 建立后等待进入 Selected | +| T8 | 未完成消息接收进展等待 | + +仓库没有内置或暗示标准默认值。T5 必须能无损转换成 StreamFrame 使用的 +整毫秒范围；Passive 监听重试不被解释成 T5。每个取值应依据团队合法获得 +的 E37/E37.1 版本、设备通信手册和互操作证据确定。 + +## 生命周期 + +~~~csharp +var options = new HsmsConnectionOptions( + IPAddress.Parse("192.0.2.10"), + port: 5000, + HsmsConnectionMode.Active, + sessionId: 0, + t3: TimeSpan.FromSeconds(45), + t5: TimeSpan.FromSeconds(10), + t6: TimeSpan.FromSeconds(5), + t7: TimeSpan.FromSeconds(10), + t8: TimeSpan.FromSeconds(5)); + +await using var connection = new HsmsConnection(options); +connection.Start(); +await connection.WaitUntilSelectedAsync(cancellationToken); + +var secondary = await connection.SendAsync( + new SecsMessage( + stream: 1, + function: 1, + replyExpected: true), + cancellationToken); +~~~ + +以上计时值只演示 API 形状，不是标准推荐值。 + +Start 只能调用一次，连接活动期由 +DisposeAsync 结束。WaitUntilSelectedAsync +不消费业务事件；取消它只取消本次等待，不停止连接。 + +GetEventsAsync 同一时刻只允许一个消费者；枚举器结束或释放 +后可以重新获取。连接内部先独占消费事务 actor 的事件，再写入公共事件 +流，因此 Selected 等待不会和业务接收争抢消息。调用方通常应使用一个 +长期事件循环处理： + +- StateChanged：状态转换和可选错误； +- DataMessageReceived：携带 + HsmsIncomingDataMessage； +- ControlMessageReceived：未被当前控制或数据事务认领的帧； +- DataMessageDecodeFailed：原始帧和严格解码错误。 + +入站消息只有 ReplyExpected 为真时可以回复。回复自动复制原 +协议 Session ID 与 System Bytes，并绑定原 transport session；开始回复 +后令牌不可再次使用，也不会在替换连接上重放。 + +## 取消与失败 + +- 出站无 W-Bit 消息在整帧写完后完成； +- 出站 W-Bit Primary 在整帧写完后启动 T3； +- 取消发送会结束本地等待，不会自动关闭 Selected 会话，也不能撤回已经 + 上线的字节；迟到 Secondary 作为未匹配入站消息上报； +- Deselect、Separate、断线、替换会话或释放会结束绑定旧会话的等待； +- 事件枚举的取消只结束该消费者；连接仍须显式释放。 + +当前具体异常类型仍以 TimeoutException、 +IOException、OperationCanceledException 和参数/ +状态异常为稳定分类；更细的公共诊断模型将在互操作证据后再收敛。 + +## 验证边界 + +真实 TCP 测试使用两个公共连接完成 Select、Linktest、嵌套 Item +Primary/Secondary、重复回复拒绝和取消后保持 Selected。它证明当前实现 +各层能够组合工作，不替代 SEMI 一致性认证或第三方设备互操作。 diff --git a/docs/HSMS-DATA-TRANSACTIONS.md b/docs/HSMS-DATA-TRANSACTIONS.md index 8676a25..4b7a21e 100644 --- a/docs/HSMS-DATA-TRANSACTIONS.md +++ b/docs/HSMS-DATA-TRANSACTIONS.md @@ -86,8 +86,12 @@ Active/Passive 双端真实 TCP 回环另行覆盖 Select 后的 Primary/Seconda ## 标准核对边界 -实现追踪 SEMI E5-0725、E37-0222 与 E37.1-0819。发布公共事务 API 前， -仍须使用团队合法获得的版本核对 Primary/Secondary 定义、W-Bit 与 -Function 约束、T3 的精确启停、SxF0/S9Fx 对事务的影响、System Bytes -分配规则及 Reject 后处理。仓库不得提交标准正文、状态表或由付费材料 -机械转换的测试资料，也不据当前工程测试声明完整合规。 +实现追踪 SEMI E5-0725、E37-0222 与 E37.1-0819。扩大公共事务 API 或 +声明合规前，仍须使用团队合法获得的版本核对 Primary/Secondary 定义、 +W-Bit 与 Function 约束、T3 的精确启停、SxF0/S9Fx 对事务的影响、 +System Bytes 分配规则及 Reject 后处理。仓库不得提交标准正文、状态表 +或由付费材料机械转换的测试资料，也不据当前工程测试声明完整合规。 + +HsmsDataTransactionManager 仍是内部 actor；公共调用方通过 +HsmsConnection 使用其动态发送、入站回复和事件能力。公共 +生命周期与取消语义见 [HSMS-CONNECTION.md](HSMS-CONNECTION.md)。 diff --git a/docs/HSMS-SESSION-STATE-MACHINE.md b/docs/HSMS-SESSION-STATE-MACHINE.md index faa16e0..c7a0984 100644 --- a/docs/HSMS-SESSION-STATE-MACHINE.md +++ b/docs/HSMS-SESSION-STATE-MACHINE.md @@ -77,13 +77,15 @@ System Bytes。旧会话的帧、发送完成和计时器回调不会推进替 StreamFrameHsmsTransport 完成 Select、Linktest 和 Deselect 流程；其上的事务回环还完成带嵌套 Item 的 Primary/Secondary 往返。数据 事务细节见 [HSMS-DATA-TRANSACTIONS.md](HSMS-DATA-TRANSACTIONS.md)。 +公共 HsmsConnection 另行验证同一路径的外部组合契约，见 +[HSMS-CONNECTION.md](HSMS-CONNECTION.md)。 测试证明当前实现路径可互操作，不替代认证工具或第三方设备互操作证据。 ## 尚未实现 - 授权标准核对后的 T5/T8 默认值和完整标准语义； - 自动周期 Linktest 调度； -- 公共会话 API、Host/Equipment 能力 API； +- Host/Equipment 能力 API； - 与 secs4net 或真实设备的跨实现互操作矩阵。 实现这些行为前，需要使用团队合法获得的 SEMI E37-0222 与 E37.1-0819 diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index ebcd755..3c6b2c0 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -16,14 +16,17 @@ - [x] Linktest、Reject、Deselect 控制消息与对应 T6 分支。 - [x] Selected 数据发送、Primary/Secondary 关联、T3 与入站同事务回复。 - [x] 显式 T5 固定重连节流、T8 接收进展监视和跨会话旧回调隔离。 +- [x] 最小公共连接外观、显式 T3/T5/T6/T7/T8 配置、动态消息事件与 + 确定的取消/释放语义。 - [ ] 使用授权 E37/E37.1 核对 T5/T8 默认值与精确启停边界。 -- Host、Equipment 双角色回环集成测试，以及与 secs4net 的互操作测试。 - -阶段 1 的下一最小切片是稳定最小公共连接外观：公开动态消息收发、会话 -状态、显式 T3/T5/T6/T7/T8 配置和确定的取消/关闭语义，同时继续把 -StreamFrame #38/#39 细节隔离在内部。随后建立与 secs4net 的跨实现 -Select、Linktest 和 Primary/Secondary 互操作夹具。任何标准默认值和 -完整合规结论都依赖团队合法获得的 E37/E37.1 副本与一致性测试。 +- Host、Equipment 双角色回环集成测试，以及与 nuget.org 官方 secs4net + 包的互操作测试。 + +阶段 1 的下一最小切片是跨实现互操作夹具：通过中央包管理固定 +nuget.org 官方 secs4net 包版本，覆盖 Select、Linktest 和 +Primary/Secondary。不得引用、构建或链接本地 secs4net 源码 checkout。 +随后根据互操作结果收敛公共诊断模型。任何标准默认值和完整合规结论都 +依赖团队合法获得的 E37/E37.1 副本与一致性测试。 ## 阶段 2：GEM 通用能力 diff --git a/docs/STANDARDS.md b/docs/STANDARDS.md index 22c4562..eac65ce 100644 --- a/docs/STANDARDS.md +++ b/docs/STANDARDS.md @@ -56,5 +56,7 @@ - 未经验证的“完整合规”“通过认证”等声明。 本项目只读参考 StreamFrame 和 MIT 许可的 secs4net 以交叉理解公开协议 -实现，没有复制参考仓库代码。未来如引入实质性开源代码，必须保留对应 -版权与许可证文本，并在变更说明中记录来源。 +实现，没有复制参考仓库代码。该 secs4net 源码 checkout 不参与互操作 +构建；互操作夹具只允许使用 nuget.org 发布并固定版本的官方 NuGet 包。 +未来如引入实质性开源代码，必须保留对应版权与许可证文本，并在变更说明 +中记录来源。 diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnection.cs b/src/SecsFrame/Hsms/Connection/HsmsConnection.cs new file mode 100644 index 0000000..ac4509d --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsConnection.cs @@ -0,0 +1,393 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; + +namespace SecsFrame; + +/// +/// Provides dynamic SECS-II messaging over one HSMS-SS connection. +/// +/// +/// This type owns its transport, session state machine, and data transaction +/// manager. Call once and dispose the connection to stop it. +/// +public sealed class HsmsConnection : IAsyncDisposable +{ + private readonly HsmsDataTransactionManager _transactions; + private readonly Channel _events; +#if NET9_0_OR_GREATER + private readonly Lock _stateGate = new(); +#else + private readonly object _stateGate = new(); +#endif + private TaskCompletionSource _selectedSignal = + CreateSelectionSignal(); + private Task? _eventPump; + private int _state = (int)HsmsSessionState.Disconnected; + private int _eventReaderClaimed; + private int _started; + private int _disposed; + + /// Creates an HSMS connection without starting network activity. + /// The explicit connection and timer settings. + public HsmsConnection(HsmsConnectionOptions options) + { + Options = options ?? throw new ArgumentNullException(nameof(options)); + _transactions = CreateTransactions(options); + _events = Channel.CreateUnbounded( + new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = true, + AllowSynchronousContinuations = false, + }); + } + + /// Gets the immutable connection settings. + public HsmsConnectionOptions Options { get; } + + /// Gets the latest observed HSMS session state. + public HsmsSessionState State + => (HsmsSessionState)Volatile.Read(ref _state); + + /// Starts network, session, transaction, and event processing. + public void Start() + { + ThrowIfDisposed(); + if (Interlocked.Exchange(ref _started, 1) != 0) + throw new InvalidOperationException("The HSMS connection has already been started."); + + try + { + _transactions.Start(CancellationToken.None); + _eventPump = PumpEventsAsync(); + } + catch (Exception ex) + { + FailSelectionWaiters(ex); + _events.Writer.TryComplete(ex); + throw; + } + } + + /// + /// Gets the single-consumer stream of state, incoming-message, control, and + /// decoding events. + /// + /// Cancels only this event-stream read. + /// + /// Only one event stream can be active at a time. A new stream can be + /// acquired after the previous enumerator ends or is disposed. Use + /// for readiness checks without + /// consuming state events. + /// + public IAsyncEnumerable GetEventsAsync( + CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + return ReadEventsAsync(cancellationToken); + } + + /// Waits until this connection next reaches Selected state. + /// Cancels only this wait operation. + public Task WaitUntilSelectedAsync( + CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + cancellationToken.ThrowIfCancellationRequested(); + + Task selected; + lock (_stateGate) + { + if ((HsmsSessionState)_state == HsmsSessionState.Selected) + return Task.CompletedTask; + + selected = _selectedSignal.Task; + } + + return AwaitSelectionAsync(selected, cancellationToken); + } + + /// + /// Sends a dynamic Primary message with the configured protocol Session ID. + /// + /// The message to send. + /// Cancels this send or open transaction. + /// + /// The matching Secondary when the W-Bit is set; otherwise + /// after the entire frame is written. + /// + public Task SendAsync( + SecsMessage primary, + CancellationToken cancellationToken = default) + { + if (primary is null) + throw new ArgumentNullException(nameof(primary)); + + ThrowIfNotRunning(); + return _transactions.SendAsync( + Options.SessionId, + primary, + cancellationToken); + } + + /// Replies once to an incoming Primary on its original session. + /// The incoming message token from the event stream. + /// The Secondary message, with W-Bit clear. + /// Cancels this reply operation. + public Task ReplyAsync( + HsmsIncomingDataMessage incoming, + SecsMessage secondary, + CancellationToken cancellationToken = default) + { + if (incoming is null) + throw new ArgumentNullException(nameof(incoming)); + if (secondary is null) + throw new ArgumentNullException(nameof(secondary)); + + ThrowIfNotRunning(); + return _transactions.ReplyAsync( + incoming, + secondary, + cancellationToken); + } + + /// Sends Linktest Request and waits for its matching response. + /// Cancels this control command. + public Task LinktestAsync(CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + return _transactions.LinktestAsync(cancellationToken); + } + + /// Sends Deselect Request and waits for its matching response. + /// Cancels this control command. + public Task DeselectAsync(CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + return _transactions.DeselectAsync(cancellationToken); + } + + /// Sends Separate Request and closes the selected TCP session. + /// Cancels this control command. + public Task SeparateAsync(CancellationToken cancellationToken = default) + { + ThrowIfNotRunning(); + return _transactions.SeparateAsync(cancellationToken); + } + + /// Stops the connection and completes pending operations and events. + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + FailSelectionWaiters(new ObjectDisposedException(nameof(HsmsConnection))); + await _transactions.DisposeAsync().ConfigureAwait(false); + if (_eventPump is not null) + await _eventPump.ConfigureAwait(false); + else + _events.Writer.TryComplete(); + } + + private static HsmsDataTransactionManager CreateTransactions( + HsmsConnectionOptions options) + { + var isActive = options.ConnectionMode == HsmsConnectionMode.Active; + var transport = StreamFrameHsmsTransport.Create( + options.IpAddress, + options.Port, + isActive, + new HsmsTransportOptions(options.T5, options.T8)); + var session = new HsmsSessionStateMachine( + transport, + new HsmsSessionOptions( + options.ConnectionMode, + options.T6, + options.T7)); + return new HsmsDataTransactionManager( + session, + new HsmsDataTransactionOptions(options.T3)); + } + + private async Task PumpEventsAsync() + { + Exception? completionError = null; + var events = _transactions.GetEventsAsync(CancellationToken.None) + .GetAsyncEnumerator(); + try + { + while (await events.MoveNextAsync().ConfigureAwait(false)) + { + var transactionEvent = events.Current; + if (transactionEvent.Kind == + HsmsDataTransactionEventKind.StateChanged) + { + ApplyState(transactionEvent.State); + } + + if (!_events.Writer.TryWrite(MapEvent(transactionEvent))) + break; + } + + if (Volatile.Read(ref _disposed) == 0) + { + completionError = new IOException( + "The internal HSMS event stream ended before the connection was disposed."); + FailSelectionWaiters(completionError); + } + } + catch (Exception ex) + { + completionError = ex; + FailSelectionWaiters(ex); + } + finally + { + await events.DisposeAsync().ConfigureAwait(false); + _events.Writer.TryComplete(completionError); + } + } + + private void ApplyState(HsmsSessionState state) + { + lock (_stateGate) + { + Volatile.Write(ref _state, (int)state); + if (Volatile.Read(ref _disposed) != 0) + return; + + if (state == HsmsSessionState.Selected) + { + _selectedSignal.TrySetResult(null); + } + else if (_selectedSignal.Task.IsCompleted) + { + _selectedSignal = CreateSelectionSignal(); + } + } + } + + private void FailSelectionWaiters(Exception error) + { + lock (_stateGate) + { + Volatile.Write( + ref _state, + (int)HsmsSessionState.Disconnected); + if (_selectedSignal.Task.IsCompleted) + _selectedSignal = CreateSelectionSignal(); + + _selectedSignal.TrySetResult(error); + } + } + + private async IAsyncEnumerable ReadEventsAsync( + [EnumeratorCancellation] CancellationToken cancellationToken) + { + if (Interlocked.Exchange(ref _eventReaderClaimed, 1) != 0) + { + throw new InvalidOperationException( + "The HSMS connection event stream already has a consumer."); + } + + try + { + var reader = _events.Reader; + while (await reader.WaitToReadAsync(cancellationToken).ConfigureAwait(false)) + { + while (reader.TryRead(out var connectionEvent)) + yield return connectionEvent; + } + } + finally + { + Interlocked.Exchange(ref _eventReaderClaimed, 0); + } + } + + private static HsmsConnectionEvent MapEvent( + HsmsDataTransactionEvent transactionEvent) + => transactionEvent.Kind switch + { + HsmsDataTransactionEventKind.StateChanged => + HsmsConnectionEvent.StateChanged( + transactionEvent.State, + transactionEvent.Error), + HsmsDataTransactionEventKind.DataMessageReceived => + HsmsConnectionEvent.DataMessageReceived( + transactionEvent.DataMessage!), + HsmsDataTransactionEventKind.ControlMessageReceived => + HsmsConnectionEvent.ControlMessageReceived( + transactionEvent.State, + transactionEvent.Frame!), + HsmsDataTransactionEventKind.DataMessageDecodeFailed => + HsmsConnectionEvent.DataMessageDecodeFailed( + transactionEvent.Frame!, + transactionEvent.Error!), + _ => throw new ArgumentOutOfRangeException( + nameof(transactionEvent), + transactionEvent.Kind, + "Unknown HSMS data transaction event."), + }; + + private static async Task AwaitSelectionAsync( + Task operation, + CancellationToken cancellationToken) + { + if (!cancellationToken.CanBeCanceled) + { + var error = await operation.ConfigureAwait(false); + if (error is not null) + throw error; + + return; + } + + var cancellation = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + using var registration = cancellationToken.Register( + static state => + { + var pair = (CancellationState)state!; + pair.Signal.TrySetCanceled(pair.Token); + }, + new CancellationState(cancellation, cancellationToken)); + var completed = await Task.WhenAny(operation, cancellation.Task) + .ConfigureAwait(false); + await completed.ConfigureAwait(false); + var selectionError = await operation.ConfigureAwait(false); + if (selectionError is not null) + throw selectionError; + } + + private void ThrowIfNotRunning() + { + ThrowIfDisposed(); + if (Volatile.Read(ref _started) == 0) + throw new InvalidOperationException("The HSMS connection has not been started."); + } + + private void ThrowIfDisposed() + { + if (Volatile.Read(ref _disposed) != 0) + throw new ObjectDisposedException(nameof(HsmsConnection)); + } + + private static TaskCompletionSource CreateSelectionSignal() + => new(TaskCreationOptions.RunContinuationsAsynchronously); + + private sealed class CancellationState + { + public CancellationState( + TaskCompletionSource signal, + CancellationToken token) + { + Signal = signal; + Token = token; + } + + public TaskCompletionSource Signal { get; } + + public CancellationToken Token { get; } + } +} diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnectionEvent.cs b/src/SecsFrame/Hsms/Connection/HsmsConnectionEvent.cs new file mode 100644 index 0000000..9a3d9ec --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsConnectionEvent.cs @@ -0,0 +1,73 @@ +namespace SecsFrame; + +/// An immutable event emitted by . +public sealed class HsmsConnectionEvent +{ + private HsmsConnectionEvent( + HsmsConnectionEventKind kind, + HsmsSessionState state, + HsmsIncomingDataMessage? incomingMessage, + HsmsFrame? frame, + Exception? error) + { + Kind = kind; + State = state; + IncomingMessage = incomingMessage; + Frame = frame; + Error = error; + } + + /// Gets the event kind. + public HsmsConnectionEventKind Kind { get; } + + /// Gets the session state when the event was produced. + public HsmsSessionState State { get; } + + /// Gets the incoming data message, when applicable. + public HsmsIncomingDataMessage? IncomingMessage { get; } + + /// Gets the control or undecodable data frame, when applicable. + public HsmsFrame? Frame { get; } + + /// Gets the state or decoding error, when applicable. + public Exception? Error { get; } + + internal static HsmsConnectionEvent StateChanged( + HsmsSessionState state, + Exception? error) + => new( + HsmsConnectionEventKind.StateChanged, + state, + null, + null, + error); + + internal static HsmsConnectionEvent DataMessageReceived( + HsmsIncomingDataMessage incomingMessage) + => new( + HsmsConnectionEventKind.DataMessageReceived, + HsmsSessionState.Selected, + incomingMessage, + null, + null); + + internal static HsmsConnectionEvent ControlMessageReceived( + HsmsSessionState state, + HsmsFrame frame) + => new( + HsmsConnectionEventKind.ControlMessageReceived, + state, + null, + frame, + null); + + internal static HsmsConnectionEvent DataMessageDecodeFailed( + HsmsFrame frame, + Exception error) + => new( + HsmsConnectionEventKind.DataMessageDecodeFailed, + HsmsSessionState.Selected, + null, + frame, + error); +} diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnectionEventKind.cs b/src/SecsFrame/Hsms/Connection/HsmsConnectionEventKind.cs new file mode 100644 index 0000000..9184d4e --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsConnectionEventKind.cs @@ -0,0 +1,17 @@ +namespace SecsFrame; + +/// Identifies an event emitted by . +public enum HsmsConnectionEventKind +{ + /// The HSMS session state changed. + StateChanged, + + /// An unmatched or primary data message was received. + DataMessageReceived, + + /// An unconsumed HSMS control message was received. + ControlMessageReceived, + + /// An HSMS data message could not be decoded. + DataMessageDecodeFailed, +} diff --git a/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs new file mode 100644 index 0000000..497851b --- /dev/null +++ b/src/SecsFrame/Hsms/Connection/HsmsConnectionOptions.cs @@ -0,0 +1,137 @@ +using System.Net; + +namespace SecsFrame; + +/// +/// Immutable network, protocol-session, and timer settings for an +/// . +/// +public sealed class HsmsConnectionOptions +{ + /// Creates explicit HSMS connection settings. + /// The remote address in Active mode or local bind address in Passive mode. + /// The TCP port, from 1 through 65535. + /// Which peer initiates the TCP connection. + /// The protocol Session ID used by outgoing data messages. + /// The data transaction reply timeout. + /// The Active connection retry interval. + /// The control transaction reply timeout. + /// The selection timeout after TCP connection. + /// The incomplete-message receive timeout. + public HsmsConnectionOptions( + IPAddress ipAddress, + int port, + HsmsConnectionMode connectionMode, + ushort sessionId, + TimeSpan t3, + TimeSpan t5, + TimeSpan t6, + TimeSpan t7, + TimeSpan t8) + { + IpAddress = ipAddress ?? throw new ArgumentNullException(nameof(ipAddress)); + if (port is < 1 or > ushort.MaxValue) + { + throw new ArgumentOutOfRangeException( + nameof(port), + port, + "The TCP port must be between 1 and 65535."); + } + + if (connectionMode != HsmsConnectionMode.Active && + connectionMode != HsmsConnectionMode.Passive) + { + throw new ArgumentOutOfRangeException( + nameof(connectionMode), + connectionMode, + "The HSMS connection mode must be Active or Passive."); + } + + if (sessionId == ushort.MaxValue) + { + throw new ArgumentOutOfRangeException( + nameof(sessionId), + sessionId, + "The protocol Session ID cannot use the control-message value 65535."); + } + + ValidatePositive(t3, nameof(t3), "T3"); + ValidateT5(t5, nameof(t5)); + ValidatePositive(t6, nameof(t6), "T6"); + ValidatePositive(t7, nameof(t7), "T7"); + ValidatePositive(t8, nameof(t8), "T8"); + + Port = port; + ConnectionMode = connectionMode; + SessionId = sessionId; + T3 = t3; + T5 = t5; + T6 = t6; + T7 = t7; + T8 = t8; + } + + /// Gets the remote or local bind IP address. + public IPAddress IpAddress { get; } + + /// Gets the TCP port. + public int Port { get; } + + /// Gets which peer initiates the TCP connection. + public HsmsConnectionMode ConnectionMode { get; } + + /// Gets the protocol Session ID for outgoing data messages. + public ushort SessionId { get; } + + /// Gets the data transaction reply timeout. + public TimeSpan T3 { get; } + + /// Gets the Active connection retry interval. + public TimeSpan T5 { get; } + + /// Gets the control transaction reply timeout. + public TimeSpan T6 { get; } + + /// Gets the selection timeout after TCP connection. + public TimeSpan T7 { get; } + + /// Gets the incomplete-message receive timeout. + public TimeSpan T8 { get; } + + private static void ValidatePositive( + TimeSpan value, + string parameterName, + string timerName) + { + if (value <= TimeSpan.Zero) + { + throw new ArgumentOutOfRangeException( + parameterName, + value, + $"{timerName} must be positive."); + } + } + + private static void ValidateT5( + TimeSpan value, + string parameterName) + { + ValidatePositive(value, parameterName, "T5"); + if (value.Ticks % TimeSpan.TicksPerMillisecond != 0) + { + throw new ArgumentOutOfRangeException( + parameterName, + value, + "T5 must be representable as a whole number of milliseconds."); + } + + var milliseconds = value.Ticks / TimeSpan.TicksPerMillisecond; + if (milliseconds > int.MaxValue) + { + throw new ArgumentOutOfRangeException( + parameterName, + value, + "T5 exceeds the supported connection retry-delay range."); + } + } +} diff --git a/src/SecsFrame/Hsms/Session/HsmsConnectionMode.cs b/src/SecsFrame/Hsms/Session/HsmsConnectionMode.cs index de07644..ae6d9ba 100644 --- a/src/SecsFrame/Hsms/Session/HsmsConnectionMode.cs +++ b/src/SecsFrame/Hsms/Session/HsmsConnectionMode.cs @@ -1,7 +1,15 @@ namespace SecsFrame; -internal enum HsmsConnectionMode +/// Specifies which peer initiates the TCP connection. +/// +/// The connection mode is independent of the Host or Equipment application +/// role. +/// +public enum HsmsConnectionMode { + /// The local peer initiates the TCP connection. Active, + + /// The local peer listens for the TCP connection. Passive, } diff --git a/src/SecsFrame/Hsms/Session/HsmsSessionState.cs b/src/SecsFrame/Hsms/Session/HsmsSessionState.cs index 69e7ffe..a4a2650 100644 --- a/src/SecsFrame/Hsms/Session/HsmsSessionState.cs +++ b/src/SecsFrame/Hsms/Session/HsmsSessionState.cs @@ -1,9 +1,17 @@ namespace SecsFrame; -internal enum HsmsSessionState +/// Describes the current HSMS transport and selection state. +public enum HsmsSessionState { + /// No TCP session is currently open. Disconnected, + + /// A TCP session is open but is not selected. Connected, + + /// An Active connection is waiting for Select Response. Selecting, + + /// The HSMS session is selected and can exchange data messages. Selected, } diff --git a/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs b/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs index b222c4f..10a3412 100644 --- a/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs +++ b/src/SecsFrame/Hsms/Transactions/HsmsDataTransactionManager.cs @@ -117,6 +117,12 @@ public Task ReplyAsync( throw new ArgumentNullException(nameof(incoming)); if (secondary is null) throw new ArgumentNullException(nameof(secondary)); + if (!incoming.IsOwnedBy(this)) + { + throw new InvalidOperationException( + "The incoming data message belongs to another HSMS connection."); + } + if (!incoming.DataMessage.Message.ReplyExpected) { throw new InvalidOperationException( @@ -350,6 +356,7 @@ ex is SecsProtocolException || } var incoming = new HsmsIncomingDataMessage( + this, sessionEvent.SessionId, dataMessage); _events.Writer.TryWrite( diff --git a/src/SecsFrame/Hsms/Transactions/HsmsIncomingDataMessage.cs b/src/SecsFrame/Hsms/Transactions/HsmsIncomingDataMessage.cs index 5aae46d..908d2de 100644 --- a/src/SecsFrame/Hsms/Transactions/HsmsIncomingDataMessage.cs +++ b/src/SecsFrame/Hsms/Transactions/HsmsIncomingDataMessage.cs @@ -1,13 +1,19 @@ namespace SecsFrame; -internal sealed class HsmsIncomingDataMessage +/// +/// An incoming HSMS data message bound to the transport session on which it +/// was received. +/// +public sealed class HsmsIncomingDataMessage { private int _replyStarted; - public HsmsIncomingDataMessage( + internal HsmsIncomingDataMessage( + object replyOwner, HsmsTransportSessionId transportSessionId, HsmsDataMessage dataMessage) { + ReplyOwner = replyOwner ?? throw new ArgumentNullException(nameof(replyOwner)); if (!transportSessionId.IsValid) { throw new ArgumentOutOfRangeException( @@ -20,10 +26,19 @@ public HsmsIncomingDataMessage( DataMessage = dataMessage ?? throw new ArgumentNullException(nameof(dataMessage)); } + /// Gets the decoded HSMS data message. public HsmsDataMessage DataMessage { get; } + /// Gets whether the message requests a secondary reply. + public bool ReplyExpected => DataMessage.Message.ReplyExpected; + + private object ReplyOwner { get; } + internal HsmsTransportSessionId TransportSessionId { get; } + internal bool IsOwnedBy(object owner) + => ReferenceEquals(ReplyOwner, owner); + internal bool TryBeginReply() => Interlocked.CompareExchange(ref _replyStarted, 1, 0) == 0; } diff --git a/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs new file mode 100644 index 0000000..5504926 --- /dev/null +++ b/test/SecsFrame.Tests/HsmsConnectionOptionsTests.cs @@ -0,0 +1,140 @@ +using System.Net; + +namespace SecsFrame.Tests; + +public sealed class HsmsConnectionOptionsTests +{ + [Fact] + public void Explicit_values_are_preserved() + { + var address = IPAddress.Loopback; + var options = new HsmsConnectionOptions( + address, + 5000, + HsmsConnectionMode.Active, + 10, + TimeSpan.FromSeconds(45), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5)); + + Assert.Same(address, options.IpAddress); + Assert.Equal(5000, options.Port); + Assert.Equal(HsmsConnectionMode.Active, options.ConnectionMode); + Assert.Equal(10, options.SessionId); + Assert.Equal(TimeSpan.FromSeconds(45), options.T3); + Assert.Equal(TimeSpan.FromSeconds(10), options.T5); + Assert.Equal(TimeSpan.FromSeconds(5), options.T6); + Assert.Equal(TimeSpan.FromSeconds(10), options.T7); + Assert.Equal(TimeSpan.FromSeconds(5), options.T8); + } + + [Fact] + public void Null_address_is_rejected() + { + Assert.Throws( + () => new HsmsConnectionOptions( + null!, + 5000, + HsmsConnectionMode.Active, + 10, + TimeSpan.FromSeconds(45), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(10), + TimeSpan.FromSeconds(5))); + } + + [Theory] + [InlineData(0)] + [InlineData(65536)] + public void Invalid_port_is_rejected(int port) + { + Assert.Throws( + () => Create(port: port)); + } + + [Fact] + public void Undefined_connection_mode_is_rejected() + { + Assert.Throws( + () => Create(connectionMode: (HsmsConnectionMode)99)); + } + + [Fact] + public void Control_message_session_identifier_is_rejected() + { + Assert.Throws( + () => Create(sessionId: ushort.MaxValue)); + } + + [Theory] + [InlineData("T3")] + [InlineData("T5")] + [InlineData("T6")] + [InlineData("T7")] + [InlineData("T8")] + public void Nonpositive_timer_is_rejected(string timerName) + { + Assert.Throws( + () => CreateWithTimer(timerName, TimeSpan.Zero)); + Assert.Throws( + () => CreateWithTimer(timerName, TimeSpan.FromTicks(-1))); + } + + [Fact] + public void Submillisecond_T5_is_rejected_instead_of_rounded() + { + Assert.Throws( + () => Create( + t5: TimeSpan.FromTicks( + TimeSpan.TicksPerMillisecond + 1))); + } + + [Fact] + public void T5_beyond_connection_retry_range_is_rejected() + { + Assert.Throws( + () => Create( + t5: TimeSpan.FromMilliseconds( + (long)int.MaxValue + 1))); + } + + private static HsmsConnectionOptions Create( + IPAddress? ipAddress = null, + int port = 5000, + HsmsConnectionMode connectionMode = HsmsConnectionMode.Active, + ushort sessionId = 10, + TimeSpan? t3 = null, + TimeSpan? t5 = null, + TimeSpan? t6 = null, + TimeSpan? t7 = null, + TimeSpan? t8 = null) + => new( + ipAddress ?? IPAddress.Loopback, + port, + connectionMode, + sessionId, + t3 ?? TimeSpan.FromSeconds(45), + t5 ?? TimeSpan.FromSeconds(10), + t6 ?? TimeSpan.FromSeconds(5), + t7 ?? TimeSpan.FromSeconds(10), + t8 ?? TimeSpan.FromSeconds(5)); + + private static HsmsConnectionOptions CreateWithTimer( + string timerName, + TimeSpan value) + => timerName switch + { + "T3" => Create(t3: value), + "T5" => Create(t5: value), + "T6" => Create(t6: value), + "T7" => Create(t7: value), + "T8" => Create(t8: value), + _ => throw new ArgumentOutOfRangeException( + nameof(timerName), + timerName, + "Unknown HSMS timer."), + }; +} diff --git a/test/SecsFrame.Tests/HsmsConnectionTests.cs b/test/SecsFrame.Tests/HsmsConnectionTests.cs new file mode 100644 index 0000000..655392c --- /dev/null +++ b/test/SecsFrame.Tests/HsmsConnectionTests.cs @@ -0,0 +1,282 @@ +using System.Net; + +namespace SecsFrame.Tests; + +public sealed class HsmsConnectionTests +{ + [Fact] + public async Task Public_connections_exchange_dynamic_messages_over_real_tcp() + { + var port = GetFreePort(); + await using var passive = new HsmsConnection( + CreateOptions(port, HsmsConnectionMode.Passive)); + await using var active = new HsmsConnection( + CreateOptions(port, HsmsConnectionMode.Active)); + using var cancellation = new CancellationTokenSource( + TimeSpan.FromSeconds(15)); + + passive.Start(); + active.Start(); + await using var passiveEvents = passive + .GetEventsAsync(cancellation.Token) + .GetAsyncEnumerator(); + await Task.WhenAll( + passive.WaitUntilSelectedAsync(cancellation.Token), + active.WaitUntilSelectedAsync(cancellation.Token)) + .ConfigureAwait(true); + + await AssertSelectedEventAsync( + passiveEvents, + passive, + active).ConfigureAwait(true); + await active.LinktestAsync(cancellation.Token).ConfigureAwait(true); + await AssertRoundTripAsync( + passive, + active, + passiveEvents, + cancellation.Token).ConfigureAwait(true); + } + + [Fact] + public async Task Canceling_open_transaction_does_not_close_public_connection() + { + var port = GetFreePort(); + await using var passive = new HsmsConnection( + CreateOptions(port, HsmsConnectionMode.Passive)); + await using var active = new HsmsConnection( + CreateOptions(port, HsmsConnectionMode.Active)); + using var lifetime = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + using var sendCancellation = new CancellationTokenSource(); + + passive.Start(); + active.Start(); + await using var passiveEvents = passive + .GetEventsAsync(lifetime.Token) + .GetAsyncEnumerator(); + await Task.WhenAll( + passive.WaitUntilSelectedAsync(lifetime.Token), + active.WaitUntilSelectedAsync(lifetime.Token)) + .ConfigureAwait(true); + var send = active.SendAsync( + new SecsMessage(1, 1, true), + sendCancellation.Token); + _ = await NextMatchingAsync( + passiveEvents, + item => item.Kind == + HsmsConnectionEventKind.DataMessageReceived) + .ConfigureAwait(true); + sendCancellation.Cancel(); + await Assert.ThrowsAnyAsync( + () => send).ConfigureAwait(true); + Assert.Equal(HsmsSessionState.Selected, passive.State); + Assert.Equal(HsmsSessionState.Selected, active.State); + } + + [Fact] + public async Task Readiness_wait_can_be_canceled_without_stopping_connection() + { + var port = GetFreePort(); + await using var connection = new HsmsConnection( + CreateOptions(port, HsmsConnectionMode.Active)); + using var cancellation = new CancellationTokenSource( + TimeSpan.FromMilliseconds(100)); + connection.Start(); + + await Assert.ThrowsAnyAsync( + () => connection.WaitUntilSelectedAsync(cancellation.Token)) + .ConfigureAwait(true); + + Assert.Equal(HsmsSessionState.Disconnected, connection.State); + } + + [Fact] + public async Task Disposing_connection_ends_uncanceled_readiness_wait() + { + var port = GetFreePort(); + var connection = new HsmsConnection( + CreateOptions(port, HsmsConnectionMode.Active)); + connection.Start(); + var selected = connection.WaitUntilSelectedAsync(); + + await connection.DisposeAsync().ConfigureAwait(true); + + await Assert.ThrowsAsync( + () => selected).ConfigureAwait(true); + } + + [Fact] + public async Task Canceled_event_read_allows_replacement_consumer() + { + var connection = new HsmsConnection( + CreateOptions(GetFreePort(), HsmsConnectionMode.Active)); + await using var owned = connection; + using var cancellation = new CancellationTokenSource( + TimeSpan.FromMilliseconds(100)); + connection.Start(); + await using var first = connection + .GetEventsAsync(cancellation.Token) + .GetAsyncEnumerator(); + + await Assert.ThrowsAnyAsync( + () => first.MoveNextAsync().AsTask()).ConfigureAwait(true); + + var replacement = connection.GetEventsAsync(); + await using var second = replacement + .GetAsyncEnumerator() + .ConfigureAwait(true); + } + + [Fact] + public async Task Concurrent_event_readers_are_rejected() + { + var connection = new HsmsConnection( + CreateOptions(GetFreePort(), HsmsConnectionMode.Active)); + await using var owned = connection; + using var cancellation = new CancellationTokenSource(); + connection.Start(); + await using var first = connection + .GetEventsAsync(cancellation.Token) + .GetAsyncEnumerator(); + var firstRead = first.MoveNextAsync().AsTask(); + await using var second = connection + .GetEventsAsync() + .GetAsyncEnumerator(); + + await Assert.ThrowsAsync( + () => second.MoveNextAsync().AsTask()).ConfigureAwait(true); + cancellation.Cancel(); + await Assert.ThrowsAnyAsync( + () => firstRead).ConfigureAwait(true); + } + + [Fact] + public async Task Connection_rejects_commands_before_start_and_duplicate_start() + { + var connection = new HsmsConnection( + CreateOptions(GetFreePort(), HsmsConnectionMode.Active)); + await using var owned = connection; + + Assert.Throws( + () => connection.GetEventsAsync()); + Assert.Throws( + () => + { + _ = connection.WaitUntilSelectedAsync(); + }); + Assert.Throws( + () => + { + _ = connection.SendAsync(new SecsMessage(1, 1)); + }); + + connection.Start(); + Assert.Throws(() => connection.Start()); + } + + private static HsmsConnectionOptions CreateOptions( + int port, + HsmsConnectionMode connectionMode) + => new( + IPAddress.Loopback, + port, + connectionMode, + sessionId: 10, + t3: TimeSpan.FromSeconds(5), + t5: TimeSpan.FromMilliseconds(10), + t6: TimeSpan.FromSeconds(5), + t7: TimeSpan.FromSeconds(10), + t8: TimeSpan.FromSeconds(5)); + + private static async Task AssertSelectedEventAsync( + IAsyncEnumerator passiveEvents, + HsmsConnection passive, + HsmsConnection active) + { + var selected = await NextMatchingAsync( + passiveEvents, + item => item.Kind == HsmsConnectionEventKind.StateChanged && + item.State == HsmsSessionState.Selected).ConfigureAwait(true); + + Assert.Null(selected.IncomingMessage); + Assert.Null(selected.Frame); + Assert.Null(selected.Error); + Assert.Equal(HsmsSessionState.Selected, passive.State); + Assert.Equal(HsmsSessionState.Selected, active.State); + } + + private static async Task AssertRoundTripAsync( + HsmsConnection passive, + HsmsConnection active, + IAsyncEnumerator passiveEvents, + CancellationToken cancellationToken) + { + var send = active.SendAsync( + new SecsMessage( + 6, + 11, + true, + SecsItem.List( + SecsItem.Ascii("LOT-42"), + SecsItem.List(SecsItem.U1(1), SecsItem.U1(2)))), + cancellationToken); + var received = await NextMatchingAsync( + passiveEvents, + item => item.Kind == + HsmsConnectionEventKind.DataMessageReceived).ConfigureAwait(true); + var incoming = received.IncomingMessage!; + + Assert.True(incoming.ReplyExpected); + Assert.Equal(10, incoming.DataMessage.SessionId); + Assert.Equal(6, incoming.DataMessage.Message.Stream); + Assert.Equal(11, incoming.DataMessage.Message.Function); + await Assert.ThrowsAsync( + () => active.ReplyAsync( + incoming, + new SecsMessage(6, 12), + cancellationToken)).ConfigureAwait(true); + await passive.ReplyAsync( + incoming, + new SecsMessage( + 6, + 12, + rootItem: SecsItem.Boolean(true)), + cancellationToken).ConfigureAwait(true); + var secondary = await send.ConfigureAwait(true); + + Assert.NotNull(secondary); + Assert.Equal(10, secondary.SessionId); + Assert.Equal(6, secondary.Message.Stream); + Assert.Equal(12, secondary.Message.Function); + Assert.Equal(SecsItem.Boolean(true), secondary.Message.RootItem); + await Assert.ThrowsAsync( + () => passive.ReplyAsync( + incoming, + new SecsMessage(6, 12), + cancellationToken)).ConfigureAwait(true); + } + + private static int GetFreePort() + { + var listener = new System.Net.Sockets.TcpListener( + IPAddress.Loopback, + 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static async Task NextMatchingAsync( + IAsyncEnumerator events, + Func predicate) + { + while (await events.MoveNextAsync().ConfigureAwait(true)) + { + if (predicate(events.Current)) + return events.Current; + } + + Assert.Fail("The expected public HSMS connection event was not received."); + return null!; + } +}