Add configuration to resubscribe if session present

Author: tlopesPT

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

@@ -70,6 +70,7 @@ public class MqttClientConfig implements Mqtt5ClientConfig {
     private @Nullable SslContext currentSslContext;
     private boolean resubscribeIfSessionExpired;
     private boolean republishIfSessionExpired;
+    private boolean resubscribeIfSessionPresent;
 
     public MqttClientConfig(
             final @NotNull MqttVersion mqttVersion,
@@ -263,6 +264,14 @@ public void setResubscribeIfSessionExpired(final boolean resubscribeIfSessionExp
         this.resubscribeIfSessionExpired = resubscribeIfSessionExpired;
     }
 
+    public boolean isResubscribeIfSessionPresent() {
+        return resubscribeIfSessionPresent;
+    }
+
+    public void setResubscribeIfSessionPresent(final boolean resubscribeIfSessionPresent) {
+        this.resubscribeIfSessionPresent = resubscribeIfSessionPresent;
+    }
+
     public boolean isRepublishIfSessionExpired() {
         return republishIfSessionExpired;
     }

src/main/java/com/hivemq/client/internal/mqtt/handler/connect/MqttConnAckSingle.java

@@ -175,6 +175,7 @@ private static void reconnect(
             }, reconnector.getDelay(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS);
             clientConfig.setResubscribeIfSessionExpired(reconnector.isResubscribeIfSessionExpired());
             clientConfig.setRepublishIfSessionExpired(reconnector.isRepublishIfSessionExpired());
+            clientConfig.setResubscribeIfSessionPresent(reconnector.isResubscribeIfSessionPresent());
             reconnector.afterOnDisconnected();
         } else {
             clientConfig.getRawState().set(DISCONNECTED);

src/main/java/com/hivemq/client/internal/mqtt/handler/subscribe/MqttSubscriptionHandler.java

@@ -100,7 +100,7 @@ public void onSessionStartOrResume(
 
         subscriptionIdentifiersAvailable = connectionConfig.areSubscriptionIdentifiersAvailable();
 
-        if (!hasSession) {
+        if (!hasSession || clientConfig.isResubscribeIfSessionPresent()) {
             incomingPublishFlows.getSubscriptions().forEach((subscriptionIdentifier, subscriptions) -> {
                 final MqttSubscribe subscribe = new MqttSubscribe(ImmutableList.copyOf(subscriptions),
                         MqttUserPropertiesImpl.NO_USER_PROPERTIES);

src/main/java/com/hivemq/client/internal/mqtt/lifecycle/MqttClientReconnector.java

@@ -44,6 +44,7 @@ public class MqttClientReconnector implements Mqtt5ClientReconnector {
     private @Nullable CompletableFuture<?> future;
     private boolean resubscribeIfSessionExpired = DEFAULT_RESUBSCRIBE_IF_SESSION_EXPIRED;
     private boolean republishIfSessionExpired = DEFAULT_REPUBLISH_IF_SESSION_EXPIRED;
+    private boolean resubscribeIfSessionPresent = DEFAULT_RESUBSCRIBE_IF_SESSION_PRESENT;
     private long delayNanos = TimeUnit.MILLISECONDS.toNanos(DEFAULT_DELAY_MS);
     private @NotNull MqttClientTransportConfigImpl transportConfig;
     private @NotNull MqttConnect connect;
@@ -126,6 +127,19 @@ public boolean isRepublishIfSessionExpired() {
         return republishIfSessionExpired;
     }
 
+    @Override
+    public boolean isResubscribeIfSessionPresent() {
+        checkInEventLoop();
+        return resubscribeIfSessionPresent;
+    }
+
+    @Override
+    public @NotNull MqttClientReconnector resubscribeIfSessionPresent(final boolean resubscribeIfSessionPresent) {
+        checkInOnDisconnected("resubscribeIfSessionPresent");
+        this.resubscribeIfSessionPresent = resubscribeIfSessionPresent;
+        return this;
+    }
+
     @Override
     public @NotNull MqttClientReconnector delay(final long delay, final @Nullable TimeUnit timeUnit) {
         checkInOnDisconnected("delay");

src/main/java/com/hivemq/client/internal/mqtt/lifecycle/mqtt3/Mqtt3ClientReconnectorView.java

@@ -85,6 +85,17 @@ public boolean isRepublishIfSessionExpired() {
         return delegate.isRepublishIfSessionExpired();
     }
 
+    @Override
+    public @NotNull Mqtt3ClientReconnectorView resubscribeIfSessionPresent(final boolean resubscribeIfSessionPresent) {
+        delegate.resubscribeIfSessionPresent(resubscribeIfSessionPresent);
+        return this;
+    }
+
+    @Override
+    public boolean isResubscribeIfSessionPresent() {
+        return delegate.isResubscribeIfSessionPresent();
+    }
+
     @Override
     public int getAttempts() {
         return delegate.getAttempts();

src/main/java/com/hivemq/client/mqtt/lifecycle/MqttClientReconnector.java

@@ -60,6 +60,12 @@ public interface MqttClientReconnector {
      * @since 1.2
      */
     boolean DEFAULT_REPUBLISH_IF_SESSION_EXPIRED = false;
+    /**
+     * If resubscribe when the session is present when the client reconnected successfully is enabled by default.
+     *
+     * @since 1.3.7
+     */
+    boolean DEFAULT_RESUBSCRIBE_IF_SESSION_PRESENT = false;
     /**
      * Default delay in milliseconds the client will wait for before trying to reconnect.
      *
@@ -156,6 +162,32 @@ public interface MqttClientReconnector {
      */
     boolean isRepublishIfSessionExpired();
 
+    /**
+     * Instructs the client to automatically restore its subscriptions when reconnected successfully and the
+     * session is still present.
+     * <p>
+     * When the client reconnected successfully and its session is still present, the server still knows its
+     * subscriptions, so resubscribing is optional.
+     * <p>
+     * This setting only has effect if the client will reconnect (at least one of the methods {@link
+     * #reconnect(boolean)} or {@link #reconnectWhen(CompletableFuture, BiConsumer)} is called).
+     * <p>
+     * This method must only be called in {@link MqttClientDisconnectedListener#onDisconnected(MqttClientDisconnectedContext)}
+     * and not in the callback supplied to {@link #reconnectWhen(CompletableFuture, BiConsumer)}.
+     *
+     * @param resubscribeIfSessionPresent whether to resubscribe if the session is present when the client reconnected successfully.
+     * @return this reconnector.
+     * @throws UnsupportedOperationException if called outside of {@link MqttClientDisconnectedListener#onDisconnected(MqttClientDisconnectedContext)}.
+     * @since 1.3.7
+     */
+    @NotNull MqttClientReconnector resubscribeIfSessionPresent(boolean resubscribeIfSessionPresent);
+
+    /**
+     * @return whether the client will resubscribe if the session is present when it reconnects successfully.
+     * @since 1.3.7
+     */
+    boolean isResubscribeIfSessionPresent();
+
     /**
      * Sets a delay the client will wait for before trying to reconnect.
      * <p>

src/main/java/com/hivemq/client/mqtt/mqtt3/lifecycle/Mqtt3ClientReconnector.java

@@ -53,6 +53,9 @@ public interface Mqtt3ClientReconnector extends MqttClientReconnector {
     @Override
     @NotNull Mqtt3ClientReconnector republishIfSessionExpired(boolean republish);
 
+    @Override
+    @NotNull Mqtt3ClientReconnector resubscribeIfSessionPresent(boolean resubscribeIfSessionPresent);
+
     @Override
     @NotNull Mqtt3ClientReconnector delay(long delay, @NotNull TimeUnit timeUnit);
 

src/main/java/com/hivemq/client/mqtt/mqtt5/lifecycle/Mqtt5ClientReconnector.java

@@ -53,6 +53,9 @@ public interface Mqtt5ClientReconnector extends MqttClientReconnector {
     @Override
     @NotNull Mqtt5ClientReconnector republishIfSessionExpired(boolean republish);
 
+    @Override
+    @NotNull Mqtt5ClientReconnector resubscribeIfSessionPresent(boolean resubscribeIfSessionPresent);
+
     @Override
     @NotNull Mqtt5ClientReconnector delay(long delay, @NotNull TimeUnit timeUnit);