Cleanup resources in Java client (#20473)

Author: BrennanConroy

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/DefaultHttpClient.java

@@ -29,11 +29,17 @@ public DefaultHttpClient cloneWithTimeOut(int timeoutInMilliseconds) {
         return new DefaultHttpClient(timeoutInMilliseconds, newClient);
     }
 
+    @Override
+    public void close() {
+        if (this.client != null) {
+            this.client.dispatcher().executorService().shutdown();
+        }
+    }
+
     public DefaultHttpClient(int timeoutInMilliseconds, OkHttpClient client) {
         if (client != null) {
             this.client = client;
         } else {
-
             OkHttpClient.Builder builder = new OkHttpClient.Builder().cookieJar(new CookieJar() {
                 private List<Cookie> cookieList = new ArrayList<>();
                 private Lock cookieLock = new ReentrantLock();

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/HttpClient.java

@@ -74,7 +74,7 @@ public String getStatusText() {
     }
 }
 
-abstract class HttpClient {
+abstract class HttpClient implements AutoCloseable {
     public Single<HttpResponse> get(String url) {
         HttpRequest request = new HttpRequest();
         request.setUrl(url);
@@ -127,4 +127,6 @@ public Single<HttpResponse> delete(String url, HttpRequest options) {
     public abstract WebSocketWrapper createWebSocket(String url, Map<String, String> headers);
 
     public abstract HttpClient cloneWithTimeOut(int timeoutInMilliseconds);
+
+    public abstract void close();
 }

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

@@ -24,13 +24,14 @@
 /**
  * A connection used to invoke hub methods on a SignalR Server.
  */
-public class HubConnection {
+public class HubConnection implements AutoCloseable {
     private static final String RECORD_SEPARATOR = "\u001e";
     private static final List<Class<?>> emptyArray = new ArrayList<>();
     private static final int MAX_NEGOTIATE_ATTEMPTS = 100;
 
     private String baseUrl;
     private Transport transport;
+    private boolean customTransport = false;
     private OnReceiveCallBack callback;
     private final CallbackMap handlers = new CallbackMap();
     private HubProtocol protocol;
@@ -59,6 +60,7 @@ public class HubConnection {
     private String connectionId;
     private final int negotiateVersion = 1;
     private final Logger logger = LoggerFactory.getLogger(HubConnection.class);
+    private ScheduledExecutorService handshakeTimeout = null;
 
     /**
      * Sets the server timeout interval for the connection.
@@ -111,7 +113,7 @@ void setTickRate(long tickRateInMilliseconds) {
     }
 
     // For testing purposes
-    Map<String,Observable> getStreamMap() {
+    Map<String, Observable> getStreamMap() {
         return this.streamMap;
     }
 
@@ -146,6 +148,7 @@ Transport getTransport() {
 
         if (transport != null) {
             this.transport = transport;
+            this.customTransport = true;
         } else if (transportEnum != null) {
             this.transportEnum = transportEnum;
         }
@@ -246,8 +249,8 @@ Transport getTransport() {
     }
 
     private void timeoutHandshakeResponse(long timeout, TimeUnit unit) {
-        ScheduledExecutorService scheduledThreadPool = Executors.newSingleThreadScheduledExecutor();
-        scheduledThreadPool.schedule(() -> {
+        handshakeTimeout = Executors.newSingleThreadScheduledExecutor();
+        handshakeTimeout.schedule(() -> {
             // If onError is called on a completed subject the global error handler is called
             if (!(handshakeResponseSubject.hasComplete() || handshakeResponseSubject.hasThrowable()))
             {
@@ -531,6 +534,15 @@ private void stopConnection(String errorMessage) {
             transportEnum = TransportEnum.ALL;
             this.localHeaders.clear();
             this.streamMap.clear();
+
+            if (this.handshakeTimeout != null) {
+                this.handshakeTimeout.shutdownNow();
+                this.handshakeTimeout = null;
+            }
+
+            if (this.customTransport == false) {
+                this.transport = null;
+            }
         } finally {
             hubConnectionStateLock.unlock();
         }
@@ -1097,4 +1109,16 @@ public List<Class<?>> getParameterTypes(String methodName) {
             return handlers.get(0).getClasses();
         }
     }
+
+    @Override
+    public void close() {
+        try {
+            stop().blockingAwait();
+        } finally {
+            // Don't close HttpClient if it's passed in by the user
+            if (this.httpClient != null && this.httpClient instanceof DefaultHttpClient) {
+                this.httpClient.close();
+            }
+        }
+    }
 }

src/SignalR/clients/java/signalr/src/main/java/com/microsoft/signalr/LongPollingTransport.java

@@ -160,6 +160,8 @@ public Completable stop() {
                 CompletableSubject stopCompletableSubject = CompletableSubject.create();
                 return this.receiveLoop.andThen(Completable.defer(() -> {
                     logger.info("LongPolling transport stopped.");
+                    this.onReceiveThread.shutdown();
+                    this.threadPool.shutdown();
                     this.onClose.invoke(this.closeError);
                     return Completable.complete();
                 })).subscribeWith(stopCompletableSubject);

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

@@ -2479,7 +2479,7 @@ public void hubConnectionCanBeStartedAfterBeingStopped() {
     }
 
     @Test
-    public void hubConnectionCanBeStartedAfterBeingStoppedAndRedirected()  {
+    public void hubConnectionCanBeStartedAfterBeingStoppedAndRedirected() {
         MockTransport mockTransport = new MockTransport();
         TestHttpClient client = new TestHttpClient()
                 .on("POST", "http://example.com/negotiate?negotiateVersion=1", (req) -> Single.just(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\"}")))
@@ -2519,4 +2519,30 @@ public void non200FromNegotiateThrowsError() {
             () -> hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait());
         assertEquals("Unexpected status code returned from negotiate: 500 Internal server error.", exception.getMessage());
     }
+
+    @Test
+    public void hubConnectionCloseCallsStop() throws Exception {
+        MockTransport mockTransport = new MockTransport();
+        TestHttpClient client = new TestHttpClient()
+                .on("POST", "http://example.com/negotiate?negotiateVersion=1", (req) -> Single.just(new HttpResponse(200, "", "{\"url\":\"http://testexample.com/\"}")))
+                .on("POST", "http://testexample.com/negotiate?negotiateVersion=1", (req) -> Single.just(new HttpResponse(200, "", "{\"connectionId\":\"bVOiRPG8-6YiJ6d7ZcTOVQ\",\""
+                        + "availableTransports\":[{\"transport\":\"WebSockets\",\"transferFormats\":[\"Text\",\"Binary\"]}]}")));
+
+        CompletableSubject close = CompletableSubject.create();
+
+        try (HubConnection hubConnection = HubConnectionBuilder
+                .create("http://example.com")
+                .withTransportImplementation(mockTransport)
+                .withHttpClient(client)
+                .build()) {
+
+            hubConnection.onClosed(e -> {
+                close.onComplete();
+            });
+            hubConnection.start().timeout(1, TimeUnit.SECONDS).blockingAwait();
+            assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
+        }
+
+        close.timeout(1, TimeUnit.SECONDS).blockingGet();
+    }
 }

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

@@ -12,6 +12,7 @@
 class TestHttpClient extends HttpClient {
     private TestHttpRequestHandler handler;
     private List<HttpRequest> sentRequests;
+    private boolean closeCalled;
 
     public TestHttpClient() {
         this.sentRequests = new ArrayList<>();
@@ -76,6 +77,15 @@ public HttpClient cloneWithTimeOut(int timeoutInMilliseconds) {
         return this;
     }
 
+    @Override
+    public void close() {
+        this.closeCalled = true;
+    }
+
+    public boolean getCloseCalled() {
+        return this.closeCalled;
+    }
+
     interface TestHttpRequestHandler {
         Single<HttpResponse> invoke(HttpRequest request);
     }

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

@@ -55,6 +55,10 @@ public WebSocketWrapper createWebSocket(String url, Map<String, String> headers)
         public HttpClient cloneWithTimeOut(int timeoutInMilliseconds) {
             return null;
         }
+
+        @Override
+        public void close() {
+        }
     }
 
     class TestWrapper extends WebSocketWrapper {

src/SignalR/clients/java/signalr/src/test/java/com/microsoft/signalr/sample/Chat.java

@@ -10,33 +10,33 @@
 
 
 public class Chat {
-    public static void main(String[] args) {
+    public static void main(final String[] args) throws Exception {
         System.out.println("Enter the URL of the SignalR Chat you want to join");
-        Scanner reader = new Scanner(System.in);  // Reading from System.in
-        String input = reader.nextLine();
-
-        HubConnection hubConnection = HubConnectionBuilder.create(input).build();
-
-        hubConnection.on("Send", (message) -> {
-            System.out.println(message);
-        }, String.class);
-
-        hubConnection.onClosed((ex) -> {
-            if (ex != null) {
-                System.out.printf("There was an error: %s", ex.getMessage());
+        final Scanner reader = new Scanner(System.in); // Reading from System.in
+        final String input = reader.nextLine();
+
+        try (HubConnection hubConnection = HubConnectionBuilder.create(input).build()) {
+            hubConnection.on("Send", (message) -> {
+                System.out.println(message);
+            }, String.class);
+
+            hubConnection.onClosed((ex) -> {
+                if (ex != null) {
+                    System.out.printf("There was an error: %s", ex.getMessage());
+                }
+            });
+
+            //This is a blocking call
+            hubConnection.start().blockingAwait();
+
+            String message = "";
+            while (!message.equals("leave")) {
+                // Scans the next token of the input as an int.
+                message = reader.nextLine();
+                hubConnection.send("Send", message);
             }
-        });
 
-        //This is a blocking call
-        hubConnection.start().blockingAwait();
-
-        String message = "";
-        while (!message.equals("leave")) {
-            // Scans the next token of the input as an int.
-            message = reader.nextLine();
-            hubConnection.send("Send", message);
+            hubConnection.stop().blockingAwait();
         }
-
-        hubConnection.stop().blockingAwait();
     }
 }