Upgrade to RxJava 3

pom.xml

@@ -343,9 +343,9 @@
 
       <!-- for streaming-->
       <dependency>
-        <groupId>io.reactivex.rxjava2</groupId>
+        <groupId>io.reactivex.rxjava3</groupId>
         <artifactId>rxjava</artifactId>
-        <version>2.2.21</version>
+        <version>3.1.8</version>
       </dependency>
       <dependency>
         <groupId>com.pubnub</groupId>
@@ -583,4 +583,4 @@
     </plugins>
   </build>
 
-</project>
+</project>

xchange-stream-bankera/src/main/java/info/bitrich/xchangestream/bankera/BankeraStreamingExchange.java

@@ -4,8 +4,8 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.ConnectionStateModel.State;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.bankera.BankeraExchange;
 import org.knowm.xchange.bankera.service.BankeraMarketDataService;

xchange-stream-bankera/src/main/java/info/bitrich/xchangestream/bankera/BankeraStreamingMarketDataService.java

@@ -1,7 +1,7 @@
 package info.bitrich.xchangestream.bankera;
 
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.util.ArrayList;

xchange-stream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingAccountService.java

@@ -6,10 +6,10 @@
 import info.bitrich.xchangestream.binance.dto.OutboundAccountPositionBinanceWebsocketTransaction;
 import info.bitrich.xchangestream.core.StreamingAccountService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
-import io.reactivex.Observable;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.subjects.BehaviorSubject;
-import io.reactivex.subjects.Subject;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.subjects.BehaviorSubject;
+import io.reactivex.rxjava3.subjects.Subject;
 import org.knowm.xchange.currency.Currency;
 import org.knowm.xchange.dto.account.Balance;
 import org.knowm.xchange.exceptions.ExchangeException;

xchange-stream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingExchange.java

@@ -8,8 +8,8 @@
 import info.bitrich.xchangestream.service.netty.ConnectionStateModel.State;
 import info.bitrich.xchangestream.service.netty.WebSocketClientHandler;
 import info.bitrich.xchangestream.util.Events;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;

xchange-stream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingMarketDataService.java

@@ -14,14 +14,14 @@
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
-import io.reactivex.Observable;
-import io.reactivex.Scheduler;
-import io.reactivex.Single;
-import io.reactivex.disposables.CompositeDisposable;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.functions.Consumer;
-import io.reactivex.schedulers.Schedulers;
-import io.reactivex.subjects.BehaviorSubject;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.disposables.CompositeDisposable;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import io.reactivex.rxjava3.subjects.BehaviorSubject;
 import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;

xchange-stream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingTradeService.java

@@ -7,10 +7,10 @@
 import info.bitrich.xchangestream.binance.dto.ExecutionReportBinanceUserTransaction.ExecutionType;
 import info.bitrich.xchangestream.core.StreamingTradeService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
-import io.reactivex.Observable;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.subjects.PublishSubject;
-import io.reactivex.subjects.Subject;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.subjects.PublishSubject;
+import io.reactivex.rxjava3.subjects.Subject;
 import java.io.IOException;
 import org.knowm.xchange.currency.CurrencyPair;
 import org.knowm.xchange.derivative.FuturesContract;

xchange-stream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceUserDataChannel.java

@@ -2,8 +2,8 @@
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 
-import io.reactivex.Observable;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;

xchange-stream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceUserDataStreamingService.java

@@ -5,7 +5,8 @@
 import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
 import info.bitrich.xchangestream.service.netty.WebSocketClientCompressionAllowClientNoContextAndServerNoContextHandler;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
+import java.io.IOException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

xchange-stream-binance/src/test/java/info/bitrich/xchangestream/binance/BinanceFuturesPublicStreamsTest.java

@@ -6,7 +6,7 @@
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Ignore;

xchange-stream-binance/src/test/java/info/bitrich/xchangestream/binance/BinanceLiveFutureSubscriptionExample.java

@@ -3,7 +3,7 @@
 import info.bitrich.xchangestream.binancefuture.BinanceFutureStreamingExchange;
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;

xchange-stream-binance/src/test/java/info/bitrich/xchangestream/binance/BinanceLiveSubscriptionExample.java

@@ -2,7 +2,7 @@
 
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;

xchange-stream-binance/src/test/java/info/bitrich/xchangestream/binance/BinanceManualExample.java

@@ -3,7 +3,7 @@
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.currency.CurrencyPair;
 import org.slf4j.Logger;

xchange-stream-binance/src/test/java/info/bitrich/xchangestream/binance/BinancePublicStreamsTest.java

@@ -5,7 +5,7 @@
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Ignore;

xchange-stream-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingAccountService.java

@@ -2,7 +2,7 @@
 
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthBalance;
 import info.bitrich.xchangestream.core.StreamingAccountService;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import org.knowm.xchange.currency.Currency;
 import org.knowm.xchange.dto.account.Balance;
 import org.knowm.xchange.exceptions.ExchangeException;

xchange-stream-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingAdapters.java

@@ -9,7 +9,7 @@
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthOrder;
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthPreTrade;
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthTrade;
-import io.reactivex.annotations.Nullable;
+import io.reactivex.rxjava3.annotations.Nullable;
 import java.math.BigDecimal;
 import java.util.stream.Stream;
 import org.knowm.xchange.bitfinex.service.BitfinexAdapters;

xchange-stream-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingExchange.java

@@ -3,8 +3,8 @@
 import info.bitrich.xchangestream.core.ProductSubscription;
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.service.netty.ConnectionStateModel.State;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
 import org.apache.commons.lang3.StringUtils;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.bitfinex.BitfinexExchange;

xchange-stream-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingMarketDataService.java

@@ -15,7 +15,7 @@
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebsocketUpdateTrade;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.util.HashMap;
 import java.util.Map;
 import org.knowm.xchange.currency.Currency;

xchange-stream-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingService.java

@@ -11,10 +11,10 @@
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketUnSubscriptionMessage;
 import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.subjects.PublishSubject;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.subjects.PublishSubject;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;

xchange-stream-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingTradeService.java

@@ -4,7 +4,7 @@
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthPreTrade;
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthTrade;
 import info.bitrich.xchangestream.core.StreamingTradeService;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.util.function.Function;
 import org.knowm.xchange.currency.CurrencyPair;
 import org.knowm.xchange.dto.Order;

xchange-stream-bitfinex/src/test/java/info/bitrich/xchangestream/bitfinex/BitfinexManualExample.java

@@ -5,8 +5,8 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
 import info.bitrich.xchangestream.service.ConnectableService;
-import io.reactivex.Observable;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import org.apache.commons.lang3.concurrent.TimedSemaphore;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.currency.CurrencyPair;

xchange-stream-bitfinex/src/test/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingServiceTest.java

@@ -8,7 +8,7 @@
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthOrder;
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthPreTrade;
 import info.bitrich.xchangestream.bitfinex.dto.BitfinexWebSocketAuthTrade;
-import io.reactivex.observers.TestObserver;
+import io.reactivex.rxjava3.observers.TestObserver;
 import java.io.IOException;
 import java.math.BigDecimal;
 import org.junit.Before;

xchange-stream-bitflyer/src/main/java/info/bitrich/xchangestream/bitflyer/BitflyerStreamingExchange.java

@@ -4,7 +4,7 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.pubnub.PubnubStreamingService;
-import io.reactivex.Completable;
+import io.reactivex.rxjava3.core.Completable;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.bitflyer.BitflyerExchange;
 

xchange-stream-bitflyer/src/main/java/info/bitrich/xchangestream/bitflyer/BitflyerStreamingMarketDataService.java

@@ -11,7 +11,7 @@
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
 import info.bitrich.xchangestream.service.pubnub.PubnubStreamingService;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.util.HashMap;
 import java.util.Map;
 import org.knowm.xchange.currency.CurrencyPair;

xchange-stream-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingExchange.java

@@ -4,8 +4,8 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.ConnectionStateModel.State;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.bitmex.BitmexExchange;
 

xchange-stream-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingMarketDataService.java

@@ -15,7 +15,7 @@
 import info.bitrich.xchangestream.bitmex.dto.RawOrderBook;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;

xchange-stream-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingService.java

@@ -11,12 +11,12 @@
 import info.bitrich.xchangestream.service.netty.JsonNettyStreamingService;
 import io.netty.handler.codec.http.DefaultHttpHeaders;
 import io.netty.handler.codec.http.websocketx.extensions.WebSocketClientExtensionHandler;
-import io.reactivex.Completable;
-import io.reactivex.CompletableSource;
-import io.reactivex.Observable;
-import io.reactivex.ObservableEmitter;
-import io.reactivex.disposables.Disposable;
-import io.reactivex.schedulers.Schedulers;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.CompletableSource;
+import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.ObservableEmitter;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 import java.io.IOException;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;

xchange-stream-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingTradeService.java

@@ -1,7 +1,7 @@
 package info.bitrich.xchangestream.bitmex;
 
 import info.bitrich.xchangestream.bitmex.dto.BitmexOrder;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.util.Arrays;
 import java.util.stream.Collectors;
 import org.knowm.xchange.currency.CurrencyPair;

xchange-stream-bitmex/src/test/java/info/bitrich/xchangestream/bitmex/BitmexOrderIT.java

@@ -6,7 +6,7 @@
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
 import info.bitrich.xchangestream.util.LocalExchangeConfig;
 import info.bitrich.xchangestream.util.PropsLoader;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import java.io.IOException;
 import java.math.BigDecimal;
 import java.util.Date;
@@ -223,7 +223,7 @@ public void shouldGetExecutionOnFill() {
     executionObservable
         .test()
         .awaitCount(5)
-        .assertNever(execution -> Objects.equals(execution.getClOrdID(), clOrdId))
+        .assertValue(execution -> !Objects.equals(execution.getClOrdID(), clOrdId))
         .dispose();
 
     scheduler.shutdown();

xchange-stream-bitmex/src/test/java/info/bitrich/xchangestream/bitmex/BitmexTest.java

@@ -4,8 +4,8 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
 import info.bitrich.xchangestream.util.BookSanityChecker;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
 import java.util.concurrent.TimeUnit;
 import org.junit.After;
 import org.junit.Before;
@@ -51,10 +51,8 @@ private void awaitCompletable(Completable completable) {
   private <T> void awaitDataCount(Observable<T> observable) {
     observable
         .test()
-        .assertSubscribed()
         .assertNoErrors()
         .awaitCount(BitmexTest.MIN_DATA_COUNT)
-        .assertNoTimeout()
         .dispose();
   }
 
@@ -87,16 +85,14 @@ public void shouldHaveNoBookErrors() {
     streamingMarketDataService
         .getOrderBook(xbtUsd)
         .test()
-        .assertSubscribed()
-        .assertNoErrors()
+        .assertComplete()
         .awaitCount(10)
-        .assertNever(
+        .assertValue(
             book -> {
               String err = BookSanityChecker.hasErrors(book);
               LOG.info("err {}", err);
-              return err != null;
+              return err == null;
             })
-        .assertNoTimeout()
         .dispose();
   }
 }

xchange-stream-bitstamp/src/main/java/info/bitrich/xchangestream/bitstamp/v2/BitstampStreamingExchange.java

@@ -4,8 +4,8 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.ConnectionStateModel.State;
-import io.reactivex.Completable;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Observable;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.bitstamp.BitstampExchange;
 

xchange-stream-bitstamp/src/main/java/info/bitrich/xchangestream/bitstamp/v2/BitstampStreamingMarketDataService.java

@@ -4,7 +4,7 @@
 import info.bitrich.xchangestream.bitstamp.dto.BitstampWebSocketTransaction;
 import info.bitrich.xchangestream.core.StreamingMarketDataService;
 import info.bitrich.xchangestream.service.netty.StreamingObjectMapperHelper;
-import io.reactivex.Observable;
+import io.reactivex.rxjava3.core.Observable;
 import org.knowm.xchange.bitstamp.BitstampAdapters;
 import org.knowm.xchange.bitstamp.dto.marketdata.BitstampOrderBook;
 import org.knowm.xchange.currency.CurrencyPair;