diff --git a/samples/ClientSample/HubSample.cs b/samples/ClientSample/HubSample.cs index 5f5697915d..397cb8f433 100644 --- a/samples/ClientSample/HubSample.cs +++ b/samples/ClientSample/HubSample.cs @@ -36,31 +36,23 @@ namespace ClientSample try { + var sendCts = new CancellationTokenSource(); - var cts = new CancellationTokenSource(); - Console.CancelKeyPress += (sender, a) => + Console.CancelKeyPress += async (sender, a) => { a.Cancel = true; Console.WriteLine("Stopping loops..."); - cts.Cancel(); + sendCts.Cancel(); + await connection.DisposeAsync(); }; // Set up handler connection.On("Send", Console.WriteLine); - connection.Closed += e => + while (!connection.Closed.IsCompleted) { - Console.WriteLine("Connection closed."); - cts.Cancel(); - return Task.CompletedTask; - }; - - var ctsTask = Task.Delay(-1, cts.Token); - - while (!cts.Token.IsCancellationRequested) - { - var completedTask = await Task.WhenAny(Task.Run(() => Console.ReadLine(), cts.Token), ctsTask); - if (completedTask == ctsTask) + var completedTask = await Task.WhenAny(Task.Run(() => Console.ReadLine()), connection.Closed); + if (completedTask == connection.Closed) { break; } @@ -72,7 +64,7 @@ namespace ClientSample break; } - await connection.InvokeAsync("Send", line, cts.Token); + await connection.InvokeAsync("Send", line, sendCts.Token); } } catch (AggregateException aex) when (aex.InnerExceptions.All(e => e is OperationCanceledException)) diff --git a/samples/ClientSample/RawSample.cs b/samples/ClientSample/RawSample.cs index 19ba4a07a6..35786286dd 100644 --- a/samples/ClientSample/RawSample.cs +++ b/samples/ClientSample/RawSample.cs @@ -39,25 +39,19 @@ namespace ClientSample var connection = new HttpConnection(new Uri(baseUrl), loggerFactory); try { - var cts = new CancellationTokenSource(); connection.OnReceived(data => Console.Out.WriteLineAsync($"{Encoding.UTF8.GetString(data)}")); - connection.Closed += e => - { - cts.Cancel(); - return Task.CompletedTask; - }; await connection.StartAsync(); Console.WriteLine($"Connected to {baseUrl}"); - - Console.CancelKeyPress += (sender, a) => + var cts = new CancellationTokenSource(); + Console.CancelKeyPress += async (sender, a) => { a.Cancel = true; - cts.Cancel(); + await connection.DisposeAsync(); }; - while (!cts.Token.IsCancellationRequested) + while (!connection.Closed.IsCompleted) { var line = await Task.Run(() => Console.ReadLine(), cts.Token); diff --git a/src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.cs b/src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.cs index 8a85c41570..443cf294e0 100644 --- a/src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.cs +++ b/src/Microsoft.AspNetCore.SignalR.Client.Core/HubConnection.cs @@ -39,11 +39,7 @@ namespace Microsoft.AspNetCore.SignalR.Client private int _nextId = 0; private volatile bool _startCalled; - public event Func Closed - { - add { _connection.Closed += value; } - remove { _connection.Closed -= value; } - } + public Task Closed { get; } public HubConnection(IConnection connection, IHubProtocol protocol, ILoggerFactory loggerFactory) { @@ -63,7 +59,11 @@ namespace Microsoft.AspNetCore.SignalR.Client _loggerFactory = loggerFactory ?? NullLoggerFactory.Instance; _logger = _loggerFactory.CreateLogger(); _connection.OnReceived((data, state) => ((HubConnection)state).OnDataReceivedAsync(data), this); - _connection.Closed += Shutdown; + Closed = _connection.Closed.ContinueWith(task => + { + Shutdown(task.Exception); + return task; + }).Unwrap(); } public async Task StartAsync() diff --git a/src/Microsoft.AspNetCore.Sockets.Abstractions/IConnection.cs b/src/Microsoft.AspNetCore.Sockets.Abstractions/IConnection.cs index 5efbd777b3..be3f5f26fe 100644 --- a/src/Microsoft.AspNetCore.Sockets.Abstractions/IConnection.cs +++ b/src/Microsoft.AspNetCore.Sockets.Abstractions/IConnection.cs @@ -16,7 +16,7 @@ namespace Microsoft.AspNetCore.Sockets.Client IDisposable OnReceived(Func callback, object state); - event Func Closed; + Task Closed { get; } IFeatureCollection Features { get; } } diff --git a/src/Microsoft.AspNetCore.Sockets.Client.Http/HttpConnection.cs b/src/Microsoft.AspNetCore.Sockets.Client.Http/HttpConnection.cs index ab4eafb255..c2c27f5fe2 100644 --- a/src/Microsoft.AspNetCore.Sockets.Client.Http/HttpConnection.cs +++ b/src/Microsoft.AspNetCore.Sockets.Client.Http/HttpConnection.cs @@ -34,6 +34,7 @@ namespace Microsoft.AspNetCore.Sockets.Client private volatile ITransport _transport; private volatile Task _receiveLoopTask; private TaskCompletionSource _startTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly TaskCompletionSource _closedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); private TaskQueue _eventQueue = new TaskQueue(); private readonly ITransportFactory _transportFactory; private string _connectionId; @@ -47,7 +48,7 @@ namespace Microsoft.AspNetCore.Sockets.Client public IFeatureCollection Features { get; } = new FeatureCollection(); - public event Func Closed; + public Task Closed => _closedTcs.Task; public HttpConnection(Uri url) : this(url, TransportType.All) @@ -116,10 +117,12 @@ namespace Microsoft.AspNetCore.Sockets.Client if (t.IsFaulted) { _startTcs.SetException(t.Exception.InnerException); + _closedTcs.TrySetException(t.Exception.InnerException); } else if (t.IsCanceled) { _startTcs.SetCanceled(); + _closedTcs.SetCanceled(); } else { @@ -190,19 +193,18 @@ namespace Microsoft.AspNetCore.Sockets.Client await Task.WhenAny(_eventQueue.Drain().NoThrow(), Task.Delay(_eventQueueDrainTimeout)); _httpClient?.Dispose(); - _logger.RaiseClosed(_connectionId); - - var closedEventHandler = Closed; - if (closedEventHandler != null) + _logger.CompleteClosed(_connectionId); + if (t.IsFaulted) { - try - { - await closedEventHandler.Invoke(t.IsFaulted ? t.Exception.InnerException : null); - } - catch (Exception ex) - { - _logger.ExceptionThrownFromCallback(_connectionId, nameof(Closed), ex); - } + _closedTcs.TrySetException(t.Exception.InnerException); + } + if (t.IsCanceled) + { + _closedTcs.TrySetCanceled(); + } + else + { + _closedTcs.TrySetResult(null); } }); @@ -463,6 +465,7 @@ namespace Microsoft.AspNetCore.Sockets.Client await _receiveLoopTask; } + _closedTcs.TrySetResult(null); _httpClient?.Dispose(); } diff --git a/src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/SocketClientLoggerExtensions.cs b/src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/SocketClientLoggerExtensions.cs index 1ba8b1e82b..1ac5a0a0d4 100644 --- a/src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/SocketClientLoggerExtensions.cs +++ b/src/Microsoft.AspNetCore.Sockets.Client.Http/Internal/SocketClientLoggerExtensions.cs @@ -111,8 +111,8 @@ namespace Microsoft.AspNetCore.Sockets.Client.Internal private static readonly Action _drainEvents = LoggerMessage.Define(LogLevel.Debug, new EventId(5, nameof(DrainEvents)), "{time}: Connection Id {connectionId}: Draining event queue."); - private static readonly Action _raiseClosed = - LoggerMessage.Define(LogLevel.Debug, new EventId(6, nameof(RaiseClosed)), "{time}: Connection Id {connectionId}: Raising Closed event."); + private static readonly Action _completeClosed = + LoggerMessage.Define(LogLevel.Debug, new EventId(6, nameof(CompleteClosed)), "{time}: Connection Id {connectionId}: Completing Closed task."); private static readonly Action _establishingConnection = LoggerMessage.Define(LogLevel.Debug, new EventId(7, nameof(EstablishingConnection)), "{time}: Establishing Connection at: {url}."); @@ -410,11 +410,11 @@ namespace Microsoft.AspNetCore.Sockets.Client.Internal } } - public static void RaiseClosed(this ILogger logger, string connectionId) + public static void CompleteClosed(this ILogger logger, string connectionId) { if (logger.IsEnabled(LogLevel.Debug)) { - _raiseClosed(logger, DateTime.Now, connectionId, null); + _completeClosed(logger, DateTime.Now, connectionId, null); } } diff --git a/test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs b/test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs index 42c8380be1..902ed2de67 100644 --- a/test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs +++ b/test/Microsoft.AspNetCore.SignalR.Client.FunctionalTests/HubConnectionTests.cs @@ -174,22 +174,9 @@ namespace Microsoft.AspNetCore.SignalR.Client.FunctionalTests try { await connection.StartAsync().OrTimeout(); - var closeTcs = new TaskCompletionSource(); - connection.Closed += ex => - { - if (ex != null) - { - closeTcs.SetException(ex); - } - else - { - closeTcs.SetResult(null); - } - return Task.CompletedTask; - }; await connection.InvokeAsync("CallHandlerThatDoesntExist").OrTimeout(); await connection.DisposeAsync().OrTimeout(); - await closeTcs.Task.OrTimeout(); + await connection.Closed.OrTimeout(); } finally { diff --git a/test/Microsoft.AspNetCore.SignalR.Client.Tests/HttpConnectionTests.cs b/test/Microsoft.AspNetCore.SignalR.Client.Tests/HttpConnectionTests.cs index 3baddd6175..58a20ef28b 100644 --- a/test/Microsoft.AspNetCore.SignalR.Client.Tests/HttpConnectionTests.cs +++ b/test/Microsoft.AspNetCore.SignalR.Client.Tests/HttpConnectionTests.cs @@ -204,18 +204,11 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests var connection = new HttpConnection(new Uri("http://fakeuri.org/"), TransportType.LongPolling, loggerFactory: null, httpMessageHandler: mockHttpHandler.Object); - var closedEventTcs = new TaskCompletionSource(); - connection.Closed += e => - { - closedEventTcs.SetResult(e); - return Task.CompletedTask; - }; - - await connection.StartAsync(); - await connection.DisposeAsync(); + await connection.StartAsync().OrTimeout(); + await connection.DisposeAsync().OrTimeout(); + await connection.Closed.OrTimeout(); // in case of clean disconnect error should be null - Assert.Null(await closedEventTcs.Task.OrTimeout()); } [Fact] @@ -236,17 +229,11 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests }); var connection = new HttpConnection(new Uri("http://fakeuri.org/"), TransportType.LongPolling, loggerFactory: null, httpMessageHandler: mockHttpHandler.Object); - var closedEventTcs = new TaskCompletionSource(); - connection.Closed += e => - { - closedEventTcs.TrySetResult(e); - return Task.CompletedTask; - }; try { - await connection.StartAsync(); - Assert.IsType(await closedEventTcs.Task.OrTimeout()); + await connection.StartAsync().OrTimeout(); + await Assert.ThrowsAsync(() => connection.Closed.OrTimeout()); } finally { @@ -388,16 +375,10 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests mockTransport.SetupGet(t => t.Mode).Returns(TransferMode.Text); var blockReceiveCallbackTcs = new TaskCompletionSource(); - var closedTcs = new TaskCompletionSource(); var connection = new HttpConnection(new Uri("http://fakeuri.org/"), new TestTransportFactory(mockTransport.Object), loggerFactory: null, httpMessageHandler: mockHttpHandler.Object); connection.OnReceived(_ => blockReceiveCallbackTcs.Task); - connection.Closed += _ => { - closedTcs.SetResult(null); - return Task.CompletedTask; - }; - await connection.StartAsync(); channel.Out.TryWrite(Array.Empty()); @@ -438,7 +419,6 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests mockTransport.SetupGet(t => t.Mode).Returns(TransferMode.Text); var callbackInvokedTcs = new TaskCompletionSource(); - var closedTcs = new TaskCompletionSource(); var connection = new HttpConnection(new Uri("http://fakeuri.org/"), new TestTransportFactory(mockTransport.Object), loggerFactory: null, httpMessageHandler: mockHttpHandler.Object); connection.OnReceived( _ => @@ -460,15 +440,8 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests { var connection = new HttpConnection(new Uri("http://fakeuri.org/")); - bool closedEventRaised = false; - connection.Closed += e => - { - closedEventRaised = true; - return Task.CompletedTask; - }; - await connection.DisposeAsync(); - Assert.False(closedEventRaised); + Assert.False(connection.Closed.IsCompleted); } [Fact] @@ -644,21 +617,20 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests return Task.CompletedTask; }, receiveTcs); - connection.Closed += e => + _ = connection.Closed.ContinueWith(task => + { + if (task.Exception != null) { - if (e != null) - { - receiveTcs.TrySetException(e); - } - else - { - receiveTcs.TrySetCanceled(); - } - return Task.CompletedTask; - }; - - await connection.StartAsync(); + receiveTcs.TrySetException(task.Exception); + } + else + { + receiveTcs.TrySetCanceled(); + } + return Task.CompletedTask; + }); + await connection.StartAsync().OrTimeout(); Assert.Equal("42", await receiveTcs.Task.OrTimeout()); } finally @@ -707,18 +679,18 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests return Task.CompletedTask; }); - connection.Closed += e => + _ = connection.Closed.ContinueWith(task => { - if (e != null) + if (task.Exception != null) { - receiveTcs.TrySetException(e); + receiveTcs.TrySetException(task.Exception); } else { receiveTcs.TrySetCanceled(); } return Task.CompletedTask; - }; + }); await connection.StartAsync(); @@ -770,18 +742,18 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests return Task.CompletedTask; }); - connection.Closed += e => + _ = connection.Closed.ContinueWith(task => { - if (e != null) + if (task.Exception != null) { - receiveTcs.TrySetException(e); + receiveTcs.TrySetException(task.Exception); } else { receiveTcs.TrySetCanceled(); } return Task.CompletedTask; - }; + }); await connection.StartAsync(); @@ -813,20 +785,13 @@ namespace Microsoft.AspNetCore.Sockets.Client.Tests var connection = new HttpConnection(new Uri("http://fakeuri.org/"), TransportType.LongPolling, loggerFactory: null, httpMessageHandler: mockHttpHandler.Object); try { - var closeTcs = new TaskCompletionSource(); - connection.Closed += e => - { - closeTcs.TrySetResult(e); - return Task.CompletedTask; - }; - await connection.StartAsync(); + await connection.StartAsync().OrTimeout(); // Exception in send should shutdown the connection - await closeTcs.Task.OrTimeout(); + await Assert.ThrowsAsync(() => connection.Closed.OrTimeout()); - var exception = await Assert.ThrowsAsync( - async () => await connection.SendAsync(new byte[0])); + var exception = await Assert.ThrowsAsync(() => connection.SendAsync(new byte[0])); Assert.Equal("Cannot send messages when the connection is not in the Connected state.", exception.Message); } diff --git a/test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionTests.cs b/test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionTests.cs index a744073767..1114955fc0 100644 --- a/test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionTests.cs +++ b/test/Microsoft.AspNetCore.SignalR.Client.Tests/HubConnectionTests.cs @@ -72,16 +72,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests { var hubConnection = new HubConnection(new TestConnection(), Mock.Of(), null); var closedEventTcs = new TaskCompletionSource(); - hubConnection.Closed += e => - { - closedEventTcs.SetResult(e); - return Task.CompletedTask; - }; - await hubConnection.StartAsync(); - await hubConnection.DisposeAsync(); - - Assert.Null(await closedEventTcs.Task.OrTimeout()); + await hubConnection.StartAsync().OrTimeout(); + await hubConnection.DisposeAsync().OrTimeout(); + await hubConnection.Closed.OrTimeout(); } [Fact] @@ -177,12 +171,10 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests [Fact] public async Task PendingInvocationsAreTerminatedWithExceptionWhenConnectionClosesDueToError() { - var exception = new InvalidOperationException(); var mockConnection = new Mock(); mockConnection.SetupGet(p => p.Features).Returns(new FeatureCollection()); mockConnection .Setup(m => m.DisposeAsync()) - .Callback(() => mockConnection.Raise(c => c.Closed += null, exception)) .Returns(Task.FromResult(null)); var hubConnection = new HubConnection(mockConnection.Object, Mock.Of(), new LoggerFactory()); @@ -191,8 +183,7 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests var invokeTask = hubConnection.InvokeAsync("testMethod"); await hubConnection.DisposeAsync(); - var thrown = await Assert.ThrowsAsync(exception.GetType(), async () => await invokeTask); - Assert.Same(exception, thrown); + await Assert.ThrowsAsync(async () => await invokeTask); } // Moq really doesn't handle out parameters well, so to make these tests work I added a manual mock -anurse diff --git a/test/Microsoft.AspNetCore.SignalR.Client.Tests/TestConnection.cs b/test/Microsoft.AspNetCore.SignalR.Client.Tests/TestConnection.cs index 7221162cf0..d234931fd8 100644 --- a/test/Microsoft.AspNetCore.SignalR.Client.Tests/TestConnection.cs +++ b/test/Microsoft.AspNetCore.SignalR.Client.Tests/TestConnection.cs @@ -29,9 +29,9 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests private Task _receiveLoop; private TransferMode? _transferMode; + private readonly TaskCompletionSource _closeTcs = new TaskCompletionSource(); - public event Func Closed; - + public Task Closed => _closeTcs.Task; public Task Started => _started.Task; public Task Disposed => _disposed.Task; public ReadableChannel SentMessages => _sentMessages.In; @@ -133,16 +133,16 @@ namespace Microsoft.AspNetCore.SignalR.Client.Tests } } } - Closed?.Invoke(null); + _closeTcs.TrySetResult(null); } catch (OperationCanceledException) { // Do nothing, we were just asked to shut down. - Closed?.Invoke(null); + _closeTcs.TrySetResult(null); } catch (Exception ex) { - Closed?.Invoke(ex); + _closeTcs.TrySetException(ex); } } diff --git a/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs b/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs index 4917d82394..25fdf48344 100644 --- a/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs +++ b/test/Microsoft.AspNetCore.SignalR.Tests/EndToEndTests.cs @@ -165,7 +165,6 @@ namespace Microsoft.AspNetCore.SignalR.Tests try { var receiveTcs = new TaskCompletionSource(); - var closeTcs = new TaskCompletionSource(); connection.OnReceived((data, state) => { logger.LogInformation("Received {length} byte message", data.Length); @@ -179,22 +178,6 @@ namespace Microsoft.AspNetCore.SignalR.Tests return Task.CompletedTask; }, receiveTcs); - connection.Closed += e => - { - logger.LogInformation("Connection closed"); - if (e != null) - { - receiveTcs.TrySetException(e); - closeTcs.TrySetException(e); - } - else - { - receiveTcs.TrySetResult(null); - closeTcs.TrySetResult(null); - } - return Task.CompletedTask; - }; - logger.LogInformation("Starting connection to {url}", url); await connection.StartAsync().OrTimeout(); logger.LogInformation("Started connection to {url}", url); @@ -225,8 +208,7 @@ namespace Microsoft.AspNetCore.SignalR.Tests logger.LogInformation("Receiving message"); Assert.Equal(message, await receiveTcs.Task.OrTimeout()); logger.LogInformation("Completed receive"); - - await closeTcs.Task.OrTimeout(); + await connection.Closed.OrTimeout(); } catch (Exception ex) { @@ -344,24 +326,10 @@ namespace Microsoft.AspNetCore.SignalR.Tests { var closeTcs = new TaskCompletionSource(); - connection.Closed += e => - { - logger.LogInformation("Connection closed"); - if (e != null) - { - closeTcs.TrySetException(e); - } - else - { - closeTcs.TrySetResult(null); - } - return Task.CompletedTask; - }; - logger.LogInformation("Starting connection to {url}", url); await connection.StartAsync().OrTimeout(); - await closeTcs.Task.OrTimeout(); + await connection.Closed.OrTimeout(); } catch (Exception ex) {