Fix/mqtt connection reliability | Android 11 / Nvidia Shield

app/src/main/java/be/digitalia/mediasession2mqtt/MainWorker.kt

@@ -1,14 +1,15 @@
 package be.digitalia.mediasession2mqtt
 
 import android.content.Context
+import android.util.Log
 import be.digitalia.mediasession2mqtt.homeassistant.Sensor
 import be.digitalia.mediasession2mqtt.homeassistant.createSensorDiscoveryConfiguration
 import be.digitalia.mediasession2mqtt.mediasession.CurrentMediaControllerDetector
 import be.digitalia.mediasession2mqtt.mediasession.metadataFlow
 import be.digitalia.mediasession2mqtt.mediasession.playbackStateFlow
 import be.digitalia.mediasession2mqtt.mqtt.MQTTPublishClient
 import be.digitalia.mediasession2mqtt.mqtt.MQTTQoSLevel
-import be.digitalia.mediasession2mqtt.mqtt.tryConnectAndPublish
+import be.digitalia.mediasession2mqtt.mqtt.publishWithRetry
 import be.digitalia.mediasession2mqtt.mqttmediaplayer.MQTTMediaMetadata
 import be.digitalia.mediasession2mqtt.mqttmediaplayer.MQTTPlaybackState
 import be.digitalia.mediasession2mqtt.mqttmediaplayer.toMQTTPlaybackStateOrNull
@@ -17,6 +18,7 @@ import be.digitalia.mediasession2mqtt.mqttmediaplayer.toMediaTitle
 import be.digitalia.mediasession2mqtt.service.MediaSessionListenerService
 import be.digitalia.mediasession2mqtt.settings.SettingsProvider
 import dev.zacsweers.metro.Inject
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -25,10 +27,10 @@ import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.collectLatest
+import kotlinx.coroutines.flow.conflate
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.flatMapLatest
 import kotlinx.coroutines.flow.flowOf
-import kotlinx.coroutines.flow.fold
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.mapNotNull
 import kotlinx.coroutines.launch
@@ -96,15 +98,15 @@ class MainWorker(
         qosLevel: MQTTQoSLevel,
         deviceId: Int
     ) {
-        settingsProvider.isHassIntegrationEnabled.collect { isEnabled ->
+        settingsProvider.isHassIntegrationEnabled.collectLatest { isEnabled ->
             if (isEnabled) {
                 for (sensor in HASS_SENSORS) {
                     val discoveryConfig = createSensorDiscoveryConfiguration(
                         deviceId = deviceId,
                         sensor = sensor,
                         sensorTopic = "$ROOT_TOPIC/$deviceId/${sensor.subTopic}"
                     )
-                    client.tryConnectAndPublish(
+                    client.publishWithRetry(
                         qosLevel,
                         "$HASS_ROOT_TOPIC/${sensor.type}/${sensor.getUniqueId(deviceId)}/config",
                         discoveryConfig
@@ -119,8 +121,8 @@ class MainWorker(
         qosLevel: MQTTQoSLevel,
         deviceId: Int
     ) {
-        applicationIdFlow.collect { applicationId ->
-            client.tryConnectAndPublish(
+        applicationIdFlow.collectLatest { applicationId ->
+            client.publishWithRetry(
                 qosLevel,
                 "$ROOT_TOPIC/$deviceId/$APPLICATION_ID_SUB_TOPIC",
                 applicationId
@@ -133,24 +135,30 @@ class MainWorker(
         qosLevel: MQTTQoSLevel,
         deviceId: Int
     ) {
-        playbackStateFlow.fold(null as MQTTPlaybackState?) { previousPlaybackState, playbackState ->
-            val name = playbackState.name
-            if (previousPlaybackState?.name != name) {
-                client.tryConnectAndPublish(
-                    qosLevel,
-                    "$ROOT_TOPIC/$deviceId/$PLAYBACK_STATE_SUB_TOPIC",
-                    name
-                )
+        coroutineScope {
+            launch {
+                playbackStateFlow.map { it.name }.distinctUntilChanged().collectLatest { name ->
+                    client.publishWithRetry(
+                        qosLevel,
+                        "$ROOT_TOPIC/$deviceId/$PLAYBACK_STATE_SUB_TOPIC",
+                        name
+                    )
+                }
             }
-            val positionInMillis = playbackState.positionInMillis
-            if (previousPlaybackState?.positionInMillis != positionInMillis) {
-                client.tryConnectAndPublish(
-                    qosLevel,
-                    "$ROOT_TOPIC/$deviceId/$PLAYBACK_POSITION_SUB_TOPIC",
-                    positionInMillis
-                )
+            launch {
+                // Some players (e.g. Emby) update the playback position every second.
+                // Rate-limit the publications to avoid flooding the broker with retained
+                // messages: conflate() keeps only the latest position while waiting
+                playbackStateFlow.map { it.positionInMillis }.distinctUntilChanged().conflate()
+                    .collect { positionInMillis ->
+                        client.publishWithRetry(
+                            qosLevel,
+                            "$ROOT_TOPIC/$deviceId/$PLAYBACK_POSITION_SUB_TOPIC",
+                            positionInMillis
+                        )
+                        delay(POSITION_PUBLISH_MIN_INTERVAL_MILLIS)
+                    }
             }
-            playbackState
         }
     }
 
@@ -159,33 +167,38 @@ class MainWorker(
         qosLevel: MQTTQoSLevel,
         deviceId: Int
     ) {
-        mediaMetadataFlow.fold(null as MQTTMediaMetadata?) { previousMediaMetadata, mediaMetadata ->
-            val title = mediaMetadata.title
-            if (previousMediaMetadata?.title != title) {
-                client.tryConnectAndPublish(
-                    qosLevel,
-                    "$ROOT_TOPIC/$deviceId/$MEDIA_TITLE_SUB_TOPIC",
-                    title
-                )
+        coroutineScope {
+            launch {
+                mediaMetadataFlow.map { it.title }.distinctUntilChanged().collectLatest { title ->
+                    client.publishWithRetry(
+                        qosLevel,
+                        "$ROOT_TOPIC/$deviceId/$MEDIA_TITLE_SUB_TOPIC",
+                        title
+                    )
+                }
             }
-            val durationInMillis = mediaMetadata.durationInMillis
-            if (previousMediaMetadata?.durationInMillis != durationInMillis) {
-                client.tryConnectAndPublish(
-                    qosLevel,
-                    "$ROOT_TOPIC/$deviceId/$MEDIA_DURATION_SUB_TOPIC",
-                    durationInMillis
-                )
+            launch {
+                mediaMetadataFlow.map { it.durationInMillis }.distinctUntilChanged()
+                    .collectLatest { durationInMillis ->
+                        client.publishWithRetry(
+                            qosLevel,
+                            "$ROOT_TOPIC/$deviceId/$MEDIA_DURATION_SUB_TOPIC",
+                            durationInMillis
+                        )
+                    }
             }
-            mediaMetadata
         }
     }
 
     fun start() {
-        coroutineScope.launch {
+        // Both loops are restarted after an unexpected failure: an exception escaping from a
+        // media session flow (e.g. when notification access is revoked) or the MQTT client must
+        // not permanently stop the publishing pipeline while the process keeps running
+        coroutineScope.launchSupervised("monitorSettings") {
             monitorSettings()
         }
         // Watchdog to attempt rebinding MediaSessionListenerService when disconnected
-        coroutineScope.launch {
+        coroutineScope.launchSupervised("listenerWatchdog") {
             currentMediaControllerDetector.isListening.collectLatest { isListening ->
                 if (!isListening) {
                     delay(AUTO_REBIND_SERVICE_DELAY_MILLIS)
@@ -195,8 +208,27 @@ class MainWorker(
         }
     }
 
+    private fun CoroutineScope.launchSupervised(name: String, block: suspend () -> Unit) {
+        launch {
+            while (true) {
+                try {
+                    block()
+                } catch (e: CancellationException) {
+                    throw e
+                } catch (e: Throwable) {
+                    Log.e(TAG, "$name failed, restarting in ${RESTART_DELAY_MILLIS}ms", e)
+                }
+                delay(RESTART_DELAY_MILLIS)
+            }
+        }
+    }
+
     companion object {
+        private const val TAG = "MediaSession2MQTT"
+
         private const val AUTO_REBIND_SERVICE_DELAY_MILLIS = 2000L
+        private const val RESTART_DELAY_MILLIS = 5000L
+        private const val POSITION_PUBLISH_MIN_INTERVAL_MILLIS = 5000L
 
         private const val ROOT_TOPIC = "mediaSession"
         private const val APPLICATION_ID_SUB_TOPIC = "applicationId"

app/src/main/java/be/digitalia/mediasession2mqtt/mqtt/KMQTTClient.kt

@@ -4,6 +4,7 @@ import io.github.davidepianca98.MQTTClient
 import io.github.davidepianca98.mqtt.MQTTVersion
 import io.github.davidepianca98.mqtt.packets.Qos
 import io.github.davidepianca98.mqtt.packets.mqttv5.ReasonCode
+import java.io.IOException
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.NonCancellable
@@ -20,9 +21,15 @@ class KMQTTClient(
 
     private fun getConnectedClient(forceNewInstance: Boolean): MQTTClient {
         // Create the client lazily (simple implementation for single thread)
-        val client = currentClient.takeUnless { forceNewInstance }
+        var client = currentClient.takeUnless { forceNewInstance }
             ?: createClient().also { currentClient = it }
         client.step()
+        if (!client.isRunning()) {
+            // A stopped client silently ignores step() and drops published messages,
+            // so it must be detected and replaced with a new connected instance
+            client = createClient().also { currentClient = it }
+            client.step()
+        }
         return client
     }
 
@@ -39,7 +46,7 @@ class KMQTTClient(
             address = connectionSettings.hostname,
             port = connectionSettings.port,
             tls = null,
-            keepAlive = 0,
+            keepAlive = KEEP_ALIVE_SECONDS,
             webSocket = null,
             userName = username,
             password = password
@@ -54,24 +61,34 @@ class KMQTTClient(
 
     override suspend fun connectAndPublish(qosLevel: MQTTQoSLevel, topic: String, payload: String) {
         withContext(dispatcher) {
-            val client = try {
-                getConnectedClient(false)
+            try {
+                publishAndStep(getConnectedClient(false), qosLevel, topic, payload)
             } catch (e: Exception) {
                 if (e is CancellationException) {
                     throw e
                 }
                 // At that point we are already disconnected, no need to call disconnect()
-                // Try to auto-reconnect from scratch
-                getConnectedClient(true)
+                // The current connection may also be half-open after a silent network loss:
+                // in all cases, retry once from scratch with a new connection
+                ensureActive()
+                publishAndStep(getConnectedClient(true), qosLevel, topic, payload)
             }
-            ensureActive()
-            client.publish(
-                true,
-                Qos.entries[qosLevel.ordinal],
-                topic,
-                payload.encodeToByteArray().toUByteArray()
-            )
-            client.step()
+        }
+    }
+
+    private fun publishAndStep(client: MQTTClient, qosLevel: MQTTQoSLevel, topic: String, payload: String) {
+        client.publish(
+            true,
+            Qos.entries[qosLevel.ordinal],
+            topic,
+            payload.encodeToByteArray().toUByteArray()
+        )
+        client.step()
+        if (!client.isRunning()) {
+            // The connection died during the publish and the message may have been dropped:
+            // report the failure so the caller can retry
+            currentClient = null
+            throw IOException("MQTT connection lost while publishing")
         }
     }
 
@@ -95,4 +112,10 @@ class KMQTTClient(
             return KMQTTClient(connectionSettings, dispatcher)
         }
     }
+
+    companion object {
+        // Advertise a keep alive interval so the broker eventually drops dead connections
+        // instead of keeping half-open sessions alive forever
+        private const val KEEP_ALIVE_SECONDS = 60
+    }
 }

app/src/main/java/be/digitalia/mediasession2mqtt/mqtt/MQTTPublishClientExt.kt

@@ -1,6 +1,7 @@
 package be.digitalia.mediasession2mqtt.mqtt
 
 import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.delay
 
 suspend fun MQTTPublishClient.testConnection() {
     try {
@@ -27,4 +28,24 @@ suspend fun MQTTPublishClient.tryConnectAndPublish(
         }
         false
     }
-}
+}
+
+/**
+ * Publish and keep retrying with increasing delays until it succeeds or the coroutine is cancelled.
+ * Call from collectLatest so that a newer value cancels the pending retries of an older one.
+ */
+suspend fun MQTTPublishClient.publishWithRetry(
+    qosLevel: MQTTQoSLevel,
+    topic: String,
+    payload: String
+) {
+    var attemptIndex = 0
+    while (!tryConnectAndPublish(qosLevel, topic, payload)) {
+        delay(RETRY_DELAYS_MILLIS[attemptIndex])
+        if (attemptIndex < RETRY_DELAYS_MILLIS.lastIndex) {
+            attemptIndex++
+        }
+    }
+}
+
+private val RETRY_DELAYS_MILLIS = longArrayOf(1_000L, 2_000L, 5_000L, 15_000L, 30_000L)