MQTT: Do not suppress reconnection for redelivery.

Signed-off-by: Yufei Cai <yufei.cai@bosch.io>
eclipse-ditto · Jul 18, 2021 · 03a77da · 03a77da
1 parent 38e7007
commit 03a77da
Showing 1 changed file with 10 additions and 8 deletions.
diff --git a/...org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractMqttClientActor.java b/...org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/AbstractMqttClientActor.java
@@ -148,8 +148,8 @@ abstract ActorRef startConsumerActor(boolean dryRun, Source source, ActorRef inb
     @Override
     public void postStop() {
         logger.info("actor stopped, stopping clients");
-        safelyDisconnectClient(client, CONSUMER);
-        safelyDisconnectClient(publisherClient, PUBLISHER);
+        safelyDisconnectClient(client, CONSUMER, true);
+        safelyDisconnectClient(publisherClient, PUBLISHER, true);
         super.postStop();
     }
 
@@ -201,7 +201,7 @@ private FSM.State<BaseClientState, BaseClientData> doReconnectConsumerClient(fin
             final BaseClientData data) {
 
         final ClientWithCancelSwitch oldClient = getClient();
-        safelyDisconnectClient(oldClient, CONSUMER);
+        safelyDisconnectClient(oldClient, CONSUMER, false);
         return stay();
     }
 
@@ -387,7 +387,8 @@ protected CompletionStage<Status.Status> doTestConnection(final TestConnection t
                     }
                     stopCommandConsumers(testSubscriptions);
                     stopChildActor(publisherActor);
-                    safelyDisconnectClient(new ClientWithCancelSwitch(testClient, new AtomicBoolean(false)), "test");
+                    safelyDisconnectClient(new ClientWithCancelSwitch(testClient, new AtomicBoolean(false)), "test",
+                            true);
                     return status;
                 });
     }
@@ -515,8 +516,8 @@ protected void doDisconnectClient(final Connection connection, @Nullable final A
     protected void cleanupResourcesForConnection() {
         stopCommandConsumers(subscriptionHandler);
         stopChildActor(publisherActor);
-        safelyDisconnectClient(client, CONSUMER);
-        safelyDisconnectClient(publisherClient, PUBLISHER);
+        safelyDisconnectClient(client, CONSUMER, true);
+        safelyDisconnectClient(publisherClient, PUBLISHER, true);
         resetClientAndSubscriptionHandler();
     }
 
@@ -532,13 +533,14 @@ protected ActorRef getPublisherActor() {
      *
      * @param clientToDisconnect the client to disconnect
      * @param name a name describing the client's purpose
+     * @param prevenntAutomaticReconnect whether automatic reconnect should be disabled.
      */
     private void safelyDisconnectClient(@Nullable final ClientWithCancelSwitch clientToDisconnect,
-            final String name) {
+            final String name, final boolean prevenntAutomaticReconnect) {
         if (clientToDisconnect != null) {
             try {
                 logger.info("Disconnecting mqtt <{}> client, ignoring any errors.", name);
-                clientToDisconnect.disconnect(true).exceptionally(error -> {
+                clientToDisconnect.disconnect(prevenntAutomaticReconnect).exceptionally(error -> {
                     logger.error(error, "Failed to disconnect client <{}>", clientToDisconnect.mqttClient);
                     return null;
                 });