Feature - RxJava2 replaced with RxJava3 implementation #2614

redisson/pom.xml

@@ -77,9 +77,9 @@
             <version>3.3.9.RELEASE</version>
         </dependency>
         <dependency>
-            <groupId>io.reactivex.rxjava2</groupId>
+            <groupId>io.reactivex.rxjava3</groupId>
             <artifactId>rxjava</artifactId>
-            <version>2.2.19</version>
+            <version>3.0.7</version>
         </dependency>
 
         <dependency>

redisson/src/main/java/org/redisson/api/CacheRx.java

@@ -23,9 +23,9 @@ import javax.cache.event.CacheEntryRemovedListener;
 import javax.cache.integration.CacheLoader;
 import javax.cache.integration.CacheWriter;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for JCache

redisson/src/main/java/org/redisson/api/RAtomicDoubleRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Reactive interface for AtomicDouble object

redisson/src/main/java/org/redisson/api/RAtomicLongRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for AtomicLong object

redisson/src/main/java/org/redisson/api/RBatchRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Maybe;
+import io.reactivex.rxjava3.core.Maybe;
 import org.redisson.client.codec.Codec;
 
 /**

redisson/src/main/java/org/redisson/api/RBinaryStreamRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 import java.nio.ByteBuffer;
 

redisson/src/main/java/org/redisson/api/RBitSetRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.BitSet;
 

redisson/src/main/java/org/redisson/api/RBlockingDequeRx.java

@@ -17,10 +17,10 @@ package org.redisson.api;
 
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Redis based BlockingDeque object

redisson/src/main/java/org/redisson/api/RBlockingQueueRx.java

@@ -18,10 +18,10 @@ package org.redisson.api;
 import java.util.Collection;
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for BlockingQueue

redisson/src/main/java/org/redisson/api/RBucketRx.java

@@ -15,9 +15,9 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/RBucketsRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.Map;
 

redisson/src/main/java/org/redisson/api/RCollectionRx.java

@@ -19,8 +19,8 @@ import java.util.Collection;
 
 import org.reactivestreams.Publisher;
 
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Common RxJava2 interface for collection object

redisson/src/main/java/org/redisson/api/RCountDownLatchRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/RDequeRx.java

@@ -15,10 +15,10 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Deque object

redisson/src/main/java/org/redisson/api/RExpirableRx.java

@@ -18,7 +18,7 @@ package org.redisson.api;
 import java.util.Date;
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Base  interface for all Redisson objects

redisson/src/main/java/org/redisson/api/RGeoRx.java

@@ -18,7 +18,7 @@ package org.redisson.api;
 import java.util.List;
 import java.util.Map;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Geospatial items holder. Reactive interface.

redisson/src/main/java/org/redisson/api/RHyperLogLogRx.java

@@ -17,8 +17,8 @@ package org.redisson.api;
 
 import java.util.Collection;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Probabilistic data structure that lets you maintain counts of millions of items with extreme space efficiency.

redisson/src/main/java/org/redisson/api/RIdGeneratorRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Id generator of <code>Long</code> type numbers.

redisson/src/main/java/org/redisson/api/RKeysRx.java

@@ -15,10 +15,10 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/RLexSortedSetRx.java

@@ -17,7 +17,7 @@ package org.redisson.api;
 
 import java.util.Collection;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for sorted set contained values of String type.

redisson/src/main/java/org/redisson/api/RListMultimapRx.java

@@ -17,7 +17,7 @@ package org.redisson.api;
 
 import java.util.List;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for List based Multimap object

redisson/src/main/java/org/redisson/api/RListRx.java

@@ -18,10 +18,10 @@ package org.redisson.api;
 import java.util.Collection;
 import java.util.List;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  *  list functions

redisson/src/main/java/org/redisson/api/RLockRx.java

@@ -17,8 +17,8 @@ package org.redisson.api;
 
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Lock object

redisson/src/main/java/org/redisson/api/RMapCacheRx.java

@@ -17,9 +17,9 @@ package org.redisson.api;
 
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * <p>Map-based cache with ability to set TTL for each entry via

redisson/src/main/java/org/redisson/api/RMapRx.java

@@ -25,10 +25,10 @@ import java.util.function.Function;
 import org.redisson.api.map.MapLoader;
 import org.redisson.api.map.MapWriter;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Redis based implementation

redisson/src/main/java/org/redisson/api/RMultimapCacheRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/RMultimapRx.java

@@ -17,7 +17,7 @@ package org.redisson.api;
 
 import java.util.Set;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Base RxJava2 interface for Multimap object

redisson/src/main/java/org/redisson/api/RObjectRx.java

@@ -19,8 +19,8 @@ import java.util.concurrent.TimeUnit;
 
 import org.redisson.client.codec.Codec;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Base RxJava2 interface for all Redisson objects

redisson/src/main/java/org/redisson/api/RPatternTopicRx.java

@@ -20,7 +20,7 @@ import java.util.List;
 import org.redisson.api.listener.PatternMessageListener;
 import org.redisson.api.listener.PatternStatusListener;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Pattern based observer for Publish Subscribe object.

redisson/src/main/java/org/redisson/api/RPermitExpirableSemaphoreRx.java

@@ -17,9 +17,9 @@ package org.redisson.api;
 
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Semaphore object with lease time parameter support for each acquired permit.

redisson/src/main/java/org/redisson/api/RQueueRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.List;
 

redisson/src/main/java/org/redisson/api/RRateLimiterRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/RReliableTopicRx.java

@@ -15,9 +15,9 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 import org.redisson.api.listener.MessageListener;
 
 /**

redisson/src/main/java/org/redisson/api/RRingBufferRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RingBuffer based queue evicts elements from the head if queue capacity became full.

redisson/src/main/java/org/redisson/api/RScoredSortedSetRx.java

@@ -24,9 +24,9 @@ import java.util.concurrent.TimeUnit;
 import org.redisson.api.RScoredSortedSet.Aggregate;
 import org.redisson.client.protocol.ScoredEntry;
 
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for scored sorted set data structure.

redisson/src/main/java/org/redisson/api/RScriptRx.java

@@ -20,9 +20,9 @@ import java.util.List;
 import org.redisson.api.RScript.Mode;
 import org.redisson.api.RScript.ReturnType;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Redis Script feature

redisson/src/main/java/org/redisson/api/RSemaphoreRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/RSetCacheRx.java

@@ -18,7 +18,7 @@ package org.redisson.api;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for RSetCache object

redisson/src/main/java/org/redisson/api/RSetMultimapRx.java

@@ -17,7 +17,7 @@ package org.redisson.api;
 
 import java.util.Set;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Set based Multimap

redisson/src/main/java/org/redisson/api/RSetRx.java

@@ -17,9 +17,9 @@ package org.redisson.api;
 
 import java.util.Set;
 
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Redis based implementation of {@link java.util.Set}

redisson/src/main/java/org/redisson/api/RSortableRx.java

@@ -18,7 +18,7 @@ package org.redisson.api;
 import java.util.Collection;
 import java.util.List;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * 

redisson/src/main/java/org/redisson/api/RStreamRx.java

@@ -19,8 +19,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * Reactive interface for Redis Stream object.

redisson/src/main/java/org/redisson/api/RTimeSeriesRx.java

@@ -15,10 +15,10 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.Collection;
 import java.util.Map;

redisson/src/main/java/org/redisson/api/RTopicRx.java

@@ -20,9 +20,9 @@ import java.util.List;
 import org.redisson.api.listener.MessageListener;
 import org.redisson.api.listener.StatusListener;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * RxJava2 interface for Publish Subscribe object. Messages are delivered to all message listeners across Redis cluster.

redisson/src/main/java/org/redisson/api/RTransactionRx.java

@@ -17,7 +17,7 @@ package org.redisson.api;
 
 import org.redisson.client.codec.Codec;
 
-import io.reactivex.Completable;
+import io.reactivex.rxjava3.core.Completable;
 
 /**
  * RxJava2 interface for transaction object allows to execute transactions over Redisson objects.

redisson/src/main/java/org/redisson/api/RTransferQueueRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.api;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 
 import java.util.concurrent.TimeUnit;
 

redisson/src/main/java/org/redisson/api/annotation/RRemoteRx.java

@@ -27,17 +27,17 @@ import java.lang.annotation.Target;
  * All method signatures must match with remote service interface,
  * but return type must be one of the following:
  *   <ul>
- *      <li>io.reactivex.Completable</li>
- *      <li>io.reactivex.Single</li>
- *      <li>io.reactivex.Maybe</li>
+ *      <li>io.reactivex.rxjava3.core.Completable</li>
+ *      <li>io.reactivex.rxjava3.core.Single</li>
+ *      <li>io.reactivex.rxjava3.core.Maybe</li>
  *   </ul>
  * <p>
  * It's not necessary to add all methods from remote service.
  * Add only those which are needed. 
  * 
- * @see io.reactivex.Completable
- * @see io.reactivex.Single
- * @see io.reactivex.Maybe
+ * @see io.reactivex.rxjava3.core.Completable
+ * @see io.reactivex.rxjava3.core.Single
+ * @see io.reactivex.rxjava3.core.Maybe
  * 
  * @author Nikita Koksharov
  *

redisson/src/main/java/org/redisson/remote/RxRemoteProxy.java

@@ -24,10 +24,10 @@ import org.redisson.command.CommandAsyncExecutor;
 import org.redisson.executor.RemotePromise;
 import org.redisson.rx.CommandRxExecutor;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/CommandRxBatchService.java

@@ -29,7 +29,7 @@ import org.redisson.connection.ConnectionManager;
 import org.redisson.connection.NodeSource;
 import org.redisson.misc.RPromise;
 
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/CommandRxExecutor.java

@@ -20,7 +20,7 @@ import java.util.concurrent.Callable;
 import org.redisson.api.RFuture;
 import org.redisson.command.CommandAsyncExecutor;
 
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 
 /**
  *

redisson/src/main/java/org/redisson/rx/CommandRxService.java

@@ -21,10 +21,10 @@ import org.redisson.api.RFuture;
 import org.redisson.command.CommandAsyncService;
 import org.redisson.connection.ConnectionManager;
 
-import io.reactivex.Flowable;
-import io.reactivex.functions.Action;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.functions.Action;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  *

redisson/src/main/java/org/redisson/rx/ElementsStream.java

@@ -21,10 +21,10 @@ import java.util.function.Supplier;
 
 import org.redisson.api.RFuture;
 
-import io.reactivex.Flowable;
-import io.reactivex.functions.Action;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.functions.Action;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/PublisherAdder.java

@@ -15,19 +15,17 @@
  */
 package org.redisson.rx;
 
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.functions.Action;
+import io.reactivex.rxjava3.functions.Consumer;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 import org.reactivestreams.Publisher;
 import org.redisson.api.RFuture;
 
-import io.reactivex.Flowable;
-import io.reactivex.Single;
-import io.reactivex.functions.Action;
-import io.reactivex.functions.Consumer;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.internal.operators.flowable.FlowableInternalHelper;
-import io.reactivex.processors.ReplayProcessor;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
 /**
  * 
@@ -48,7 +46,7 @@ public abstract class PublisherAdder<V> {
                 final AtomicBoolean completed = new AtomicBoolean();
                 final AtomicLong values = new AtomicLong();
                 final AtomicBoolean lastSize = new AtomicBoolean();
-                
+
                 cc.subscribe(new Consumer<V>() {
                     @Override
                     public void accept(V t) throws Exception {
@@ -58,7 +56,7 @@ public abstract class PublisherAdder<V> {
                                 p.onError(e);
                                 return;
                             }
-                            
+
                             if (res) {
                                 lastSize.set(true);
                             }
@@ -82,7 +80,7 @@ public abstract class PublisherAdder<V> {
                             p.onComplete();
                         }
                     }
-                }, FlowableInternalHelper.RequestMax.INSTANCE);                
+                });
             }
         }).singleOrError();
     }

redisson/src/main/java/org/redisson/rx/RedissonBatchRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.rx;
 
-import io.reactivex.Maybe;
+import io.reactivex.rxjava3.core.Maybe;
 import org.redisson.*;
 import org.redisson.api.*;
 import org.redisson.client.codec.Codec;

redisson/src/main/java/org/redisson/rx/RedissonBinaryStreamRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.rx;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 import org.redisson.RedissonBinaryStream;
 import org.redisson.api.RBinaryStream;
 import org.redisson.api.RFuture;

redisson/src/main/java/org/redisson/rx/RedissonBlockingDequeRx.java

@@ -17,7 +17,7 @@ package org.redisson.rx;
 
 import org.redisson.RedissonBlockingDeque;
 
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonBlockingQueueRx.java

@@ -18,7 +18,7 @@ package org.redisson.rx;
 import org.redisson.api.RBlockingQueueAsync;
 import org.redisson.api.RListAsync;
 
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonKeysRx.java

@@ -23,9 +23,9 @@ import org.redisson.RedissonKeys;
 import org.redisson.client.RedisClient;
 import org.redisson.connection.MasterSlaveEntry;
 
-import io.reactivex.Flowable;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonLexSortedSetRx.java

@@ -22,8 +22,8 @@ import org.redisson.api.RLexSortedSet;
 import org.redisson.client.RedisClient;
 import org.redisson.client.protocol.decoder.ListScanResult;
 
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonListRx.java

@@ -19,9 +19,9 @@ import org.reactivestreams.Publisher;
 import org.redisson.api.RFuture;
 import org.redisson.api.RListAsync;
 
-import io.reactivex.Single;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  * Distributed and concurrent implementation of {@link java.util.List}

redisson/src/main/java/org/redisson/rx/RedissonMapRxIterator.java

@@ -22,9 +22,9 @@ import java.util.concurrent.atomic.AtomicLong;
 import org.redisson.RedissonMap;
 import org.redisson.client.RedisClient;
 
-import io.reactivex.Flowable;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonReliableTopicRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.rx;
 
-import io.reactivex.Flowable;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 import org.redisson.api.RFuture;
 import org.redisson.api.RReliableTopic;
 

redisson/src/main/java/org/redisson/rx/RedissonScoredSortedSetRx.java

@@ -22,7 +22,7 @@ import org.redisson.api.RScoredSortedSetAsync;
 import org.redisson.client.RedisClient;
 import org.redisson.client.protocol.decoder.ListScanResult;
 
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonSetCacheRx.java

@@ -15,7 +15,7 @@
  */
 package org.redisson.rx;
 
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Single;
 import org.reactivestreams.Publisher;
 import org.redisson.RedissonObject;
 import org.redisson.ScanIterator;

redisson/src/main/java/org/redisson/rx/RedissonSetRx.java

@@ -15,8 +15,8 @@
  */
 package org.redisson.rx;
 
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 import org.reactivestreams.Publisher;
 import org.redisson.RedissonObject;
 import org.redisson.ScanIterator;

redisson/src/main/java/org/redisson/rx/RedissonTopicRx.java

@@ -21,10 +21,10 @@ import org.redisson.api.RFuture;
 import org.redisson.api.RTopic;
 import org.redisson.api.listener.MessageListener;
 
-import io.reactivex.Flowable;
-import io.reactivex.functions.Action;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.functions.Action;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonTransactionRx.java

@@ -31,7 +31,7 @@ import org.redisson.client.codec.Codec;
 import org.redisson.reactive.RedissonSetReactive;
 import org.redisson.transaction.RedissonTransaction;
 
-import io.reactivex.Completable;
+import io.reactivex.rxjava3.core.Completable;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/RedissonTransferQueueRx.java

@@ -15,10 +15,10 @@
  */
 package org.redisson.rx;
 
-import io.reactivex.Flowable;
-import io.reactivex.Single;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 import org.reactivestreams.Publisher;
 import org.redisson.RedissonTransferQueue;
 import org.redisson.api.RFuture;

redisson/src/main/java/org/redisson/rx/RxProxyBuilder.java

@@ -22,9 +22,9 @@ import org.redisson.api.RFuture;
 import org.redisson.misc.ProxyBuilder;
 import org.redisson.misc.ProxyBuilder.Callback;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 
 /**
  * 

redisson/src/main/java/org/redisson/rx/SetRxIterator.java

@@ -21,9 +21,9 @@ import org.redisson.api.RFuture;
 import org.redisson.client.RedisClient;
 import org.redisson.client.protocol.decoder.ListScanResult;
 
-import io.reactivex.Flowable;
-import io.reactivex.functions.LongConsumer;
-import io.reactivex.processors.ReplayProcessor;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.functions.LongConsumer;
+import io.reactivex.rxjava3.processors.ReplayProcessor;
 
 /**
  * 

redisson/src/test/java/org/redisson/RedissonRemoteServiceTest.java

@@ -33,8 +33,8 @@ import org.redisson.codec.SerializationCodec;
 import org.redisson.remote.RemoteServiceAckTimeoutException;
 import org.redisson.remote.RemoteServiceTimeoutException;
 
-import io.reactivex.Completable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Single;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
 
@@ -445,7 +445,7 @@ public class RedissonRemoteServiceTest extends BaseTest {
         RemoteInterfaceRx ri = r2.getRemoteService().get(RemoteInterfaceRx.class);
         
         Completable f = ri.voidMethod("someName", 100L);
-        f.blockingGet();
+        f.blockingAwait();
         Single<Long> resFuture = ri.resultMethod(100L);
         assertThat(resFuture.blockingGet()).isEqualTo(200);
 

redisson/src/test/java/org/redisson/rx/BaseRxTest.java

@@ -14,10 +14,10 @@ import org.redisson.api.RScoredSortedSetRx;
 import org.redisson.api.RedissonRxClient;
 import org.redisson.config.Config;
 
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 public abstract class BaseRxTest {
 
@@ -49,7 +49,7 @@ public abstract class BaseRxTest {
     }
     
     public static void sync(Completable completable) {
-        completable.blockingGet();
+        completable.blockingAwait();
     }
 
     public static <V> V sync(Single<V> single) {

redisson/src/test/java/org/redisson/rx/RedissonBatchRxTest.java

@@ -39,9 +39,9 @@ import org.redisson.client.RedisException;
 import org.redisson.client.codec.StringCodec;
 import org.redisson.config.Config;
 
-import io.reactivex.Completable;
-import io.reactivex.Maybe;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Maybe;
+import io.reactivex.rxjava3.core.Single;
 
 @RunWith(Parameterized.class)
 public class RedissonBatchRxTest extends BaseRxTest {

redisson/src/test/java/org/redisson/rx/RedissonKeysRxTest.java

@@ -1,6 +1,6 @@
 package org.redisson.rx;
 
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 import org.junit.Assert;
 import org.junit.Test;
 import org.redisson.api.RBucketRx;

redisson/src/test/java/org/redisson/rx/RedissonTopicRxTest.java

@@ -3,9 +3,12 @@ package org.redisson.rx;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.io.Serializable;
+import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
+import org.awaitility.Awaitility;
 import org.junit.Assert;
 import org.junit.Test;
 import org.reactivestreams.Subscriber;
@@ -13,8 +16,8 @@ import org.reactivestreams.Subscription;
 import org.redisson.api.RTopicRx;
 import org.redisson.api.listener.MessageListener;
 
-import io.reactivex.Flowable;
-import io.reactivex.Single;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Single;
 
 public class RedissonTopicRxTest extends BaseRxTest {
 
@@ -116,7 +119,7 @@ public class RedissonTopicRxTest extends BaseRxTest {
         for (int i = 0; i < 15; i++) {
             sync(topic.publish("" + i));
         }
-        
-        assertThat(list).containsExactly("0", "1", "2", "3", "4", "5", "6", "7", "8", "9");
+
+        Awaitility.waitAtMost(Duration.ofSeconds(10)).until(() -> list.equals(Arrays.asList("0", "1", "2", "3", "4", "5", "6", "7", "8", "9")));
     }
 }