[SignalR] [Java] Log 'WebSocket stopped' once (#43532)

Author: BrennanConroy

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

@@ -439,15 +439,14 @@ private Completable stop(String errorMessage) {
             this.state.unlock();
         }
 
-        Completable stopTask = startTask.onErrorComplete().andThen(Completable.defer(() ->
+        CompletableSubject subject = CompletableSubject.create();
+        startTask.onErrorComplete().subscribe(() ->
         {
             Completable stop = connectionState.transport.stop();
-            stop.onErrorComplete().subscribe();
-            return stop;
-        }));
-        stopTask.onErrorComplete().subscribe();
+            stop.subscribe(() -> subject.onComplete(), e -> subject.onError(e));
+        });
 
-        return stopTask;
+        return subject;
     }
 
     private void ReceiveLoop(ByteBuffer payload)

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

@@ -82,7 +82,9 @@ public void setOnClose(TransportOnClosedCallback onCloseCallback) {
 
     @Override
     public Completable stop() {
-        return webSocketClient.stop().doOnEvent(t -> logger.info("WebSocket connection stopped."));
+        Completable stop = webSocketClient.stop();
+        stop.onErrorComplete().subscribe(() -> logger.info("WebSocket connection stopped."));
+        return stop;
     }
 
     void onClose(Integer code, String reason) {

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

@@ -2797,14 +2797,11 @@ public void SkippingNegotiateDoesNotNegotiate() {
                     .create("http://example")
                     .withTransport(TransportEnum.WEBSOCKETS)
                     .shouldSkipNegotiate(true)
+                    .withHandshakeResponseTimeout(1)
                     .withHttpClient(client)
                     .build();
 
-            try {
-                hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait();
-            } catch (Exception e) {
-                assertEquals("WebSockets isn't supported in testing currently.", e.getMessage());
-            }
+            assertThrows(RuntimeException.class, () -> hubConnection.start().timeout(30, TimeUnit.SECONDS).blockingAwait());
             assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
             assertFalse(negotiateCalled.get());
 
@@ -3986,4 +3983,43 @@ public void keepAliveIntervalIsSetThroughBuilder()
 
         assertEquals(interval, hubConnection.getKeepAliveInterval());
     }
+
+    @Test
+    public void WebsocketStopLoggedOnce() {
+        try (TestLogger logger = new TestLogger(WebSocketTransport.class.getName())) {
+            AtomicBoolean negotiateCalled = new AtomicBoolean(false);
+            TestHttpClient client = new TestHttpClient().on("POST", "http://example.com/negotiate?negotiateVersion=1",
+                    (req) -> {
+                        negotiateCalled.set(true);
+                        return Single.just(new HttpResponse(200, "",
+                            TestUtils.stringToByteBuffer("{\"connectionId\":\"bVOiRPG8-6YiJ6d7ZcTOVQ\",\""
+                                    + "availableTransports\":[{\"transport\":\"WebSockets\",\"transferFormats\":[\"Text\",\"Binary\"]}]}")));
+                    });
+
+            HubConnection hubConnection = HubConnectionBuilder
+                    .create("http://example")
+                    .withTransport(TransportEnum.WEBSOCKETS)
+                    .shouldSkipNegotiate(true)
+                    .withHandshakeResponseTimeout(100)
+                    .withHttpClient(client)
+                    .build();
+
+            Completable startTask = hubConnection.start().timeout(30, TimeUnit.SECONDS);
+            hubConnection.stop().timeout(30, TimeUnit.SECONDS).blockingAwait();
+
+            assertThrows(RuntimeException.class, () -> startTask.blockingAwait());
+            assertEquals(HubConnectionState.DISCONNECTED, hubConnection.getConnectionState());
+            assertFalse(negotiateCalled.get());
+
+            ILoggingEvent[] logs = logger.getLogs();
+            int count = 0;
+            for (ILoggingEvent iLoggingEvent : logs) {
+                if (iLoggingEvent.getFormattedMessage().startsWith("WebSocket connection stopped.")) {
+                    count++;
+                }
+            }
+
+            assertEquals(1, count);
+        }
+    }
 }

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

@@ -70,7 +70,7 @@ public TestHttpClient on(String method, String url, TestHttpRequestHandler handl
 
     @Override
     public WebSocketWrapper createWebSocket(String url, Map<String, String> headers) {
-        throw new RuntimeException("WebSockets isn't supported in testing currently.");
+        return new TestWebSocketWrapper(url, headers);
     }
 
     @Override

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

@@ -43,6 +43,15 @@ public void append(ILoggingEvent event) {
         this.logger.addAppender(this.appender);
     }
 
+    public ILoggingEvent[] getLogs() {
+        lock.lock();
+        try {
+            return list.toArray(new ILoggingEvent[0]);
+        } finally {
+            lock.unlock();
+        }
+    }
+
     public ILoggingEvent assertLog(String logMessage) {
         ILoggingEvent[] localList;
         lock.lock();

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

@@ -0,0 +1,68 @@
+package com.microsoft.signalr;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
+
+class WebSocketTestHttpClient extends HttpClient {
+    @Override
+    public Single<HttpResponse> send(HttpRequest request) {
+        return null;
+    }
+
+    @Override
+    public Single<HttpResponse> send(HttpRequest request, ByteBuffer body) {
+        return null;
+    }
+
+    @Override
+    public WebSocketWrapper createWebSocket(String url, Map<String, String> headers) {
+        return new TestWebSocketWrapper(url, headers);
+    }
+
+    @Override
+    public HttpClient cloneWithTimeOut(int timeoutInMilliseconds) {
+        return null;
+    }
+
+    @Override
+    public void close() {
+    }
+}
+
+class TestWebSocketWrapper extends WebSocketWrapper {
+    private WebSocketOnClosedCallback onClose;
+
+    public TestWebSocketWrapper(String url, Map<String, String> headers)
+    {
+    }
+
+    @Override
+    public Completable start() {
+        return Completable.complete();
+    }
+
+    @Override
+    public Completable stop() {
+        if (onClose != null) {
+            onClose.invoke(null, "");
+        }
+        return Completable.complete();
+    }
+
+    @Override
+    public Completable send(ByteBuffer message) {
+        return Completable.complete();
+    }
+
+    @Override
+    public void setOnReceive(OnReceiveCallBack onReceive) {
+    }
+
+    @Override
+    public void setOnClose(WebSocketOnClosedCallback onClose) {
+        this.onClose = onClose;
+    }
+}

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

@@ -5,17 +5,11 @@
 
 import static org.junit.jupiter.api.Assertions.*;
 
-import java.nio.ByteBuffer;
 import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.junit.jupiter.api.Test;
 
-import io.reactivex.rxjava3.core.Completable;
-import io.reactivex.rxjava3.core.Single;
-
 class WebSocketTransportTest {
     @Test
     public void CanPassNullExitCodeToOnClosed() {
@@ -28,61 +22,4 @@ public void CanPassNullExitCodeToOnClosed() {
         transport.stop();
         assertTrue(closed.get());
     }
-
-    class WebSocketTestHttpClient extends HttpClient {
-        @Override
-        public Single<HttpResponse> send(HttpRequest request) {
-            return null;
-        }
-
-        @Override
-        public Single<HttpResponse> send(HttpRequest request, ByteBuffer body) {
-            return null;
-        }
-
-        @Override
-        public WebSocketWrapper createWebSocket(String url, Map<String, String> headers) {
-            return new TestWrapper();
-        }
-
-        @Override
-        public HttpClient cloneWithTimeOut(int timeoutInMilliseconds) {
-            return null;
-        }
-
-        @Override
-        public void close() {
-        }
-    }
-
-    class TestWrapper extends WebSocketWrapper {
-        private WebSocketOnClosedCallback onClose;
-
-        @Override
-        public Completable start() {
-            return Completable.complete();
-        }
-
-        @Override
-        public Completable stop() {
-            if (onClose != null) {
-                onClose.invoke(null, "");
-            }
-            return Completable.complete();
-        }
-
-        @Override
-        public Completable send(ByteBuffer message) {
-            return null;
-        }
-
-        @Override
-        public void setOnReceive(OnReceiveCallBack onReceive) {
-        }
-
-        @Override
-        public void setOnClose(WebSocketOnClosedCallback onClose) {
-            this.onClose = onClose;
-        }
-    }
 }