Send CloseMessage from client to server (#48577)

Author: BrennanConroy

src/SignalR/clients/csharp/Client.Core/src/HubConnection.cs

@@ -489,6 +489,8 @@ private async Task StartAsyncCore(CancellationToken cancellationToken)
         {
             Log.ErrorStartingConnection(_logger, ex);
 
+            startingConnectionState.Cleanup();
+
             // Can't have any invocations to cancel, we're in the lock.
             await CloseAsync(startingConnectionState.Connection).ConfigureAwait(false);
             throw;
@@ -543,6 +545,8 @@ private async Task StopAsyncCore(bool disposing)
 
         ConnectionState? connectionState;
 
+        var connectionStateStopTask = Task.CompletedTask;
+
         try
         {
             if (disposing && _disposed)
@@ -559,6 +563,19 @@ private async Task StopAsyncCore(bool disposing)
             if (connectionState != null)
             {
                 connectionState.Stopping = true;
+                // Try to send CloseMessage
+                var writeTask = SendHubMessage(connectionState, CloseMessage.Empty);
+                if (writeTask.IsFaulted || writeTask.IsCanceled || !writeTask.IsCompleted)
+                {
+                    // Ignore exception from write, this is a best effort attempt to let the server know the client closed gracefully.
+                    // We are already closing the connection via an explicit StopAsync call from the user so don't care about any potential
+                    // errors that might happen.
+                    _ = writeTask.ContinueWith(
+                        static t => _ = t.Exception,
+                        CancellationToken.None,
+                        TaskContinuationOptions.ExecuteSynchronously | TaskContinuationOptions.OnlyOnFaulted,
+                        TaskScheduler.Default);
+                }
             }
             else
             {
@@ -579,17 +596,20 @@ private async Task StopAsyncCore(bool disposing)
                     (_serviceProvider as IDisposable)?.Dispose();
                 }
             }
+
+            if (connectionState != null)
+            {
+                // Start Stop inside the lock so a closure from the transport side at the same time as this doesn't cause an ODE
+                // But don't await the call in the lock as that could deadlock with HandleConnectionClose in the ReceiveLoop
+                connectionStateStopTask = connectionState.StopAsync();
+            }
         }
         finally
         {
             _state.ReleaseConnectionLock();
         }
 
-        // Now stop the connection we captured
-        if (connectionState != null)
-        {
-            await connectionState.StopAsync().ConfigureAwait(false);
-        }
+        await connectionStateStopTask.ConfigureAwait(false);
     }
 
     /// <summary>
@@ -1459,6 +1479,7 @@ private async Task HandleConnectionClose(ConnectionState connectionState)
 
             // Cancel any outstanding invocations within the connection lock
             connectionState.CancelOutstandingInvocations(connectionState.CloseException);
+            connectionState.Cleanup();
 
             if (connectionState.Stopping || _reconnectPolicy == null)
             {
@@ -1965,9 +1986,9 @@ public Task StopAsync()
 
         private async Task StopAsyncCore()
         {
-            Log.Stopping(_logger);
+            _hubConnection._state.AssertInConnectionLock();
 
-            _messageBuffer?.Dispose();
+            Log.Stopping(_logger);
 
             // Complete our write pipe, which should cause everything to shut down
             Log.TerminatingReceiveLoop(_logger);
@@ -1983,6 +2004,11 @@ private async Task StopAsyncCore()
             _stopTcs!.TrySetResult(null);
         }
 
+        public void Cleanup()
+        {
+            _messageBuffer?.Dispose();
+        }
+
         public async Task TimerLoop(TimerAwaitable timer)
         {
             // initialize the timers

src/SignalR/clients/csharp/Client/test/FunctionalTests/Startup.cs

@@ -59,10 +59,9 @@ public void ConfigureServices(IServiceCollection services)
                 });
         services.AddAuthentication(NegotiateDefaults.AuthenticationScheme).AddNegotiate();
 
-        // Since tests run in parallel, it's possible multiple servers will startup and read files being written by another test
-        // Use a unique directory per server to avoid this collision
-        services.AddDataProtection()
-            .PersistKeysToFileSystem(Directory.CreateDirectory(Path.GetRandomFileName()));
+        // Since tests run in parallel, it's possible multiple servers will startup,
+        // we use an ephemeral key provider to avoid filesystem contention issues
+        services.AddSingleton<IDataProtectionProvider, EphemeralDataProtectionProvider>();
     }
 
     public void Configure(IApplicationBuilder app)

src/SignalR/clients/csharp/Client/test/UnitTests/HubConnectionTests.Protocol.cs

@@ -668,7 +668,9 @@ public async Task ClientWithInherentKeepAliveDoesNotPing()
                 await hubConnection.DisposeAsync().DefaultTimeout();
                 await connection.DisposeAsync().DefaultTimeout();
 
-                Assert.Equal(0, (await connection.ReadAllSentMessagesAsync(ignorePings: false).DefaultTimeout()).Count);
+                var messages = await connection.ReadAllSentMessagesAsync(ignorePings: false).DefaultTimeout();
+                var message = Assert.Single(messages);
+                Assert.Equal("{\"type\":7}", message);
             }
             finally
             {

src/SignalR/clients/csharp/Client/test/UnitTests/HubConnectionTests.cs

@@ -193,7 +193,7 @@ await Assert.ThrowsAsync<TaskCanceledException>(() =>
             await hubConnection.StopAsync().DefaultTimeout();
 
             // Assert that InvokeAsync didn't send a message
-            Assert.Null(await connection.ReadSentTextMessageAsync().DefaultTimeout());
+            Assert.Equal("{\"type\":7}", await connection.ReadSentTextMessageAsync().DefaultTimeout());
         }
     }
 
@@ -212,7 +212,7 @@ await Assert.ThrowsAsync<TaskCanceledException>(() =>
             await hubConnection.StopAsync().DefaultTimeout();
 
             // Assert that SendAsync didn't send a message
-            Assert.Null(await connection.ReadSentTextMessageAsync().DefaultTimeout());
+            Assert.Equal("{\"type\":7}", await connection.ReadSentTextMessageAsync().DefaultTimeout());
         }
     }
 
@@ -254,7 +254,7 @@ await Assert.ThrowsAsync<TaskCanceledException>(() =>
             await hubConnection.StopAsync().DefaultTimeout();
 
             // Assert that StreamAsChannelAsync didn't send a message
-            Assert.Null(await connection.ReadSentTextMessageAsync().DefaultTimeout());
+            Assert.Equal("{\"type\":7}", await connection.ReadSentTextMessageAsync().DefaultTimeout());
         }
     }
 
@@ -273,7 +273,7 @@ public async Task StreamAsyncCanceledWhenPassedCanceledToken()
             await hubConnection.StopAsync().DefaultTimeout();
 
             // Assert that StreamAsync didn't send a message
-            Assert.Null(await connection.ReadSentTextMessageAsync().DefaultTimeout());
+            Assert.Equal("{\"type\":7}", await connection.ReadSentTextMessageAsync().DefaultTimeout());
         }
     }
 

src/SignalR/clients/csharp/Http.Connections.Client/src/Internal/WebSocketsTransport.cs

@@ -491,7 +491,6 @@ private async Task StartSending(WebSocket socket, bool ignoreFirstCanceled)
                 {
                     if (result.IsCanceled && !ignoreFirstCanceled)
                     {
-                        _logger.LogInformation("send canceled");
                         break;
                     }
 

src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/CloseMessage.java

@@ -4,6 +4,8 @@
 package com.microsoft.signalr;
 
 public final class CloseMessage extends HubMessage {
+    private final int type = HubMessageType.CLOSE.value;
+
     private final String error;
     private final boolean allowReconnect;
 

src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/GsonHubProtocol.java

@@ -19,7 +19,6 @@
 import com.google.gson.stream.JsonToken;
 
 public final class GsonHubProtocol implements HubProtocol {
-    private final JsonParser jsonParser = new JsonParser();
     private final Gson gson;
     private static final String RECORD_SEPARATOR = "\u001e";
 
@@ -95,7 +94,7 @@ public List<HubMessage> parseMessages(ByteBuffer payload, InvocationBinder binde
                         case "result":
                         case "item":
                             if (invocationId == null || binder.getReturnType(invocationId) == null) {
-                                resultToken = jsonParser.parse(reader);
+                                resultToken = JsonParser.parseReader(reader);
                             } else {
                                 result = gson.fromJson(reader, binder.getReturnType(invocationId));
                             }
@@ -123,7 +122,7 @@ public List<HubMessage> parseMessages(ByteBuffer payload, InvocationBinder binde
                                     }
                                 }
                             } else {
-                                argumentsToken = (JsonArray)jsonParser.parse(reader);
+                                argumentsToken = (JsonArray)JsonParser.parseReader(reader);
                             }
                             break;
                         case "headers":

src/SignalR/clients/java/signalr/core/src/main/java/com/microsoft/signalr/HubConnection.java

@@ -431,6 +431,9 @@ private Completable stop(String errorMessage) {
                 connectionState.stopError = errorMessage;
                 logger.error("HubConnection disconnected with an error: {}.", errorMessage);
             } else {
+                if (this.state.getHubConnectionState() == HubConnectionState.CONNECTED) {
+                    sendHubMessageWithLock(new CloseMessage());
+                }
                 logger.debug("Stopping HubConnection.");
             }
 

src/SignalR/clients/java/signalr/test/src/main/java/com/microsoft/signalr/GsonHubProtocolTest.java

@@ -56,6 +56,15 @@ public void parsePingMessage() {
         assertEquals(HubMessageType.PING, messages.get(0).getMessageType());
     }
 
+    @Test
+    public void writeCloseMessage() {
+        CloseMessage closeMessage = new CloseMessage();
+        String result = TestUtils.byteBufferToString(hubProtocol.writeMessage(closeMessage));
+        String expectedResult = "{\"type\":7,\"allowReconnect\":false}\u001E";
+
+        assertEquals(expectedResult, result);
+    }
+
     @Test
     public void parseCloseMessage() {
         String stringifiedMessage = "{\"type\":7}\u001E";

src/SignalR/clients/java/signalr/test/src/main/java/com/microsoft/signalr/HubConnectionTest.java

@@ -4022,4 +4022,21 @@ public void WebsocketStopLoggedOnce() {
             assertEquals(1, count);
         }
     }
+
+    @Test
+    public void sendsCloseMessageOnStop() throws InterruptedException {
+        MockTransport mockTransport = new MockTransport(true, false);
+        HubConnection hubConnection = TestUtils.createHubConnection("http://example.com", mockTransport);
+
+        hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
+
+        hubConnection.stop().timeout(30, TimeUnit.SECONDS).blockingAwait();
+
+        ByteBuffer[] messages = mockTransport.getSentMessages();
+
+        // handshake, close
+        assertEquals(2, messages.length);
+        String message = TestUtils.byteBufferToString(messages[1]);
+        assertEquals("{\"type\":7,\"allowReconnect\":false}" + RECORD_SEPARATOR, message);
+    }
 }