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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
## [Unreleased]

### 修复
- **终态连接等待([#65](https://github.com/CSJ608/StreamFrame/issues/65))**:取得连接等待器后复查停机状态,关闭首次注册与 Shutdown 的竞态;Dispose 或生命周期取消后的新等待也以任务取消结束,无需调用方令牌。补充终态调用、1000 次首次注册竞速及调用方取消隔离回归。
- **坏头后完整帧交付([#64](https://github.com/CSJ608/StreamFrame/issues/64))**:定界器返回 false 但已消费无效前缀时继续解析现有缓冲,无进展才等待输入;保留非法长度头四字节丢弃策略,避免完整帧挂起及误触发半帧超时。补充推进契约及连续坏头、粘包、半帧、EOF、自定义定界器和错误元数据回归。
- **监听绑定失败恢复([#63](https://github.com/CSJ608/StreamFrame/issues/63))**:监听 Socket 完成配置、Bind/Listen 后才发布,失败释放并保持字段为空,端口释放后可自动重试接入;发布与停机释放互斥,保留接受循环代次及单客户端监听关闭语义。新增同步信号门控的端口占用恢复、双向消息及取消/Dispose 竞态回归。
- **Soak 目标矩阵([#66](https://github.com/CSJ608/StreamFrame/issues/66))**:显式选择 Ubuntu net8.0/net10.0、Windows net8.0/net10.0/net48,避免 Ubuntu 启动 .NET Framework 测试;job 名、详细日志与始终尝试上传的 TRX 产物按 OS/TFM 区分,测试断言失败继续使 job 失败。普通 CI 必需检查名不变。
Expand Down
10 changes: 9 additions & 1 deletion src/StreamFrame/Connection/StreamConnection.cs
Original file line number Diff line number Diff line change
Expand Up @@ -1074,10 +1074,18 @@ public void Fault(Exception exception)
/// <inheritdoc />
public Task WaitForConnectedAsync(CancellationToken ct = default)
{
if (State == ConnectionState.Connected)
if (!IsDisposed && State == ConnectionState.Connected)
return Task.CompletedTask;

var tcs = GetOrCreateWhenConnected();
// 必须在注册后复查:Shutdown 可能已清空旧等待器,随后本次才发布新等待器。
// 若终态发生在复查之后,Shutdown 会取得已发布的等待器并负责取消。
if (IsDisposed)
{
tcs.TrySetCanceled();
return tcs.Task;
}

return State == ConnectionState.Connected
? Task.CompletedTask // 获取等待器的间隙恰好连上了
: tcs.Task.WaitAsync(ct);
Expand Down
71 changes: 70 additions & 1 deletion test/StreamFrame.Tests/ConnectionUsabilityTests.cs
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,12 @@ public async Task WaitForConnectedAsync_CompletesWhenConnectionEstablished()
await using var server = CreateConnection(port, isActive: false);

client.Start(CancellationToken.None);
var pending = client.WaitForConnectedAsync();
// 服务端稍后才启动:等待者必须跨过若干次连接失败重试
await Task.Delay(400);
server.Start(CancellationToken.None);

await client.WaitForConnectedAsync().WaitAsync(TimeSpan.FromSeconds(5));
await pending.WaitAsync(TimeSpan.FromSeconds(5));
await server.WaitForConnectedAsync().WaitAsync(TimeSpan.FromSeconds(5));
Assert.Equal(ConnectionState.Connected, client.State);
}
Expand Down Expand Up @@ -93,6 +94,74 @@ public async Task WaitForConnectedAsync_Dispose_CancelsPendingWaiter()
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => waitTask);
}

[Fact]
public async Task WaitForConnectedAsync_AfterDispose_IsCanceled()
{
await using var client = CreateConnection(12345, isActive: true);
await client.DisposeAsync();

var wait = client.WaitForConnectedAsync();
Assert.True(wait.IsCanceled);
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => wait);
}

[Fact]
public async Task WaitForConnectedAsync_FirstRegistrationRacesDispose_NoOrphanedWaiter()
{
for (var iteration = 0; iteration < 1000; iteration++)
{
await using var client = CreateConnection(12345, isActive: true);
using var start = new Barrier(2);
Task? wait = null;
var register = Task.Run(() =>
{
Assert.True(start.SignalAndWait(TimeSpan.FromSeconds(5)));
wait = client.WaitForConnectedAsync();
});
var shutdown = Task.Run(async () =>
{
Assert.True(start.SignalAndWait(TimeSpan.FromSeconds(5)));
await client.DisposeAsync();
});
await Task.WhenAll(register, shutdown).WaitAsync(TimeSpan.FromSeconds(10));
Assert.NotNull(wait);
// net48 的 WaitAsync 兼容包装通过异步延续传播取消,必须等待传播完成。
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => wait.WaitAsync(TimeSpan.FromSeconds(5)));
Assert.True(wait.IsCanceled);
}
}

[Fact]
public async Task WaitForConnectedAsync_CallerCancellation_DoesNotCancelSharedWaiter()
{
await using var client = CreateConnection(12345, isActive: true);
using var caller = new CancellationTokenSource();
var shared = client.WaitForConnectedAsync();
var canceled = client.WaitForConnectedAsync(caller.Token);
caller.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(() => canceled);
Assert.False(shared.IsCompleted);
await client.DisposeAsync();
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => shared.WaitAsync(TimeSpan.FromSeconds(5)));
Assert.True(shared.IsCanceled);
}

[Fact]
public async Task WaitForConnectedAsync_LifetimeCancellation_CancelsExistingAndFutureWaiters()
{
await using var client = CreateConnection(GetFreePort(), isActive: true);
using var lifetime = new CancellationTokenSource();
client.Start(lifetime.Token);
var wait = client.WaitForConnectedAsync();
lifetime.Cancel();
await Assert.ThrowsAnyAsync<OperationCanceledException>(
() => wait.WaitAsync(TimeSpan.FromSeconds(5)));
Assert.True(wait.IsCanceled);
Assert.True(client.WaitForConnectedAsync().IsCanceled);
}

[Fact]
public async Task Start_SecondCall_ThrowsInvalidOperation()
{
Expand Down
Loading