[SignalR Java] Update to RxJava3 (#31426)

Author: BrennanConroy

src/SignalR/clients/java/signalr/core/build.gradle

@@ -8,7 +8,7 @@ group 'com.microsoft.signalr'
 dependencies {
     implementation 'com.google.code.gson:gson:2.8.5'
     implementation 'com.squareup.okhttp3:okhttp:3.11.0'
-    api 'io.reactivex.rxjava2:rxjava:2.2.3'
+    api 'io.reactivex.rxjava3:rxjava:3.0.11'
     implementation 'org.slf4j:slf4j-api:1.7.25'
 }
 

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

@@ -13,8 +13,8 @@
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-import io.reactivex.Single;
-import io.reactivex.subjects.SingleSubject;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.subjects.SingleSubject;
 import okhttp3.*;
 import okio.ByteString;
 

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

@@ -7,7 +7,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 class HttpRequest {
     private String method;

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

@@ -6,7 +6,7 @@
 import java.util.HashMap;
 import java.util.Map;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 import okhttp3.OkHttpClient;
 
 /**

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

@@ -19,10 +19,10 @@
 
 import com.google.gson.stream.JsonReader;
 
-import io.reactivex.Completable;
-import io.reactivex.Observable;
-import io.reactivex.Single;
-import io.reactivex.subjects.*;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.subjects.*;
 import okhttp3.OkHttpClient;
 
 /**

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

@@ -6,8 +6,8 @@
 import java.lang.reflect.Type;
 import java.util.concurrent.CancellationException;
 
-import io.reactivex.subjects.ReplaySubject;
-import io.reactivex.subjects.Subject;
+import io.reactivex.rxjava3.subjects.ReplaySubject;
+import io.reactivex.rxjava3.subjects.Subject;
 
 class InvocationRequest {
     private final Type returnType;

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

@@ -13,9 +13,9 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
-import io.reactivex.subjects.CompletableSubject;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.subjects.CompletableSubject;
 
 class LongPollingTransport implements Transport {
     private OnReceiveCallBack onReceiveCallBack;

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

@@ -11,8 +11,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.reactivex.Completable;
-import io.reactivex.subjects.CompletableSubject;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.subjects.CompletableSubject;
 import okhttp3.Headers;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;

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

@@ -5,7 +5,7 @@
 
 import java.nio.ByteBuffer;
 
-import io.reactivex.Completable;
+import io.reactivex.rxjava3.core.Completable;
 
 interface Transport {
     Completable start(String url);

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

@@ -9,7 +9,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import io.reactivex.Completable;
+import io.reactivex.rxjava3.core.Completable;
 
 class WebSocketTransport implements Transport {
     private WebSocketWrapper webSocketClient;

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

@@ -5,7 +5,7 @@
 
 import java.nio.ByteBuffer;
 
-import io.reactivex.Completable;
+import io.reactivex.rxjava3.core.Completable;
 
 abstract class WebSocketWrapper {
     public abstract Completable start();

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

@@ -19,15 +19,15 @@
 import org.junit.jupiter.api.Test;
 
 import ch.qos.logback.classic.spi.ILoggingEvent;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
-import io.reactivex.Single;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subjects.CompletableSubject;
-import io.reactivex.subjects.PublishSubject;
-import io.reactivex.subjects.ReplaySubject;
-import io.reactivex.subjects.SingleSubject;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.subjects.CompletableSubject;
+import io.reactivex.rxjava3.subjects.PublishSubject;
+import io.reactivex.rxjava3.subjects.ReplaySubject;
+import io.reactivex.rxjava3.subjects.SingleSubject;
 
 class HubConnectionTest {
     private static final String RECORD_SEPARATOR = "\u001e";
@@ -1423,7 +1423,7 @@ public void invokeNoReturnValueWaitsForCompletion() {
 
         mockTransport.receiveMessage("{\"type\":3,\"invocationId\":\"1\"}" + RECORD_SEPARATOR);
 
-        assertNull(result.timeout(30, TimeUnit.SECONDS).blockingGet());
+        assertTrue(result.blockingAwait(30, TimeUnit.SECONDS));
         assertTrue(done.get());
     }
 
@@ -1445,7 +1445,7 @@ public void invokeNoReturnValueWaitsForCompletionWithMessagePack() {
 
         mockTransport.receiveMessage(ByteBuffer.wrap(new byte[] { 0x06, (byte) 0x94, 0x03, (byte) 0x80, (byte) 0xA1, 0x31, 0x02 }));
 
-        assertNull(result.timeout(30, TimeUnit.SECONDS).blockingGet());
+        assertTrue(result.blockingAwait(30, TimeUnit.SECONDS));
         assertTrue(done.get());
     }
 
@@ -1466,7 +1466,7 @@ public void invokeCompletedByCompletionMessageWithResult() {
 
         mockTransport.receiveMessage("{\"type\":3,\"invocationId\":\"1\",\"result\":42}" + RECORD_SEPARATOR);
 
-        assertNull(result.timeout(30, TimeUnit.SECONDS).blockingGet());
+        assertTrue(result.blockingAwait(30, TimeUnit.SECONDS));
         assertTrue(done.get());
     }
 
@@ -1488,7 +1488,7 @@ public void invokeCompletedByCompletionMessageWithResultWithMessagePack() {
 
         mockTransport.receiveMessage(ByteBuffer.wrap(new byte[] { 0x07, (byte) 0x95, 0x03, (byte) 0x80, (byte) 0xA1, 0x31, 0x03, 0x2A }));
 
-        assertNull(result.timeout(30, TimeUnit.SECONDS).blockingGet());
+        assertTrue(result.blockingAwait(30, TimeUnit.SECONDS));
         assertTrue(done.get());
     }
 
@@ -1529,7 +1529,7 @@ public void invokeNoReturnValueHandlesError() {
 
         mockTransport.receiveMessage("{\"type\":3,\"invocationId\":\"1\",\"error\":\"There was an error\"}" + RECORD_SEPARATOR);
 
-        result.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertTrue(result.onErrorComplete().blockingAwait(30, TimeUnit.SECONDS));
 
         AtomicReference<String> errorMessage = new AtomicReference<>();
         result.doOnError(error -> {
@@ -1559,7 +1559,7 @@ public void invokeNoReturnValueHandlesErrorWithMessagePack() {
                 0x72, 0x65, 0x20, 0x77, 0x61, 0x73, 0x20, 0x61, 0x6E, 0x20, 0x65, 0x72, 0x72, 0x6F, 0x72 };
         mockTransport.receiveMessage(ByteBuffer.wrap(completionMessageErrorBytes));
 
-        result.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertTrue(result.onErrorComplete().blockingAwait(30, TimeUnit.SECONDS));
 
         AtomicReference<String> errorMessage = new AtomicReference<>();
         result.doOnError(error -> {
@@ -3885,6 +3885,6 @@ public void hubConnectionCloseCallsStop() {
             assertEquals(HubConnectionState.CONNECTED, hubConnection.getConnectionState());
         }
 
-        close.timeout(30, TimeUnit.SECONDS).blockingGet();
+        assertTrue(close.blockingAwait(30, TimeUnit.SECONDS));
     }
 }

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

@@ -15,8 +15,8 @@
 
 import org.junit.jupiter.api.Test;
 
-import io.reactivex.Single;
-import io.reactivex.subjects.CompletableSubject;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.subjects.CompletableSubject;
 
 public class LongPollingTransportTest {
 

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

@@ -6,9 +6,9 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 
-import io.reactivex.Completable;
-import io.reactivex.subjects.CompletableSubject;
-import io.reactivex.subjects.SingleSubject;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.subjects.CompletableSubject;
+import io.reactivex.rxjava3.subjects.SingleSubject;
 
 class MockTransport implements Transport {
     private OnReceiveCallBack onReceiveCallBack;

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

@@ -8,7 +8,7 @@
 import java.util.List;
 import java.util.Map;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 class TestHttpClient extends HttpClient {
     private TestHttpRequestHandler handler;

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

@@ -13,8 +13,8 @@
 
 import org.junit.jupiter.api.Test;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 class WebSocketTransportTest {
     @Test