diff --git a/FeatBit.Agent.sln b/FeatBit.Agent.sln deleted file mode 100644 index bfb7bff..0000000 --- a/FeatBit.Agent.sln +++ /dev/null @@ -1,27 +0,0 @@ - -Microsoft Visual Studio Solution File, Format Version 12.00 -# Visual Studio Version 17 -VisualStudioVersion = 17.0.31903.59 -MinimumVisualStudioVersion = 10.0.40219.1 -Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{7E91AF53-D0F4-4DCE-8E64-1F6C5E03559B}" -EndProject -Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Api", "src\Api\Api.csproj", "{D6BC46AB-0B65-4066-88AB-DE75FB3AC597}" -EndProject -Global - GlobalSection(SolutionConfigurationPlatforms) = preSolution - Debug|Any CPU = Debug|Any CPU - Release|Any CPU = Release|Any CPU - EndGlobalSection - GlobalSection(SolutionProperties) = preSolution - HideSolutionNode = FALSE - EndGlobalSection - GlobalSection(ProjectConfigurationPlatforms) = postSolution - {D6BC46AB-0B65-4066-88AB-DE75FB3AC597}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {D6BC46AB-0B65-4066-88AB-DE75FB3AC597}.Debug|Any CPU.Build.0 = Debug|Any CPU - {D6BC46AB-0B65-4066-88AB-DE75FB3AC597}.Release|Any CPU.ActiveCfg = Release|Any CPU - {D6BC46AB-0B65-4066-88AB-DE75FB3AC597}.Release|Any CPU.Build.0 = Release|Any CPU - EndGlobalSection - GlobalSection(NestedProjects) = preSolution - {D6BC46AB-0B65-4066-88AB-DE75FB3AC597} = {7E91AF53-D0F4-4DCE-8E64-1F6C5E03559B} - EndGlobalSection -EndGlobal diff --git a/FeatBit.Agent.slnx b/FeatBit.Agent.slnx new file mode 100644 index 0000000..88918a9 --- /dev/null +++ b/FeatBit.Agent.slnx @@ -0,0 +1,9 @@ + + + + + + + + + diff --git a/src/Api/Properties/AssemblyInfo.cs b/src/Api/Properties/AssemblyInfo.cs new file mode 100644 index 0000000..c2f70bc --- /dev/null +++ b/src/Api/Properties/AssemblyInfo.cs @@ -0,0 +1,4 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Api.UnitTests")] +[assembly: InternalsVisibleTo("Api.IntegrationTests")] diff --git a/src/Api/Transport/FbWebSocket.cs b/src/Api/Transport/FbWebSocket.cs index abf4d5c..8cf7dae 100644 --- a/src/Api/Transport/FbWebSocket.cs +++ b/src/Api/Transport/FbWebSocket.cs @@ -5,6 +5,8 @@ namespace Api.Transport; internal sealed partial class FbWebSocket { + internal const WebSocketCloseStatus ServerRejectedCloseStatus = (WebSocketCloseStatus)4003; + public event Func? OnConnected; public event Func? OnKeepAlive; public event Func? OnReconnecting; @@ -42,12 +44,15 @@ internal FbWebSocket(Uri websocketUri, ILoggerFactory loggerFactory) public async Task ConnectAsync(bool isReconnecting = false) { _websocket = new ClientWebSocket(); - using var connectCts = new CancellationTokenSource(_connectTimeout); + + using var connectCts = CancellationTokenSource.CreateLinkedTokenSource(_stopCts.Token); + connectCts.CancelAfter(_connectTimeout); try { Log.Starting(_logger, _websocketUri); await _websocket.ConnectAsync(_websocketUri, connectCts.Token); + _stopCts.Token.ThrowIfCancellationRequested(); Log.Connected(_logger); @@ -61,12 +66,14 @@ public async Task ConnectAsync(bool isReconnecting = false) { Log.ErrorStarting(_logger, ex); - if (ex is TaskCanceledException tce && tce.CancellationToken == connectCts.Token) + if (ex is OperationCanceledException && !_stopCts.IsCancellationRequested) { Log.TimedOut(_logger, _connectTimeout.TotalSeconds); } - if (!isReconnecting) + _websocket.Dispose(); + + if (!isReconnecting && !_stopCts.IsCancellationRequested) { _ = ReconnectAsync(); } @@ -102,7 +109,6 @@ public async Task CloseAsync(CancellationToken cancellationToken) } _stopCts.Cancel(); - _stopCts.Dispose(); _keepAliveTimer?.Dispose(); @@ -159,10 +165,24 @@ private void HandleConnectionClose() } } - private bool ShouldReconnect() + private bool ShouldReconnect() => ShouldReconnect(_websocket?.CloseStatus, _stopCts.IsCancellationRequested); + + internal static bool ShouldReconnect(WebSocketCloseStatus? closeStatus, bool stopRequested) { - var closeStatus = _websocket?.CloseStatus; - return closeStatus != WebSocketCloseStatus.NormalClosure && closeStatus != (WebSocketCloseStatus)4003; + if (closeStatus == ServerRejectedCloseStatus) + { + // server rejected the connection explicitly, so we should not reconnect + return false; + } + + if (stopRequested) + { + // connection was stopped manually, so we should not reconnect + return false; + } + + // otherwise, we should reconnect + return true; } private async Task ReconnectAsync() @@ -274,4 +294,4 @@ private async Task KeepAliveAsync(CancellationToken ct = default) Log.InvokingEventHandler(_logger, nameof(OnKeepAlive)); _ = OnKeepAlive?.Invoke(); } -} \ No newline at end of file +} diff --git a/tests/Api.IntegrationTests/Api.IntegrationTests.csproj b/tests/Api.IntegrationTests/Api.IntegrationTests.csproj new file mode 100644 index 0000000..c4f1b9f --- /dev/null +++ b/tests/Api.IntegrationTests/Api.IntegrationTests.csproj @@ -0,0 +1,25 @@ + + + + net10.0 + enable + enable + false + true + + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + + + + diff --git a/tests/Api.IntegrationTests/Transport/WebSocketReconnectTests.cs b/tests/Api.IntegrationTests/Transport/WebSocketReconnectTests.cs new file mode 100644 index 0000000..ec97139 --- /dev/null +++ b/tests/Api.IntegrationTests/Transport/WebSocketReconnectTests.cs @@ -0,0 +1,100 @@ +using System.Net.WebSockets; +using Api.Transport; +using Microsoft.Extensions.Logging.Abstractions; +using Xunit; + +namespace Api.IntegrationTests.Transport; + +public class WebSocketReconnectTests +{ + [Fact] + public async Task NormalClosure_ReconnectsRepeatedly() + { + await using var server = await WebSocketStubServer.StartAsync(); + var webSocket = new FbWebSocket(server.WebSocketUri, NullLoggerFactory.Instance); + + try + { + await webSocket.ConnectAsync(); + await server.WaitForConnectionsAsync(1); + + await server.CloseAsync(1, WebSocketCloseStatus.NormalClosure); + await server.WaitForConnectionsAsync(2); + + await server.CloseAsync(2, WebSocketCloseStatus.NormalClosure); + await server.WaitForConnectionsAsync(3); + + Assert.Equal(3, server.ConnectionCount); + } + finally + { + server.AbortConnections(); + await webSocket.CloseAsync(CancellationToken.None); + } + } + + [Fact] + public async Task InvalidRequestClose_DoesNotReconnect() + { + await using var server = await WebSocketStubServer.StartAsync( + FbWebSocket.ServerRejectedCloseStatus + ); + var webSocket = new FbWebSocket(server.WebSocketUri, NullLoggerFactory.Instance); + var closed = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously + ); + webSocket.OnClosed += (_, status, _) => + { + closed.TrySetResult(status); + return Task.CompletedTask; + }; + + try + { + await webSocket.ConnectAsync(); + + var closeStatus = await closed.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(TimeSpan.FromMilliseconds(250)); + + Assert.Equal(FbWebSocket.ServerRejectedCloseStatus, closeStatus); + Assert.Equal(1, server.ConnectionCount); + } + finally + { + server.AbortConnections(); + await webSocket.CloseAsync(CancellationToken.None); + } + } + + [Fact] + public async Task NormalClosure_WhenStoppedDuringReconnect_DoesNotReconnect() + { + await using var server = await WebSocketStubServer.StartAsync(); + var webSocket = new FbWebSocket(server.WebSocketUri, NullLoggerFactory.Instance); + var reconnecting = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously + ); + webSocket.OnReconnecting += _ => + { + reconnecting.TrySetResult(); + return webSocket.CloseAsync(CancellationToken.None); + }; + + try + { + await webSocket.ConnectAsync(); + await server.WaitForConnectionsAsync(1); + + await server.CloseAsync(1, WebSocketCloseStatus.NormalClosure); + await reconnecting.Task.WaitAsync(TimeSpan.FromSeconds(5)); + await Task.Delay(TimeSpan.FromMilliseconds(250)); + + Assert.Equal(1, server.ConnectionCount); + } + finally + { + server.AbortConnections(); + await webSocket.CloseAsync(CancellationToken.None); + } + } +} diff --git a/tests/Api.IntegrationTests/Transport/WebSocketStubServer.cs b/tests/Api.IntegrationTests/Transport/WebSocketStubServer.cs new file mode 100644 index 0000000..750f85d --- /dev/null +++ b/tests/Api.IntegrationTests/Transport/WebSocketStubServer.cs @@ -0,0 +1,142 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.WebSockets; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.Hosting.Server.Features; +using Microsoft.AspNetCore.Http; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; + +namespace Api.IntegrationTests.Transport; + +public sealed class WebSocketStubServer : IAsyncDisposable +{ + private readonly WebApplication _application; + private readonly WebSocketCloseStatus? _closeImmediatelyWith; + private readonly ConcurrentDictionary _connections = new(); + private readonly SemaphoreSlim _connectionSignal = new(0); + private int _connectionCount; + + private WebSocketStubServer(WebApplication application, WebSocketCloseStatus? closeImmediatelyWith) + { + _application = application; + _closeImmediatelyWith = closeImmediatelyWith; + } + + public int ConnectionCount => Volatile.Read(ref _connectionCount); + + public Uri WebSocketUri { get; private set; } = null!; + + public static async Task StartAsync(WebSocketCloseStatus? closeImmediatelyWith = null) + { + var builder = WebApplication.CreateBuilder(); + builder.Logging.ClearProviders(); + builder.WebHost.ConfigureKestrel(options => options.Listen(IPAddress.Loopback, 0)); + + var application = builder.Build(); + var server = new WebSocketStubServer(application, closeImmediatelyWith); + + application.UseWebSockets(); + application.Map("/streaming", server.HandleConnectionAsync); + + await application.StartAsync(); + + var addresses = application.Services + .GetRequiredService() + .Features + .Get()!; + var httpUri = new Uri(addresses.Addresses.Single()); + server.WebSocketUri = new UriBuilder(new Uri(httpUri, "/streaming")) { Scheme = "ws" }.Uri; + + return server; + } + + public async Task WaitForConnectionsAsync(int expectedCount) + { + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + while (ConnectionCount < expectedCount) + { + await _connectionSignal.WaitAsync(timeout.Token); + } + } + + public async Task CloseAsync(int connectionId, WebSocketCloseStatus closeStatus) + { + if (!_connections.TryGetValue(connectionId, out var connection)) + { + throw new InvalidOperationException( + $"WebSocket connection {connectionId} is no longer active." + ); + } + + await connection.CloseOutputAsync(closeStatus, "closed by test server", CancellationToken.None); + } + + public void AbortConnections() + { + foreach (var connection in _connections.Values) + { + connection.Abort(); + } + } + + private async Task HandleConnectionAsync(HttpContext context) + { + if (!context.WebSockets.IsWebSocketRequest) + { + context.Response.StatusCode = StatusCodes.Status400BadRequest; + return; + } + + using var webSocket = await context.WebSockets.AcceptWebSocketAsync(); + var connectionId = Interlocked.Increment(ref _connectionCount); + _connections[connectionId] = webSocket; + _connectionSignal.Release(); + + try + { + if (_closeImmediatelyWith.HasValue) + { + await webSocket.CloseOutputAsync( + _closeImmediatelyWith.Value, + "closed by test server", + CancellationToken.None + ); + } + + var buffer = new byte[1024]; + while (webSocket.State is WebSocketState.Open or WebSocketState.CloseSent) + { + var result = await webSocket.ReceiveAsync(buffer, context.RequestAborted); + if (result.MessageType == WebSocketMessageType.Close) + { + break; + } + } + } + catch (OperationCanceledException) + { + // The test server is stopping. + } + catch (WebSocketException) + { + // The client disposed the connection after observing the close frame. + } + finally + { + _connections.TryRemove(connectionId, out _); + } + } + + public async ValueTask DisposeAsync() + { + AbortConnections(); + + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(5)); + await _application.StopAsync(timeout.Token); + await _application.DisposeAsync(); + _connectionSignal.Dispose(); + } +} diff --git a/tests/Api.UnitTests/Api.UnitTests.csproj b/tests/Api.UnitTests/Api.UnitTests.csproj new file mode 100644 index 0000000..6d940cf --- /dev/null +++ b/tests/Api.UnitTests/Api.UnitTests.csproj @@ -0,0 +1,24 @@ + + + + net10.0 + enable + enable + false + true + + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + + + diff --git a/tests/Api.UnitTests/Transport/ReconnectTests.cs b/tests/Api.UnitTests/Transport/ReconnectTests.cs new file mode 100644 index 0000000..e89a788 --- /dev/null +++ b/tests/Api.UnitTests/Transport/ReconnectTests.cs @@ -0,0 +1,42 @@ +using System.Net.WebSockets; +using Api.Transport; +using Xunit; + +namespace Api.UnitTests.Transport; + +public class FbWebSocketTests +{ + [Theory] + [InlineData(null)] + [InlineData(WebSocketCloseStatus.NormalClosure)] + [InlineData(WebSocketCloseStatus.EndpointUnavailable)] + [InlineData(WebSocketCloseStatus.InternalServerError)] + public void ShouldReconnect_WhenConnectionWasNotStopped_ReturnsTrue(WebSocketCloseStatus? closeStatus) + { + var shouldReconnect = FbWebSocket.ShouldReconnect(closeStatus, stopRequested: false); + + Assert.True(shouldReconnect); + } + + [Fact] + public void ShouldReconnect_WhenConnectionWasStopped_ReturnsFalse() + { + var shouldReconnect = FbWebSocket.ShouldReconnect( + WebSocketCloseStatus.NormalClosure, + stopRequested: true + ); + + Assert.False(shouldReconnect); + } + + [Fact] + public void ShouldReconnect_WhenRequestIsInvalid_ReturnsFalse() + { + var shouldReconnect = FbWebSocket.ShouldReconnect( + FbWebSocket.ServerRejectedCloseStatus, + stopRequested: false + ); + + Assert.False(shouldReconnect); + } +}