Convert to restro stream APIs

build.gradle.kts

@@ -77,6 +77,8 @@ dependencies {
     implementation("org.jctools:jctools-core:${property("jctools.version")}")
     implementation("org.jetbrains:annotations:${property("annotations.version")}")
     implementation("com.google.dagger:dagger:${property("dagger.version")}")
+    implementation ("net.sourceforge.streamsupport:android-retrostreams:1.7.4")
+    implementation ("net.sourceforge.streamsupport:android-retrofuture:1.7.4")
 
     compileOnly("org.slf4j:slf4j-api:${property("slf4j.version")}")
 

examples/build.gradle.kts

@@ -17,4 +17,6 @@ metadata {
 
 dependencies {
     implementation(rootProject)
+    implementation ("net.sourceforge.streamsupport:android-retrostreams:1.7.4")
+    implementation ("net.sourceforge.streamsupport:android-retrofuture:1.7.4")
 }

examples/src/main/java/com/hivemq/client/mqtt/examples/ReconnectStrategy.java

@@ -21,7 +21,7 @@
 import com.hivemq.client.mqtt.mqtt5.lifecycle.Mqtt5ClientDisconnectedContext;
 
 import java.time.LocalTime;
-import java.util.concurrent.CompletableFuture;
+import java9.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
 /**

reactor/src/main/java/com/hivemq/client/internal/rx/reactor/operators/FluxWithSingleMap.java

@@ -28,7 +28,7 @@
 import reactor.core.Fuseable;
 import reactor.util.context.Context;
 
-import java.util.function.Function;
+import java9.util.function.Function;
 
 /**
  * @author Silvio Giebl

reactor/src/main/java/com/hivemq/client/rx/reactor/FluxWithSingle.java

@@ -36,8 +36,8 @@
 import java.util.NoSuchElementException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Consumer;
-import java.util.function.Function;
+import java9.util.function.Consumer;
+import java9.util.function.Function;
 
 /**
  * A {@link Flux} which emits a flow of items of type <code>F</code> and a single item of type <code>S</code>.

reactor/src/test/java/com/hivemq/client/rx/reactor/FluxWithSingleTest.java

@@ -38,7 +38,7 @@
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Stream;
+import java9.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.*;
 

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

@@ -46,9 +46,9 @@
 import org.jetbrains.annotations.Nullable;
 import org.reactivestreams.Subscription;
 
-import java.util.concurrent.CompletableFuture;
+import java9.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
-import java.util.function.Consumer;
+import java9.util.function.Consumer;
 
 /**
  * @author Silvio Giebl

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

@@ -52,7 +52,7 @@
 import org.reactivestreams.Subscription;
 
 import java.util.LinkedList;
-import java.util.Optional;
+import java9.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;

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

@@ -41,7 +41,7 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.Optional;
+import java9.util.Optional;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**

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

@@ -27,7 +27,7 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
-import java.util.Optional;
+import java9.util.Optional;
 
 /**
  * @author Silvio Giebl

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

@@ -22,8 +22,8 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.util.Objects;
-import java.util.Optional;
-import java.util.OptionalInt;
+import java9.util.Optional;
+import java9.util.OptionalInt;
 import java.util.concurrent.Executor;
 
 /**
@@ -93,7 +93,7 @@ public boolean equals(final @Nullable Object o) {
 
     @Override
     public int hashCode() {
-        int result = Objects.hashCode(nettyExecutor);
+        int result = java9.util.Objects.hashCode(nettyExecutor);
         result = 31 * result + nettyThreads;
         result = 31 * result + applicationScheduler.hashCode();
         return result;

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

@@ -23,7 +23,7 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.util.concurrent.Executor;
-import java.util.function.Function;
+import java9.util.function.Function;
 
 /**
  * @author Silvio Giebl

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

@@ -24,7 +24,9 @@
 import javax.net.ssl.*;
 import java.util.List;
 import java.util.Objects;
-import java.util.Optional;
+
+import java9.lang.Integers;
+import java9.util.Optional;
 
 /**
  * @author David Katz
@@ -145,12 +147,12 @@ public boolean equals(final @Nullable Object o) {
 
     @Override
     public int hashCode() {
-        int result = Objects.hashCode(keyManagerFactory);
-        result = 31 * result + Objects.hashCode(trustManagerFactory);
-        result = 31 * result + Objects.hashCode(cipherSuites);
-        result = 31 * result + Objects.hashCode(protocols);
-        result = 31 * result + Integer.hashCode(handshakeTimeoutMs);
-        result = 31 * result + Objects.hashCode(hostnameVerifier);
+        int result = java9.util.Objects.hashCode(keyManagerFactory);
+        result = 31 * result + java9.util.Objects.hashCode(trustManagerFactory);
+        result = 31 * result + java9.util.Objects.hashCode(cipherSuites);
+        result = 31 * result + java9.util.Objects.hashCode(protocols);
+        result = 31 * result + Integers.hashCode(handshakeTimeoutMs);
+        result = 31 * result + java9.util.Objects.hashCode(hostnameVerifier);
         return result;
     }
 }

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

@@ -27,7 +27,7 @@
 import javax.net.ssl.TrustManagerFactory;
 import java.util.Collection;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java9.util.function.Function;
 
 /**
  * @author Silvio Giebl

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

@@ -23,7 +23,9 @@
 
 import java.net.InetSocketAddress;
 import java.util.Objects;
-import java.util.Optional;
+
+import java9.lang.Integers;
+import java9.util.Optional;
 
 /**
  * @author Silvio Giebl
@@ -140,12 +142,12 @@ public boolean equals(final @Nullable Object o) {
     @Override
     public int hashCode() {
         int result = serverAddress.hashCode();
-        result = 31 * result + Objects.hashCode(localAddress);
-        result = 31 * result + Objects.hashCode(sslConfig);
-        result = 31 * result + Objects.hashCode(webSocketConfig);
-        result = 31 * result + Objects.hashCode(proxyConfig);
-        result = 31 * result + Integer.hashCode(socketConnectTimeoutMs);
-        result = 31 * result + Integer.hashCode(mqttConnectTimeoutMs);
+        result = 31 * result + java9.util.Objects.hashCode(localAddress);
+        result = 31 * result + java9.util.Objects.hashCode(sslConfig);
+        result = 31 * result + java9.util.Objects.hashCode(webSocketConfig);
+        result = 31 * result + java9.util.Objects.hashCode(proxyConfig);
+        result = 31 * result + Integers.hashCode(socketConnectTimeoutMs);
+        result = 31 * result + Integers.hashCode(mqttConnectTimeoutMs);
         return result;
     }
 }

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

@@ -28,7 +28,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java9.util.function.Function;
 
 import static com.hivemq.client.mqtt.MqttClient.*;
 

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

@@ -23,7 +23,9 @@
 
 import java.net.InetSocketAddress;
 import java.util.Objects;
-import java.util.Optional;
+
+import java9.lang.Integers;
+import java9.util.Optional;
 
 /**
  * @author Silvio Giebl
@@ -106,9 +108,9 @@ public boolean equals(final @Nullable Object o) {
     public int hashCode() {
         int result = protocol.hashCode();
         result = 31 * result + address.hashCode();
-        result = 31 * result + Objects.hashCode(username);
-        result = 31 * result + Objects.hashCode(password);
-        result = 31 * result + Integer.hashCode(handshakeTimeoutMs);
+        result = 31 * result + java9.util.Objects.hashCode(username);
+        result = 31 * result + java9.util.Objects.hashCode(password);
+        result = 31 * result + Integers.hashCode(handshakeTimeoutMs);
         return result;
     }
 }

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

@@ -26,7 +26,7 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java9.util.function.Function;
 
 /**
  * @author Silvio Giebl

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

@@ -20,6 +20,8 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
+import java9.lang.Integers;
+
 /**
  * @author David Katz
  * @author Christian Hoff
@@ -91,7 +93,7 @@ public int hashCode() {
         int result = serverPath.hashCode();
         result = 31 * result + queryString.hashCode();
         result = 31 * result + subprotocol.hashCode();
-        result = 31 * result + Integer.hashCode(handshakeTimeoutMs);
+        result = 31 * result + Integers.hashCode(handshakeTimeoutMs);
         return result;
     }
 }

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

@@ -22,7 +22,7 @@
 import org.jetbrains.annotations.Nullable;
 
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java9.util.function.Function;
 
 /**
  * @author Silvio Giebl

src/main/java/com/hivemq/client/internal/mqtt/advanced/MqttClientAdvancedConfig.java

@@ -82,7 +82,7 @@ public boolean equals(final @Nullable Object o) {
     public int hashCode() {
         int result = Boolean.hashCode(allowServerReAuth);
         result = 31 * result + Boolean.hashCode(validatePayloadFormat);
-        result = 31 * result + Objects.hashCode(interceptors);
+        result = 31 * result + java9.util.Objects.hashCode(interceptors);
         return result;
     }
 }