applied manual reconnect for MQTT connections

* triggering a reconnect based on the situation, e.g. for reconenct because of redelivery, directly reconnect without any delay * don't setup new hivemq clients for the manual reconnect, but rather reconnect the existing ones * applied some simplifications caused by this new approach Signed-off-by: Thomas Jaeckle <thomas.jaeckle@bosch.io>
eclipse-ditto · May 19, 2021 · 3ecf151 · 3ecf151
1 parent bed5bb2
commit 3ecf151
Show file tree

Hide file tree

Showing 19 changed files with 140 additions and 104 deletions.
diff --git a/...ervice/src/main/java/org/eclipse/ditto/connectivity/service/config/DefaultMqttConfig.java b/...ervice/src/main/java/org/eclipse/ditto/connectivity/service/config/DefaultMqttConfig.java
@@ -22,6 +22,7 @@
 import org.eclipse.ditto.internal.utils.config.ScopedConfig;
 
 import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
 
 /**
  * This class is the default implementation of {@link MqttConfig}.
@@ -30,13 +31,15 @@
 public final class DefaultMqttConfig implements MqttConfig {
 
     private static final String CONFIG_PATH = "mqtt";
+    private static final String RECONNECT_BACKOFF_PATH = "reconnect";
 
     private final int sourceBufferSize;
     private final int eventLoopThreads;
     private final boolean cleanSession;
     private final boolean reconnectForRedelivery;
     private final boolean useSeparateClientForPublisher;
     private final Duration reconnectForRedeliveryDelay;
+    private final BackOffConfig reconnectBackOffConfig;
 
     private DefaultMqttConfig(final ScopedConfig config) {
         sourceBufferSize = config.getInt(MqttConfigValue.SOURCE_BUFFER_SIZE.getConfigPath());
@@ -46,6 +49,9 @@ private DefaultMqttConfig(final ScopedConfig config) {
         reconnectForRedeliveryDelay =
                 config.getDuration(MqttConfigValue.RECONNECT_FOR_REDELIVERY_DELAY.getConfigPath());
         useSeparateClientForPublisher = config.getBoolean(MqttConfigValue.SEPARATE_PUBLISHER_CLIENT.getConfigPath());
+        reconnectBackOffConfig = DefaultBackOffConfig.of(config.hasPath(RECONNECT_BACKOFF_PATH)
+                ? config.getConfig(RECONNECT_BACKOFF_PATH)
+                : ConfigFactory.parseString("backoff" + "={}"));
     }
 
     /**
@@ -89,6 +95,11 @@ public boolean shouldUseSeparatePublisherClient() {
         return useSeparateClientForPublisher;
     }
 
+    @Override
+    public BackOffConfig getReconnectBackOffConfig() {
+        return reconnectBackOffConfig;
+    }
+
     @Override
     public boolean equals(@Nullable final Object o) {
         if (this == o) {
@@ -103,13 +114,14 @@ public boolean equals(@Nullable final Object o) {
                 Objects.equals(cleanSession, that.cleanSession) &&
                 Objects.equals(reconnectForRedelivery, that.reconnectForRedelivery) &&
                 Objects.equals(reconnectForRedeliveryDelay, that.reconnectForRedeliveryDelay) &&
-                Objects.equals(useSeparateClientForPublisher, that.useSeparateClientForPublisher);
+                Objects.equals(useSeparateClientForPublisher, that.useSeparateClientForPublisher) &&
+                Objects.equals(reconnectBackOffConfig, that.reconnectBackOffConfig);
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(sourceBufferSize, eventLoopThreads, cleanSession, reconnectForRedelivery,
-                reconnectForRedeliveryDelay, useSeparateClientForPublisher);
+                reconnectForRedeliveryDelay, useSeparateClientForPublisher, reconnectBackOffConfig);
     }
 
     @Override
@@ -121,6 +133,7 @@ public String toString() {
                 ", reconnectForRedelivery=" + reconnectForRedelivery +
                 ", reconnectForRedeliveryDelay=" + reconnectForRedeliveryDelay +
                 ", useSeparateClientForPublisher=" + useSeparateClientForPublisher +
+                ", reconnectBackOffConfig=" + reconnectBackOffConfig +
                 "]";
     }
 

diff --git a/...ivity/service/src/main/java/org/eclipse/ditto/connectivity/service/config/MqttConfig.java b/...ivity/service/src/main/java/org/eclipse/ditto/connectivity/service/config/MqttConfig.java
@@ -74,6 +74,14 @@ public interface MqttConfig {
      */
     boolean shouldUseSeparatePublisherClient();
 
+    /**
+     * Returns the reconnect backoff configuration to apply when reconnecting failed MQTT connections.
+     *
+     * @return the reconnect backoff configuration to apply.
+     * @since 2.0.0
+     */
+    BackOffConfig getReconnectBackOffConfig();
+
     /**
      * An enumeration of the known config path expressions and their associated default values for
      * {@code MqttConfig}.

diff --git a/...eclipse/ditto/connectivity/service/messaging/backoff/DuplicationRetryTimeoutStrategy.java b/...eclipse/ditto/connectivity/service/messaging/backoff/DuplicationRetryTimeoutStrategy.java
@@ -25,6 +25,8 @@
  */
 final class DuplicationRetryTimeoutStrategy implements RetryTimeoutStrategy {
 
+    private static final Duration ZERO_MIN_DURATION_FIRST_INCREMENT = Duration.ofMillis(100);
+
     private final Duration minTimeout;
     private final Duration maxTimeout;
     private Duration currentTimeout;
@@ -54,6 +56,9 @@ public void reset() {
     public Duration getNextTimeout() {
         final Duration timeout = this.currentTimeout;
         this.increase();
+        if (timeout.isZero()) {
+            this.currentTimeout = ZERO_MIN_DURATION_FIRST_INCREMENT;
+        }
         return timeout;
     }
 

diff --git a/...n/java/org/eclipse/ditto/connectivity/service/messaging/backoff/RetryTimeoutStrategy.java b/...n/java/org/eclipse/ditto/connectivity/service/messaging/backoff/RetryTimeoutStrategy.java
@@ -15,10 +15,24 @@
 
 import java.time.Duration;
 
+import org.eclipse.ditto.connectivity.service.config.TimeoutConfig;
+
 /**
  * Retry timeout strategy that provides increasing timeouts for retrying to connect.
  */
-interface RetryTimeoutStrategy {
+public interface RetryTimeoutStrategy {
+
+    /**
+     * Creates a new RetryTimeoutStrategy duplicating the {@link #getNextTimeout()} until the configured max timeout
+     * of the passed {@code timeoutConfig} is reached with the formula: {@code timeout = minTimeout * 2^x}
+     *
+     * @param timeoutConfig the timeout config to apply.
+     * @return the created RetryTimeoutStrategy
+     * @since 2.0.0
+     */
+    static RetryTimeoutStrategy newDuplicationRetryTimeoutStrategy(final TimeoutConfig timeoutConfig) {
+        return DuplicationRetryTimeoutStrategy.fromConfig(timeoutConfig);
+    }
 
     /**
      * Resets the timeout and the current tries.

diff --git a/...rg/eclipse/ditto/connectivity/service/messaging/monitoring/logs/DefaultEvictingQueue.java b/...rg/eclipse/ditto/connectivity/service/messaging/monitoring/logs/DefaultEvictingQueue.java
@@ -100,7 +100,7 @@ public int hashCode() {
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
-                ", capacity=" + capacity +
+                "capacity=" + capacity +
                 ", elements=" + elements +
                 "]";
     }

diff --git a/...ditto/connectivity/service/messaging/monitoring/logs/DefaultMuteableConnectionLogger.java b/...ditto/connectivity/service/messaging/monitoring/logs/DefaultMuteableConnectionLogger.java
@@ -175,7 +175,7 @@ public int hashCode() {
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
-                ", connectionId=" + connectionId +
+                "connectionId=" + connectionId +
                 ", delegate=" + delegate +
                 ", active=" + active +
                 "]";

diff --git a/...clipse/ditto/connectivity/service/messaging/monitoring/logs/EvictingConnectionLogger.java b/...clipse/ditto/connectivity/service/messaging/monitoring/logs/EvictingConnectionLogger.java
@@ -237,7 +237,7 @@ public int hashCode() {
     @Override
     public String toString() {
         return getClass().getSimpleName() + " [" +
-                ", category=" + category +
+                "category=" + category +
                 ", type=" + type +
                 ", successLogs=" + successLogs +
                 ", failureLogs=" + failureLogs +

diff --git a/...lipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractHiveMqttClientFactory.java b/...lipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractHiveMqttClientFactory.java
@@ -114,7 +114,6 @@ <T extends MqttClientBuilderBase<T>> T configureClientBuilder(
             final T newBuilder,
             final Connection connection,
             final String identifier,
-            final boolean allowReconnect,
             @Nullable final MqttClientConnectedListener connectedListener,
             @Nullable final MqttClientDisconnectedListener disconnectedListener,
             final ConnectionLogger connectionLogger,
@@ -135,10 +134,6 @@ <T extends MqttClientBuilderBase<T>> T configureClientBuilder(
                 .serverHost(uri.getHost())
                 .serverPort(uri.getPort());
 
-        if (allowReconnect && connection.isFailoverEnabled()) {
-            builder = builder.automaticReconnectWithDefaultConfig();
-        }
-
         if (isSecuredConnection(connection.getProtocol())) {
 
             // create DittoTrustManagerFactory to apply hostname verification