Upgrade to RxJava 3

knowm · May 19, 2024 · 45ffa6f · 45ffa6f
1 parent 52641e0
commit 45ffa6f
Show file tree

Hide file tree

Showing 151 changed files with 266 additions and 292 deletions.
diff --git a/pom.xml b/pom.xml
@@ -362,9 +362,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>
@@ -601,4 +601,4 @@
     </plugins>
   </build>
 
-</project>
+</project>
diff --git a/...ance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingAccountService.java b/...ance/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;

diff --git a/...am-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingExchange.java b/...am-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;

diff --git a/...e/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingMarketDataService.java b/...e/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;

diff --git a/...inance/src/main/java/info/bitrich/xchangestream/binance/BinanceStreamingTradeService.java b/...inance/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;

diff --git a/...ream-binance/src/main/java/info/bitrich/xchangestream/binance/BinanceUserDataChannel.java b/...ream-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;

diff --git a/...nce/src/main/java/info/bitrich/xchangestream/binance/BinanceUserDataStreamingService.java b/...nce/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;
 

diff --git a/...nce/src/test/java/info/bitrich/xchangestream/binance/BinanceFuturesPublicStreamsTest.java b/...nce/src/test/java/info/bitrich/xchangestream/binance/BinanceFuturesPublicStreamsTest.java
@@ -8,7 +8,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;

diff --git a/...rc/test/java/info/bitrich/xchangestream/binance/BinanceLiveFutureSubscriptionExample.java b/...rc/test/java/info/bitrich/xchangestream/binance/BinanceLiveFutureSubscriptionExample.java
@@ -6,7 +6,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;

diff --git a/...ance/src/test/java/info/bitrich/xchangestream/binance/BinanceLiveSubscriptionExample.java b/...ance/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;

diff --git a/...stream-binance/src/test/java/info/bitrich/xchangestream/binance/BinanceManualExample.java b/...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;

diff --git a/...am-binance/src/test/java/info/bitrich/xchangestream/binance/BinancePublicStreamsTest.java b/...am-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;

diff --git a/...rc/test/java/info/bitrich/xchangestream/binance/BinanceStreamExchangeTypeIntegration.java b/...rc/test/java/info/bitrich/xchangestream/binance/BinanceStreamExchangeTypeIntegration.java
@@ -10,7 +10,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.io.IOException;
 import org.junit.Ignore;
 import org.junit.Test;

diff --git a/...ex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingAccountService.java b/...ex/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;

diff --git a/...bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingAdapters.java b/...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;

diff --git a/...bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingExchange.java b/...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;

diff --git a/...src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingMarketDataService.java b/...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;

diff --git a/...-bitfinex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingService.java b/...-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;

diff --git a/...inex/src/main/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingTradeService.java b/...inex/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;

diff --git a/...eam-bitfinex/src/test/java/info/bitrich/xchangestream/bitfinex/BitfinexManualExample.java b/...eam-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;

diff --git a/...finex/src/test/java/info/bitrich/xchangestream/bitfinex/BitfinexStreamingServiceTest.java b/...finex/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;

diff --git a/...bitflyer/src/main/java/info/bitrich/xchangestream/bitflyer/BitflyerStreamingExchange.java b/...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;
 

diff --git a/...src/main/java/info/bitrich/xchangestream/bitflyer/BitflyerStreamingMarketDataService.java b/...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;

diff --git a/...tream-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingExchange.java b/...tream-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;
 

diff --git a/...mex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingMarketDataService.java b/...mex/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;

diff --git a/...stream-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingService.java b/...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;

diff --git a/...m-bitmex/src/main/java/info/bitrich/xchangestream/bitmex/BitmexStreamingTradeService.java b/...m-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;

diff --git a/xchange-stream-bitmex/src/test/java/info/bitrich/xchangestream/bitmex/BitmexOrderIT.java b/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();

diff --git a/xchange-stream-bitmex/src/test/java/info/bitrich/xchangestream/bitmex/BitmexTest.java b/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();
   }
 }
diff --git a/...stamp/src/main/java/info/bitrich/xchangestream/bitstamp/v2/BitstampStreamingExchange.java b/...stamp/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;
 

diff --git a/.../main/java/info/bitrich/xchangestream/bitstamp/v2/BitstampStreamingMarketDataService.java b/.../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;

diff --git a/...eam-bitstamp/src/test/java/info/bitrich/xchangestream/bitstamp/BitstampManualExample.java b/...eam-bitstamp/src/test/java/info/bitrich/xchangestream/bitstamp/BitstampManualExample.java
@@ -7,7 +7,7 @@
 import info.bitrich.xchangestream.core.StreamingExchange;
 import info.bitrich.xchangestream.core.StreamingExchangeFactory;
 import info.bitrich.xchangestream.service.ConnectableService;
-import io.reactivex.disposables.Disposable;
+import io.reactivex.rxjava3.disposables.Disposable;
 import org.apache.commons.lang3.concurrent.TimedSemaphore;
 import org.knowm.xchange.ExchangeSpecification;
 import org.knowm.xchange.currency.CurrencyPair;