feat: add callback when mqtt connection is lost (#107)

open200 · Aug 12, 2024 · 9661e8b · 9661e8b
1 parent c021a68
commit 9661e8b
Show file tree

Hide file tree

Showing 3 changed files with 54 additions and 2 deletions.
diff --git a/xesar-connect/src/main/kotlin/com/open200/xesar/connect/XesarConnect.kt b/xesar-connect/src/main/kotlin/com/open200/xesar/connect/XesarConnect.kt
@@ -34,6 +34,9 @@ class XesarConnect(private val client: IXesarMqttClient, val config: Config) {
     private val coroutineScopeForSendCommand =
         CoroutineScope(config.dispatcherForCommandsAndCleanUp)
     private val coroutineScopeForCleanUp = CoroutineScope(config.dispatcherForCommandsAndCleanUp)
+
+    /** Callback for when the mqtt connection is lost. */
+    var onConnectionLost: ((ConnectionFailedException) -> Unit)? = null
     lateinit var token: Token
 
     /**
@@ -48,7 +51,12 @@ class XesarConnect(private val client: IXesarMqttClient, val config: Config) {
     }
 
     init {
-        client.onDisconnect = { connectionChannel.trySend(ConnectionEvent.DISCONNECTED) }
+        client.onDisconnect = { ex ->
+            connectionChannel.trySend(ConnectionEvent.DISCONNECTED)
+            if (ex != null) {
+                onConnectionLost?.let { it(ex) }
+            }
+        }
         client.onMessage = { topic, message ->
             // call all listeners on the topic
             val decodedMessage = message.decodeToString()

diff --git a/xesar-connect/src/main/kotlin/com/open200/xesar/connect/XesarMqttClient.kt b/xesar-connect/src/main/kotlin/com/open200/xesar/connect/XesarMqttClient.kt
@@ -25,7 +25,9 @@ class XesarMqttClient(private val client: MqttAsyncClient) : IXesarMqttClient {
 
                 override fun connectionLost(cause: Throwable?) {
                     log.error("lost connection: $cause")
-                    val exception = ConnectionFailedException("lost connection: $cause")
+                    val exception =
+                        cause?.let { ConnectionFailedException("lost connection", it) }
+                            ?: ConnectionFailedException("lost connection")
                     onDisconnect(exception)
                 }
 

diff --git a/xesar-connect/src/test/kotlin/com/open200/xesar/connect/it/MqttConnectionLostTest.kt b/xesar-connect/src/test/kotlin/com/open200/xesar/connect/it/MqttConnectionLostTest.kt
@@ -0,0 +1,42 @@
+package com.open200.xesar.connect.it
+
+import com.open200.xesar.connect.XesarConnect
+import com.open200.xesar.connect.XesarMqttClient
+import io.kotest.common.runBlocking
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.extensions.testcontainers.perTest
+import java.util.*
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeout
+
+class MqttConnectionLostTest :
+    FunSpec({
+        val container = MosquittoContainer.container()
+        val config = MosquittoContainer.config(container)
+        listener(container.perTest())
+
+        test("connect with custom mqtt client id") {
+            runBlocking {
+                val connected = CompletableDeferred<Unit>()
+                val connectionLost = CompletableDeferred<Unit>()
+                launch {
+                    val xesarMqttClient = XesarMqttClient.connectAsync(config).await()
+                    connected.complete(Unit)
+
+                    val xesarConnect = XesarConnect(xesarMqttClient, config)
+                    xesarConnect.onConnectionLost = { cause -> connectionLost.complete(Unit) }
+                    try {
+                        xesarConnect.delay()
+                    } catch (e: Exception) {
+                        // ignore
+                    }
+                }
+                launch {
+                    withTimeout(1000) { connected.await() }
+                    container.stop()
+                    withTimeout(1000) { connectionLost.await() }
+                }
+            }
+        }
+    })