use coroutines

build.gradle

@@ -52,6 +52,7 @@ repositories {
 
 dependencies {
     implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk8:1.4.10"
+    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.0-M1'
     implementation 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
 }
 

src/main/kotlin/MQTTSubscriber.kt

@@ -6,11 +6,12 @@ class MqttSubscriber : MqttCallbackExtended {
     private val broker = "tcp://mqtt.k8s.kmlabz.com:1883"
     private val clientId = MqttClient.generateClientId()
     private val persistence = MemoryPersistence()
-    private val mqttClient = MqttClient(broker, clientId, persistence)
+    private val mqttClient = MqttAsyncClient(broker, clientId, persistence)
     private val benchValues = HashMap<Int, Instant>()
 
     override fun connectionLost(arg0: Throwable) {
         System.err.println("connection lost")
+        mqttClient.reconnect()
     }
 
     override fun deliveryComplete(arg0: IMqttDeliveryToken) {
@@ -21,7 +22,6 @@ class MqttSubscriber : MqttCallbackExtended {
         println("topic: $topic")
         benchValues.put(topic.split("/")[1].toInt(),Instant.now())
         println("message: " + String(message.payload))
-
     }
 
     override fun connectComplete(reconnect: Boolean, serverURI: String?) {

src/main/kotlin/Main.kt

@@ -1,6 +1,12 @@
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+
 fun main(){
-    val mqtt = MqttSubscriber()
-    mqtt.connect()
-    Thread.sleep(10000)
-    mqtt.disconnect()
+    GlobalScope.launch {
+        val mqtt = MqttSubscriber()
+        mqtt.connect()
+        delay(10000L)
+        mqtt.disconnect()
+    }
 }