Refactor LongPolling in Java to avoid stackoverflow (#33564)

Author: BrennanConroy

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

@@ -4,7 +4,6 @@
 package com.microsoft.signalr;
 
 import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -15,6 +14,8 @@
 
 import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.subjects.BehaviorSubject;
 import io.reactivex.rxjava3.subjects.CompletableSubject;
 
 class LongPollingTransport implements Transport {
@@ -29,7 +30,7 @@ class LongPollingTransport implements Transport {
     private volatile Boolean active = false;
     private String pollUrl;
     private String closeError;
-    private CompletableSubject receiveLoop = CompletableSubject.create();
+    private BehaviorSubject<String> receiveLoopSubject = BehaviorSubject.create();
     private CompletableSubject closeSubject = CompletableSubject.create();
     private ExecutorService threadPool;
     private ExecutorService onReceiveThread;
@@ -78,27 +79,30 @@ public Completable start(String url) {
                 this.threadPool = Executors.newCachedThreadPool();
                 threadPool.execute(() -> {
                     this.onReceiveThread = Executors.newSingleThreadExecutor();
-                    receiveLoop.subscribe(() -> {
-                        this.stop().onErrorComplete().subscribe();
+                    receiveLoopSubject.observeOn(Schedulers.io()).subscribe(u -> {
+                        poll(u);
                     }, e -> {
                         this.stop().onErrorComplete().subscribe();
+                    }, () -> {
+                        this.stop().onErrorComplete().subscribe();
                     });
-                    poll(url).subscribeWith(receiveLoop);
+                    // start polling
+                    receiveLoopSubject.onNext(url);
                 });
 
                 return Completable.complete();
             });
         }));
     }
 
-    private Completable poll(String url) {
+    private void poll(String url) {
         if (this.active) {
             pollUrl = url + "&_=" + System.currentTimeMillis();
             logger.debug("Polling {}.", pollUrl);
-            return this.updateHeaderToken().andThen(Completable.defer(() -> {
+            this.updateHeaderToken().andThen(Completable.defer(() -> {
                 HttpRequest request = new HttpRequest();
                 request.addHeaders(headers);
-                Completable pollingCompletable = this.pollingClient.get(pollUrl, request).flatMapCompletable(response -> {
+                this.pollingClient.get(pollUrl, request).subscribe(response -> {
                     if (response.getStatusCode() == 204) {
                         logger.info("LongPolling transport terminated by server.");
                         this.active = false;
@@ -107,22 +111,32 @@ private Completable poll(String url) {
                         this.active = false;
                         this.closeError = "Unexpected response code " + response.getStatusCode() + ".";
                     } else {
-                        if (response.getContent() != null) {
+                        if (response.getContent() != null && response.getContent().hasRemaining()) {
                             logger.debug("Message received.");
-                            onReceiveThread.submit(() -> this.onReceive(response.getContent()));
+                            try {
+                                onReceiveThread.submit(() -> this.onReceive(response.getContent()));
+                            // it's possible for stop to be called while a request is in progress, if stop throws it wont wait for
+                            // an in-progress poll to complete and will shutdown the thread. We should ignore the exception so we don't
+                            // get an unhandled RX error
+                            } catch(Exception e) {}
                         } else {
                             logger.debug("Poll timed out, reissuing.");
                         }
                     }
-                    return poll(url);
+                    receiveLoopSubject.onNext(url);
+                }, e -> {
+                    receiveLoopSubject.onError(e);
                 });
 
-                return pollingCompletable;
-            }));
+                return Completable.complete();
+            }))
+            .subscribe(() -> {},
+            e -> {
+                receiveLoopSubject.onError(e);
+            });
         } else {
             logger.debug("Long Polling transport polling complete.");
-            receiveLoop.onComplete();
-            return Completable.complete();
+            receiveLoopSubject.onComplete();
         }
     }
 
@@ -162,7 +176,7 @@ public Completable stop() {
                 HttpRequest request = new HttpRequest();
                 request.addHeaders(headers);
                 return this.pollingClient.delete(this.url, request).ignoreElement()
-                    .andThen(receiveLoop)
+                    .andThen(receiveLoopSubject.ignoreElements())
                     .doOnComplete(() -> {
                         cleanup(this.closeError);
                     });