Add an MQTT client session, which allows to automatically re-connect.

.github/workflows/ci.yml

@@ -17,6 +17,13 @@ jobs:
       matrix:
         os: [ubuntu-latest]
         jdk: [8, 11]
+        include:
+          - os: ubuntu-latest
+            jdk: 11
+            jdk_release: latest
+          - os: ubuntu-latest
+            jdk: 8
+            jdk_release: "jdk8u282-b08" # required due to https://bugs.openjdk.java.net/browse/JDK-8266279
     runs-on: ${{ matrix.os }}
     steps:
       - name: Checkout
@@ -25,6 +32,7 @@ jobs:
         uses: joschi/setup-jdk@v2
         with:
           java-version: ${{ matrix.jdk }}
+          release: ${{ matrix.jdk_release }}
       - name: Run tests
         run: mvn -q clean verify -B
   Deploy:

pom.xml

@@ -18,6 +18,12 @@
       <organization>Red Hat</organization>
       <organizationUrl>http://www.redhat.com</organizationUrl>
     </developer>
+    <developer>
+      <name>Jens Reimann</name>
+      <email>[email protected]</email>
+      <organization>Red Hat</organization>
+      <organizationUrl>https://www.redhat.com</organizationUrl>
+    </developer>
   </developers>
 
   <properties>

src/main/generated/io/vertx/mqtt/MqttClientSessionOptionsConverter.java

@@ -0,0 +1,43 @@
+package io.vertx.mqtt;
+
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.impl.JsonUtil;
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * Converter and mapper for {@link io.vertx.mqtt.MqttClientSessionOptions}.
+ * NOTE: This class has been automatically generated from the {@link io.vertx.mqtt.MqttClientSessionOptions} original class using Vert.x codegen.
+ */
+public class MqttClientSessionOptionsConverter {
+
+
+  public static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, MqttClientSessionOptions obj) {
+    for (java.util.Map.Entry<String, Object> member : json) {
+      switch (member.getKey()) {
+        case "hostname":
+          if (member.getValue() instanceof String) {
+            obj.setHostname((String)member.getValue());
+          }
+          break;
+        case "port":
+          if (member.getValue() instanceof Number) {
+            obj.setPort(((Number)member.getValue()).intValue());
+          }
+          break;
+      }
+    }
+  }
+
+  public static void toJson(MqttClientSessionOptions obj, JsonObject json) {
+    toJson(obj, json.getMap());
+  }
+
+  public static void toJson(MqttClientSessionOptions obj, java.util.Map<String, Object> json) {
+    if (obj.getHostname() != null) {
+      json.put("hostname", obj.getHostname());
+    }
+    json.put("port", obj.getPort());
+  }
+}

src/main/generated/io/vertx/mqtt/reconnect/ConstantReconnectDelayOptionsConverter.java

@@ -0,0 +1,29 @@
+package io.vertx.mqtt.reconnect;
+
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.impl.JsonUtil;
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * Converter and mapper for {@link io.vertx.mqtt.reconnect.ConstantReconnectDelayOptions}.
+ * NOTE: This class has been automatically generated from the {@link io.vertx.mqtt.reconnect.ConstantReconnectDelayOptions} original class using Vert.x codegen.
+ */
+public class ConstantReconnectDelayOptionsConverter {
+
+
+  public static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, ConstantReconnectDelayOptions obj) {
+    for (java.util.Map.Entry<String, Object> member : json) {
+      switch (member.getKey()) {
+      }
+    }
+  }
+
+  public static void toJson(ConstantReconnectDelayOptions obj, JsonObject json) {
+    toJson(obj, json.getMap());
+  }
+
+  public static void toJson(ConstantReconnectDelayOptions obj, java.util.Map<String, Object> json) {
+  }
+}

src/main/generated/io/vertx/mqtt/reconnect/ExponentialBackoffDelayOptionsConverter.java

@@ -0,0 +1,29 @@
+package io.vertx.mqtt.reconnect;
+
+import io.vertx.core.json.JsonObject;
+import io.vertx.core.json.JsonArray;
+import io.vertx.core.json.impl.JsonUtil;
+import java.time.Instant;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * Converter and mapper for {@link io.vertx.mqtt.reconnect.ExponentialBackoffDelayOptions}.
+ * NOTE: This class has been automatically generated from the {@link io.vertx.mqtt.reconnect.ExponentialBackoffDelayOptions} original class using Vert.x codegen.
+ */
+public class ExponentialBackoffDelayOptionsConverter {
+
+
+  public static void fromJson(Iterable<java.util.Map.Entry<String, Object>> json, ExponentialBackoffDelayOptions obj) {
+    for (java.util.Map.Entry<String, Object> member : json) {
+      switch (member.getKey()) {
+      }
+    }
+  }
+
+  public static void toJson(ExponentialBackoffDelayOptions obj, JsonObject json) {
+    toJson(obj, json.getMap());
+  }
+
+  public static void toJson(ExponentialBackoffDelayOptions obj, java.util.Map<String, Object> json) {
+  }
+}

src/main/java/examples/VertxMqttClientSessionExample.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2021 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package examples;
+
+import io.vertx.core.Vertx;
+import io.vertx.mqtt.MqttClientSession;
+import io.vertx.mqtt.MqttClientSessionOptions;
+
+public class VertxMqttClientSessionExample {
+
+  public static void main(String[] args) {
+    Vertx vertx = Vertx.vertx();
+
+    MqttClientSessionOptions options = new MqttClientSessionOptions();
+
+    MqttClientSession session = MqttClientSession.create(vertx, options)
+      .sessionStateHandler(state -> System.out.format("State changed - state: %s, cause: %s%n", state.getSessionState(), state.getCause()))
+      .subscriptionStateHandler(state -> System.out.format("Subscription changed [%s] - state: %s, QoS: %s%n", state.getTopic(), state.getSubscriptionState(), state.getQos()))
+      .messageHandler(message -> System.out.format("Message received: %s%n", message))
+      .subscribe(MqttClientSession.RequestedQoS.QOS_1, "foo", "bar", "baz/#");
+
+    session.start();
+
+  }
+
+}

src/main/java/io/vertx/mqtt/MqttClient.java

@@ -29,6 +29,7 @@
 import io.vertx.mqtt.messages.MqttPublishMessage;
 import io.vertx.mqtt.messages.MqttSubAckMessage;
 
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -253,6 +254,14 @@ static MqttClient create(Vertx vertx) {
    */
   Future<Integer> unsubscribe(String topic);
 
+  /**
+   * Unsubscribe from receiving messages on given topics
+   *
+   * @param topics Topics you want to unsubscribe from
+   * @return a {@code Future} completed after UNSUBSCRIBE packet sent with packetid
+   */
+  Future<Integer> unsubscribe(List<String> topics);
+
   /**
    * Unsubscribe from receiving messages on given topic
    *
@@ -263,6 +272,16 @@ static MqttClient create(Vertx vertx) {
   @Fluent
   MqttClient unsubscribe(String topic, Handler<AsyncResult<Integer>> unsubscribeSentHandler);
 
+  /**
+   * Unsubscribe from receiving messages on given topics
+   *
+   * @param topics Topics you want to unsubscribe from
+   * @param unsubscribeSentHandler  handler called after UNSUBSCRIBE packet sent
+   * @return current MQTT client instance
+   */
+  @Fluent
+  MqttClient unsubscribe(List<String> topics, Handler<AsyncResult<Integer>> unsubscribeSentHandler);
+
   /**
    * Sets handler which will be called after PINGRESP packet receiving
    *