use finally block for manual acknowledgment

hivemq · Sep 27, 2023 · 1f19344 · 1f19344
1 parent ef0f960
commit 1f19344
Show file tree

Hide file tree

Showing 2 changed files with 62 additions and 54 deletions.
diff --git a/src/main/java/com/hivemq/cli/mqtt/SubscribeMqtt3PublishCallback.java b/src/main/java/com/hivemq/cli/mqtt/SubscribeMqtt3PublishCallback.java
@@ -50,39 +50,43 @@ public class SubscribeMqtt3PublishCallback implements Consumer<Mqtt3Publish> {
 
     @Override
     public void accept(final @NotNull Mqtt3Publish mqtt3Publish) {
-        String message;
         try {
-            if (isJsonOutput) {
-                message = new JsonMqttPublish(mqtt3Publish, isBase64).toString();
-            } else {
-                message = MqttPublishUtils.formatPayload(mqtt3Publish.getPayloadAsBytes(), isBase64);
-            }
+            String message;
+            try {
+                if (isJsonOutput) {
+                    message = new JsonMqttPublish(mqtt3Publish, isBase64).toString();
+                } else {
+                    message = MqttPublishUtils.formatPayload(mqtt3Publish.getPayloadAsBytes(), isBase64);
+                }
 
-            if (showTopics) {
-                message = mqtt3Publish.getTopic() + ": " + message;
-            }
+                if (showTopics) {
+                    message = mqtt3Publish.getTopic() + ": " + message;
+                }
 
-            Logger.debug("{} received PUBLISH ('{}')\n    {}",
-                    LoggerUtils.getClientPrefix(client.getConfig()),
-                    new String(mqtt3Publish.getPayloadAsBytes(), StandardCharsets.UTF_8),
-                    mqtt3Publish);
-        } catch (final Exception e) {
-            Logger.error("An error occurred while processing an incoming PUBLISH.", e);
-            mqtt3Publish.acknowledge();
-            return;
-        }
+                Logger.debug("{} received PUBLISH ('{}')\n    {}",
+                        LoggerUtils.getClientPrefix(client.getConfig()),
+                        new String(mqtt3Publish.getPayloadAsBytes(), StandardCharsets.UTF_8),
+                        mqtt3Publish);
+            } catch (final Exception e) {
+                Logger.error("An error occurred while processing an incoming PUBLISH.", e);
+                return;
+            }
 
-        if (outputFile != null) {
-            MqttPublishUtils.printToFile(outputFile, message);
-        }
+            if (outputFile != null) {
+                MqttPublishUtils.printToFile(outputFile, message);
+            }
 
-        if (printToStdout) {
-            if (System.out.checkError()) {
-                //TODO: Handle SIGPIPE
-                //throw new RuntimeException("SIGNAL RECEIVED. PIPE CLOSED");
+            if (printToStdout) {
+                if (System.out.checkError()) {
+                    //TODO: Handle SIGPIPE
+                    //throw new RuntimeException("SIGNAL RECEIVED. PIPE CLOSED");
+                }
+                System.out.println(message);
             }
-            System.out.println(message);
+
+        } finally {
+            //Necessary to ensure log ordering
+            mqtt3Publish.acknowledge();
         }
-        mqtt3Publish.acknowledge();
     }
 }
diff --git a/src/main/java/com/hivemq/cli/mqtt/SubscribeMqtt5PublishCallback.java b/src/main/java/com/hivemq/cli/mqtt/SubscribeMqtt5PublishCallback.java
@@ -50,39 +50,43 @@ public class SubscribeMqtt5PublishCallback implements Consumer<Mqtt5Publish> {
 
     @Override
     public void accept(final @NotNull Mqtt5Publish mqtt5Publish) {
-        String message;
         try {
-            if (isJsonOutput) {
-                message = new JsonMqttPublish(mqtt5Publish, isBase64).toString();
-            } else {
-                message = MqttPublishUtils.formatPayload(mqtt5Publish.getPayloadAsBytes(), isBase64);
-            }
+            String message;
+            try {
+                if (isJsonOutput) {
+                    message = new JsonMqttPublish(mqtt5Publish, isBase64).toString();
+                } else {
+                    message = MqttPublishUtils.formatPayload(mqtt5Publish.getPayloadAsBytes(), isBase64);
+                }
 
-            if (showTopics) {
-                message = mqtt5Publish.getTopic() + ": " + message;
-            }
+                if (showTopics) {
+                    message = mqtt5Publish.getTopic() + ": " + message;
+                }
 
-            Logger.debug("{} received PUBLISH ('{}')\n    {}",
-                    LoggerUtils.getClientPrefix(client.getConfig()),
-                    new String(mqtt5Publish.getPayloadAsBytes(), StandardCharsets.UTF_8),
-                    mqtt5Publish);
-        } catch (final Exception e) {
-            Logger.error("An error occurred while processing an incoming PUBLISH.", e);
-            mqtt5Publish.acknowledge();
-            return;
-        }
+                Logger.debug("{} received PUBLISH ('{}')\n    {}",
+                        LoggerUtils.getClientPrefix(client.getConfig()),
+                        new String(mqtt5Publish.getPayloadAsBytes(), StandardCharsets.UTF_8),
+                        mqtt5Publish);
+            } catch (final Exception e) {
+                Logger.error("An error occurred while processing an incoming PUBLISH.", e);
+                return;
+            }
 
-        if (outputFile != null) {
-            MqttPublishUtils.printToFile(outputFile, message);
-        }
+            if (outputFile != null) {
+                MqttPublishUtils.printToFile(outputFile, message);
+            }
 
-        if (printToStdout) {
-            if (System.out.checkError()) {
-                //TODO: Handle SIGPIPE
-                //throw new RuntimeException("SIGNAL RECEIVED. PIPE CLOSED");
+            if (printToStdout) {
+                if (System.out.checkError()) {
+                    //TODO: Handle SIGPIPE
+                    //throw new RuntimeException("SIGNAL RECEIVED. PIPE CLOSED");
+                }
+                System.out.println(message);
             }
-            System.out.println(message);
+
+        } finally {
+            //Necessary to ensure log ordering
+            mqtt5Publish.acknowledge();
         }
-        mqtt5Publish.acknowledge();
     }
 }