Refactor MQTT handling

build.gradle.kts

@@ -229,6 +229,8 @@ dependencies {
     testImplementation("org.mockito:mockito-core:${property("mockito.version")}")
     testImplementation("com.squareup.okhttp3:mockwebserver:${property("ok-http.version")}")
     testImplementation("com.ginsberg:junit5-system-exit:${property("system-exit.version")}")
+    testImplementation("com.github.stefanbirkner:system-lambda:${property("system-lambda.version")}")
+    testImplementation("org.assertj:assertj-core:${property("assertj.version")}")
 }
 
 /* ******************** integration Tests ******************** */

gradle.properties

@@ -30,6 +30,8 @@ tinylog.version=2.5.0
 junit-jupiter.version=5.9.0
 mockito.version=4.7.0
 system-exit.version=1.1.2
+system-lambda.version=1.2.1
+assertj.version=3.24.2
 #
 # integration test dependencies
 #

src/main/java/com/hivemq/cli/MqttCLIMain.java

@@ -18,21 +18,11 @@
 
 import com.hivemq.cli.ioc.DaggerMqttCLI;
 import com.hivemq.cli.ioc.MqttCLI;
-import com.hivemq.cli.mqtt.ClientData;
-import com.hivemq.cli.mqtt.ClientKey;
-import com.hivemq.cli.mqtt.MqttClientExecutor;
-import com.hivemq.client.mqtt.MqttClient;
-import com.hivemq.client.mqtt.mqtt3.Mqtt3Client;
-import com.hivemq.client.mqtt.mqtt5.Mqtt5Client;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import picocli.CommandLine;
 
 import java.security.Security;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.CompletableFuture;
 
 public class MqttCLIMain {
 
@@ -58,39 +48,12 @@ public static void main(final @NotNull String... args) {
             System.exit(0);
         }
 
-        Runtime.getRuntime().addShutdownHook(new DisconnectAllClientsTask());
-
         final int exitCode = commandLine.execute(args);
 
         System.exit(exitCode);
 
     }
 
-    private static class DisconnectAllClientsTask extends Thread {
-
-        @Override
-        public void run() {
-            final Map<ClientKey, ClientData> clientKeyToClientData = MqttClientExecutor.getClientDataMap();
-
-            final List<CompletableFuture<Void>> disconnectFutures = new ArrayList<>();
-
-            for (final Map.Entry<ClientKey, ClientData> entry : clientKeyToClientData.entrySet()) {
-                final MqttClient client = entry.getValue().getClient();
-                if (client.getConfig().getState().isConnectedOrReconnect()) {
-                    switch (client.getConfig().getMqttVersion()) {
-                        case MQTT_5_0:
-                            disconnectFutures.add(((Mqtt5Client) client).toAsync().disconnect());
-                            break;
-                        case MQTT_3_1_1:
-                            disconnectFutures.add(((Mqtt3Client) client).toAsync().disconnect());
-                            break;
-                    }
-                }
-            }
-            CompletableFuture.allOf(disconnectFutures.toArray(new CompletableFuture<?>[0])).join();
-        }
-    }
-
     public static class CLIVersionProvider implements CommandLine.IVersionProvider {
 
         @Override

src/main/java/com/hivemq/cli/commands/cli/PublishCommand.java

@@ -21,9 +21,8 @@
 import com.hivemq.cli.commands.options.DebugOptions;
 import com.hivemq.cli.commands.options.DefaultOptions;
 import com.hivemq.cli.commands.options.PublishOptions;
-import com.hivemq.cli.mqtt.MqttClientExecutor;
+import com.hivemq.cli.mqtt.clients.CliMqttClient;
 import com.hivemq.cli.utils.LoggerUtils;
-import com.hivemq.client.mqtt.MqttClient;
 import org.jetbrains.annotations.NotNull;
 import org.tinylog.Logger;
 import picocli.CommandLine;
@@ -55,11 +54,8 @@ public class PublishCommand implements Callable<Integer> {
     @CommandLine.Mixin
     private final @NotNull DefaultOptions defaultOptions = new DefaultOptions();
 
-    private final @NotNull MqttClientExecutor mqttClientExecutor;
-
     @Inject
-    public PublishCommand(final @NotNull MqttClientExecutor mqttClientExecutor) {
-        this.mqttClientExecutor = mqttClientExecutor;
+    public PublishCommand() {
     }
 
     @Override
@@ -80,17 +76,17 @@ public PublishCommand(final @NotNull MqttClientExecutor mqttClientExecutor) {
         publishOptions.logUnusedOptions(connectOptions.getVersion());
         publishOptions.arrangeQosToMatchTopics();
 
-        final MqttClient client;
+        final CliMqttClient client;
         try {
-            client = mqttClientExecutor.connect(connectOptions, null);
-        } catch (final Exception exception) {
+            client = CliMqttClient.connectWith(connectOptions).send();
+        } catch (final @NotNull Exception exception) {
             LoggerUtils.logCommandError("Unable to connect", exception, debugOptions);
             return 1;
         }
 
         try {
-            mqttClientExecutor.publish(client, publishOptions);
-        } catch (final Exception exception) {
+            client.publish(publishOptions);
+        } catch (final @NotNull Exception exception) {
             LoggerUtils.logCommandError("Unable to publish", exception, debugOptions);
             return 1;
         }
@@ -111,8 +107,6 @@ public PublishCommand(final @NotNull MqttClientExecutor mqttClientExecutor) {
                 debugOptions +
                 ", defaultOptions=" +
                 defaultOptions +
-                ", mqttClientExecutor=" +
-                mqttClientExecutor +
                 '}';
     }
 }

src/main/java/com/hivemq/cli/commands/cli/SubscribeCommand.java

@@ -20,18 +20,15 @@
 import com.hivemq.cli.commands.options.ConnectOptions;
 import com.hivemq.cli.commands.options.DebugOptions;
 import com.hivemq.cli.commands.options.DefaultOptions;
+import com.hivemq.cli.commands.options.DisconnectOptions;
 import com.hivemq.cli.commands.options.SubscribeOptions;
-import com.hivemq.cli.mqtt.MqttClientExecutor;
+import com.hivemq.cli.mqtt.clients.CliMqttClient;
 import com.hivemq.cli.utils.LoggerUtils;
-import com.hivemq.client.mqtt.MqttClient;
-import com.hivemq.client.mqtt.exceptions.ConnectionFailedException;
 import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
 import org.tinylog.Logger;
 import picocli.CommandLine;
 
 import javax.inject.Inject;
-import java.util.Objects;
 import java.util.concurrent.Callable;
 
 @CommandLine.Command(name = "sub",
@@ -41,8 +38,6 @@
 public class SubscribeCommand implements Callable<Integer> {
 
     private static final int IDLE_TIME = 5000;
-    private final @NotNull MqttClientExecutor mqttClientExecutor;
-    private @Nullable MqttClient subscribeClient;
 
     @SuppressWarnings("unused")
     @CommandLine.Option(names = {"-l"},
@@ -73,8 +68,7 @@ private void printToSTDOUT(final boolean printToSTDOUT) {
     private final @NotNull DefaultOptions defaultOptions = new DefaultOptions();
 
     @Inject
-    public SubscribeCommand(final @NotNull MqttClientExecutor mqttClientExecutor) {
-        this.mqttClientExecutor = mqttClientExecutor;
+    public SubscribeCommand() {
     }
 
     @Override
@@ -100,43 +94,44 @@ public SubscribeCommand(final @NotNull MqttClientExecutor mqttClientExecutor) {
             return 1;
         }
 
+        final CliMqttClient client;
         try {
-            subscribeClient = mqttClientExecutor.connect(connectOptions, subscribeOptions);
-        } catch (final Exception exception) {
+            client = CliMqttClient.connectWith(connectOptions)
+                    .subscribeOptions(subscribeOptions)
+                    .send();
+        } catch (final @NotNull Exception exception) {
             LoggerUtils.logCommandError("Unable to connect", exception, debugOptions);
             return 1;
         }
 
+        Runtime.getRuntime().addShutdownHook(new Thread(() -> client.disconnect(new DisconnectOptions())));
+
         try {
-            mqttClientExecutor.subscribe(subscribeClient, subscribeOptions);
-        } catch (final ConnectionFailedException exception) {
+            client.subscribe(subscribeOptions);
+        } catch (final @NotNull Exception exception) {
             LoggerUtils.logCommandError("Unable to subscribe", exception, debugOptions);
             return 1;
         }
 
         try {
-            stay();
-        } catch (final InterruptedException exception) {
+            stay(client);
+        } catch (final @NotNull InterruptedException exception) {
             LoggerUtils.logCommandError("Unable to stay", exception, debugOptions);
             return 1;
         }
 
         return 0;
     }
 
-    private void stay() throws InterruptedException {
-        while (Objects.requireNonNull(subscribeClient).getState().isConnectedOrReconnect()) {
+    private void stay(final @NotNull CliMqttClient client) throws InterruptedException {
+        while (client.isConnected()) {
             Thread.sleep(IDLE_TIME);
         }
     }
 
     @Override
     public @NotNull String toString() {
         return "SubscribeCommand{" +
-                "mqttClientExecutor=" +
-                mqttClientExecutor +
-                ", subscribeClient=" +
-                subscribeClient +
                 ", logToLogfile=" +
                 logToLogfile +
                 ", connectOptions=" +

src/main/java/com/hivemq/cli/commands/options/ConnectOptions.java

@@ -125,7 +125,7 @@ public int getPort() {
         return sessionExpiryInterval;
     }
 
-    public @Nullable Mqtt5UserProperties getConnectUserProperties() {
+    public @NotNull Mqtt5UserProperties getUserProperties() {
         return MqttUtils.convertToMqtt5UserProperties(connectUserProperties);
     }
 

src/main/java/com/hivemq/cli/commands/options/DisconnectOptions.java

@@ -74,7 +74,7 @@ public boolean isDisconnectAll() {
         return reasonString;
     }
 
-    public @Nullable Mqtt5UserProperties getUserProperties() {
+    public @NotNull Mqtt5UserProperties getUserProperties() {
         return MqttUtils.convertToMqtt5UserProperties(userProperties);
     }
 

src/main/java/com/hivemq/cli/commands/options/PublishOptions.java

@@ -127,7 +127,7 @@ public class PublishOptions {
         return correlationData;
     }
 
-    public @Nullable Mqtt5UserProperties getUserProperties() {
+    public @NotNull Mqtt5UserProperties getUserProperties() {
         return MqttUtils.convertToMqtt5UserProperties(userProperties);
     }
 

src/main/java/com/hivemq/cli/commands/options/SubscribeOptions.java

@@ -106,7 +106,7 @@ public boolean isJsonOutput() {
         return jsonOutput;
     }
 
-    public @Nullable Mqtt5UserProperties getUserProperties() {
+    public @NotNull Mqtt5UserProperties getUserProperties() {
         return MqttUtils.convertToMqtt5UserProperties(userProperties);
     }
 

src/main/java/com/hivemq/cli/commands/options/WillOptions.java

@@ -143,7 +143,7 @@ public class WillOptions {
         return willCorrelationData;
     }
 
-    public @Nullable Mqtt5UserProperties getWillUserProperties() {
+    public @NotNull Mqtt5UserProperties getUserProperties() {
         return MqttUtils.convertToMqtt5UserProperties(willUserProperties);
     }
 

src/main/java/com/hivemq/cli/commands/shell/ContextDisconnectCommand.java

@@ -18,8 +18,8 @@
 
 import com.hivemq.cli.commands.options.DefaultOptions;
 import com.hivemq.cli.commands.options.DisconnectOptions;
-import com.hivemq.cli.mqtt.ClientKey;
-import com.hivemq.cli.mqtt.MqttClientExecutor;
+import com.hivemq.cli.mqtt.clients.CliMqttClient;
+import com.hivemq.cli.mqtt.clients.ShellClients;
 import com.hivemq.cli.utils.LoggerUtils;
 import org.jetbrains.annotations.NotNull;
 import org.tinylog.Logger;
@@ -29,36 +29,37 @@
 import java.util.concurrent.Callable;
 
 @CommandLine.Command(name = "dis", aliases = "disconnect", description = "Disconnects this MQTT client")
-public class ContextDisconnectCommand extends ShellContextCommand implements Callable<Integer> {
+public class ContextDisconnectCommand implements Callable<Integer> {
 
     @CommandLine.Mixin
     private final @NotNull DisconnectOptions disconnectOptions = new DisconnectOptions();
 
     @CommandLine.Mixin
     private final @NotNull DefaultOptions defaultOptions = new DefaultOptions();
 
+    private final @NotNull ShellClients shellClients;
+
     @Inject
-    public ContextDisconnectCommand(final @NotNull MqttClientExecutor executor) {
-        super(executor);
+    public ContextDisconnectCommand(final @NotNull ShellClients shellClients) {
+        this.shellClients = shellClients;
     }
 
     @Override
     public @NotNull Integer call() {
         Logger.trace("Command {} ", this);
 
-        if (contextClient != null) {
-            disconnectOptions.logUnusedDisconnectOptions(contextClient.getConfig().getMqttVersion());
+        final CliMqttClient client = shellClients.getContextClient();
+        if (client != null) {
+            disconnectOptions.logUnusedDisconnectOptions(client.getMqttVersion());
         }
 
         try {
             if (disconnectOptions.isDisconnectAll()) {
-                mqttClientExecutor.disconnectAllClients(disconnectOptions);
-            } else if (disconnectOptions.getClientIdentifier() != null && disconnectOptions.getHost() != null) {
-                final ClientKey clientKey =
-                        ClientKey.of(disconnectOptions.getClientIdentifier(), disconnectOptions.getHost());
-                mqttClientExecutor.disconnect(clientKey, disconnectOptions);
-            } else if (contextClient != null) {
-                mqttClientExecutor.disconnect(contextClient, disconnectOptions);
+                shellClients.disconnectAllClients(disconnectOptions);
+            } else if (disconnectOptions.getClientIdentifier() != null) {
+                shellClients.disconnect(disconnectOptions);
+            } else if (client != null) {
+                client.disconnect(disconnectOptions);
             }
         } catch (final Exception ex) {
             LoggerUtils.logShellError("Unable to disconnect", ex);

src/main/java/com/hivemq/cli/commands/shell/ContextExitCommand.java

@@ -16,24 +16,26 @@
 
 package com.hivemq.cli.commands.shell;
 
-import com.hivemq.cli.mqtt.MqttClientExecutor;
+import com.hivemq.cli.mqtt.clients.ShellClients;
 import org.jetbrains.annotations.NotNull;
 import picocli.CommandLine;
 
 import javax.inject.Inject;
 import java.util.concurrent.Callable;
 
 @CommandLine.Command(name = "exit", description = "Exit the current context", mixinStandardHelpOptions = true)
-public class ContextExitCommand extends ShellContextCommand implements Callable<Integer> {
+public class ContextExitCommand implements Callable<Integer> {
+
+    private final @NotNull ShellClients shellClients;
 
     @Inject
-    public ContextExitCommand(final @NotNull MqttClientExecutor mqttClientExecutor) {
-        super(mqttClientExecutor);
+    public ContextExitCommand(final @NotNull ShellClients shellClients) {
+        this.shellClients = shellClients;
     }
 
     @Override
     public @NotNull Integer call() {
-        removeContext();
+        shellClients.removeContextClient();
         return 0;
     }