Replace RxJava 2 with RxJava 3.

Author: wilx

gradle/libs.versions.toml

@@ -16,7 +16,7 @@ pmd = "5.8.1"
 reactiveStreams = "1.0.4"
 reactor = "3.8.5"
 reactor-adapter = "3.6.0"
-rxjava = "2.2.21"
+rxjava = "3.1.12"
 slf4j = "1.7.36"
 testcontainers = "2.0.5"
 
@@ -49,7 +49,7 @@ reactiveStreams = { module = "org.reactivestreams:reactive-streams", version.ref
 reactor = { module = "io.projectreactor:reactor-core", version.ref = "reactor" }
 reactor-adapter = { module = "io.projectreactor.addons:reactor-adapter", version.ref = "reactor-adapter" }
 reactor-test = { module = "io.projectreactor:reactor-test", version.ref = "reactor" }
-rxjava = { module = "io.reactivex.rxjava2:rxjava", version.ref = "rxjava" }
+rxjava = { module = "io.reactivex.rxjava3:rxjava", version.ref = "rxjava" }
 slf4j-api = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" }
 slf4j-simple = { module = "org.slf4j:slf4j-simple", version.ref = "slf4j" }
 testcontainers = { module = "org.testcontainers:testcontainers", version.ref = "testcontainers" }

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;
 
 /**

src/main/java/com/hivemq/client/internal/mqtt/handler/auth/MqttReAuthCompletable.java

@@ -22,9 +22,9 @@
 import com.hivemq.client.internal.rx.CompletableFlow;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandler;
-import io.reactivex.Completable;
-import io.reactivex.CompletableObserver;
-import io.reactivex.internal.disposables.EmptyDisposable;
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.CompletableObserver;
+import io.reactivex.rxjava3.internal.disposables.EmptyDisposable;
 import org.jetbrains.annotations.NotNull;
 
 /**