Replace RxJava 2 with RxJava 3.

Author: wilx

build.gradle.kts

@@ -66,7 +66,7 @@ allprojects {
 /* ******************** dependencies ******************** */
 
 dependencies {
-    api("io.reactivex.rxjava2:rxjava:${property("rxjava.version")}")
+    api("io.reactivex.rxjava3:rxjava:${property("rxjava.version")}")
     api("org.reactivestreams:reactive-streams:${property("reactive-streams.version")}")
 
     implementation("io.netty:netty-buffer:${property("netty.version")}")

gradle.properties

@@ -3,7 +3,7 @@ prevVersion=1.3.4
 #
 # main dependencies
 #
-rxjava.version=2.2.21
+rxjava.version=3.1.10
 reactive-streams.version=1.0.4
 netty.version=4.1.119.Final
 jctools.version=2.1.2

reactor/src/main/java/com/hivemq/client/internal/mqtt/mqtt3/reactor/Mqtt3ReactorClientView.java

@@ -34,7 +34,7 @@
 import com.hivemq.client.mqtt.mqtt3.message.unsubscribe.Mqtt3Unsubscribe;
 import com.hivemq.client.mqtt.mqtt3.reactor.Mqtt3ReactorClient;
 import com.hivemq.client.rx.reactor.FluxWithSingle;
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 import org.jetbrains.annotations.NotNull;
 import org.reactivestreams.Publisher;
 import reactor.adapter.rxjava.RxJava2Adapter;

reactor/src/main/java/com/hivemq/client/internal/mqtt/reactor/MqttReactorClient.java

@@ -38,7 +38,7 @@
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.unsuback.Mqtt5UnsubAck;
 import com.hivemq.client.mqtt.mqtt5.reactor.Mqtt5ReactorClient;
 import com.hivemq.client.rx.reactor.FluxWithSingle;
-import io.reactivex.Flowable;
+import io.reactivex.rxjava3.core.Flowable;
 import org.jetbrains.annotations.NotNull;
 import org.reactivestreams.Publisher;
 import reactor.adapter.rxjava.RxJava2Adapter;

src/main/java/com/hivemq/client/internal/mqtt/MqttAsyncClient.java

@@ -40,8 +40,8 @@
 import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.Mqtt5Unsubscribe;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.unsuback.Mqtt5UnsubAck;
-import io.reactivex.FlowableSubscriber;
-import io.reactivex.schedulers.Schedulers;
+import io.reactivex.rxjava3.core.FlowableSubscriber;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.reactivestreams.Subscription;

src/main/java/com/hivemq/client/internal/mqtt/MqttBlockingClient.java

@@ -45,9 +45,9 @@
 import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.Mqtt5Unsubscribe;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.unsuback.Mqtt5UnsubAck;
-import io.reactivex.Flowable;
-import io.reactivex.FlowableSubscriber;
-import io.reactivex.internal.subscriptions.SubscriptionHelper;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.FlowableSubscriber;
+import io.reactivex.rxjava3.internal.subscriptions.SubscriptionHelper;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.reactivestreams.Subscription;

src/main/java/com/hivemq/client/internal/mqtt/MqttClientExecutorConfigImpl.java

@@ -17,7 +17,7 @@
 package com.hivemq.client.internal.mqtt;
 
 import com.hivemq.client.mqtt.MqttClientExecutorConfig;
-import io.reactivex.Scheduler;
+import io.reactivex.rxjava3.core.Scheduler;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 

src/main/java/com/hivemq/client/internal/mqtt/MqttClientExecutorConfigImplBuilder.java

@@ -18,7 +18,7 @@
 
 import com.hivemq.client.internal.util.Checks;
 import com.hivemq.client.mqtt.MqttClientExecutorConfigBuilder;
-import io.reactivex.Scheduler;
+import io.reactivex.rxjava3.core.Scheduler;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 

src/main/java/com/hivemq/client/internal/mqtt/MqttRxClient.java

@@ -23,7 +23,6 @@
 import com.hivemq.client.internal.mqtt.handler.publish.incoming.MqttSubscribedPublishFlowable;
 import com.hivemq.client.internal.mqtt.handler.publish.outgoing.MqttAckFlowable;
 import com.hivemq.client.internal.mqtt.handler.publish.outgoing.MqttAckSingle;
-import com.hivemq.client.internal.mqtt.handler.publish.outgoing.MqttAckSingleFlowable;
 import com.hivemq.client.internal.mqtt.handler.subscribe.MqttSubAckSingle;
 import com.hivemq.client.internal.mqtt.handler.subscribe.MqttUnsubAckSingle;
 import com.hivemq.client.internal.mqtt.message.connect.MqttConnect;
@@ -49,12 +48,11 @@
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.Mqtt5Unsubscribe;
 import com.hivemq.client.mqtt.mqtt5.message.unsubscribe.unsuback.Mqtt5UnsubAck;
 import com.hivemq.client.rx.FlowableWithSingle;
-import io.reactivex.Completable;
-import io.reactivex.Flowable;
-import io.reactivex.Scheduler;
-import io.reactivex.Single;
-import io.reactivex.functions.Function;
-import io.reactivex.internal.fuseable.ScalarCallable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.core.Single;
+import io.reactivex.rxjava3.functions.Function;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -214,20 +212,6 @@ public MqttRxClient(final @NotNull MqttClientConfig clientConfig) {
             final @NotNull Flowable<P> publishFlowable, final @NotNull Function<P, MqttPublish> publishMapper) {
 
         final Scheduler applicationScheduler = clientConfig.getExecutorConfig().getApplicationScheduler();
-        if (publishFlowable instanceof ScalarCallable) {
-            //noinspection unchecked
-            final P publish = ((ScalarCallable<P>) publishFlowable).call();
-            if (publish == null) {
-                return Flowable.empty();
-            }
-            final MqttPublish mqttPublish;
-            try {
-                mqttPublish = publishMapper.apply(publish);
-            } catch (final Throwable t) {
-                return Flowable.error(t);
-            }
-            return new MqttAckSingleFlowable(clientConfig, mqttPublish).observeOn(applicationScheduler, true);
-        }
         return new MqttAckFlowable(
                 clientConfig, publishFlowable.subscribeOn(applicationScheduler).map(publishMapper)).observeOn(
                 applicationScheduler, true);

src/main/java/com/hivemq/client/internal/mqtt/exceptions/mqtt3/Mqtt3ExceptionFactory.java

@@ -24,7 +24,7 @@
 import com.hivemq.client.mqtt.mqtt3.exceptions.*;
 import com.hivemq.client.mqtt.mqtt5.exceptions.Mqtt5MessageException;
 import com.hivemq.client.mqtt.mqtt5.message.Mqtt5Message;
-import io.reactivex.functions.Function;
+import io.reactivex.rxjava3.functions.Function;
 import org.jetbrains.annotations.NotNull;
 
 /**