Merge pull request #38 from boswelja/wearos-status-improvements

WearOS flow improvements

wearos/src/main/kotlin/com/boswelja/watchconnection/wearos/Helpers.kt

@@ -0,0 +1,24 @@
+package com.boswelja.watchconnection.wearos
+
+import kotlinx.coroutines.delay
+
+/**
+ * Schedule a suspendable code block to be executed at a regular interval.
+ * @param interval The interval in milliseconds to execute the code block after. Must be greater
+ * than 0.
+ * @param initialDelay The initial delay in milliseconds before first execution. Values less than or
+ * equal to 0 are ignored.
+ * @param action The suspendable code block to be executed.
+ */
+internal suspend fun repeating(
+    interval: Long,
+    initialDelay: Long = 0,
+    action: suspend () -> Unit
+) {
+    if (interval <= 0) throw IllegalArgumentException("interval must be greater than 0")
+    if (initialDelay > 0) delay(initialDelay)
+    while (true) {
+        action()
+        delay(interval)
+    }
+}

wearos/src/main/kotlin/com/boswelja/watchconnection/wearos/WearOSDiscoveryPlatform.kt

@@ -6,49 +6,51 @@ import com.boswelja.watchconnection.core.discovery.DiscoveryPlatform
 import com.boswelja.watchconnection.core.discovery.Status
 import com.boswelja.watchconnection.wearos.Constants.WEAROS_PLATFORM
 import com.google.android.gms.wearable.CapabilityClient
-import com.google.android.gms.wearable.CapabilityInfo
 import com.google.android.gms.wearable.NodeClient
 import com.google.android.gms.wearable.Wearable
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.channels.ProducerScope
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.tasks.await
 
 class WearOSDiscoveryPlatform(
     private val appCapability: String,
     private val capabilities: List<String>,
     private val nodeClient: NodeClient,
-    private val capabilityClient: CapabilityClient
+    private val capabilityClient: CapabilityClient,
+    private val scanRepeatInterval: Long = 2000
 ) : DiscoveryPlatform {
 
     constructor(
         context: Context,
         appCapability: String,
-        capabilities: List<String>
+        capabilities: List<String>,
+        scanRepeatInterval: Long
     ) : this(
         appCapability,
         capabilities,
         Wearable.getNodeClient(context),
-        Wearable.getCapabilityClient(context)
+        Wearable.getCapabilityClient(context),
+        scanRepeatInterval
     )
 
     override val platformIdentifier: String = WEAROS_PLATFORM
 
     override fun allWatches(): Flow<List<Watch>> = flow {
-        emit(
-            nodeClient.connectedNodes.await().map { node ->
-                Watch(
-                    node.displayName,
-                    node.id,
-                    platformIdentifier
-                )
-            }
-        )
+        repeating(interval = scanRepeatInterval) {
+            emit(
+                nodeClient.connectedNodes.await().map { node ->
+                    Watch(
+                        node.displayName,
+                        node.id,
+                        platformIdentifier
+                    )
+                }
+            )
+        }
     }
 
     @ExperimentalCoroutinesApi
@@ -88,66 +90,49 @@ class WearOSDiscoveryPlatform(
     }
 
     override fun getCapabilitiesFor(watchId: String): Flow<List<String>> = flow {
-        val discoveredCapabilities = mutableListOf<String>()
-        capabilities.forEach { capability ->
-            // Get capability info
-            val capabilityInfo = capabilityClient
-                .getCapability(capability, CapabilityClient.FILTER_ALL)
-                .await()
-            // If node is found with same ID as watch, emit capability
-            if (capabilityInfo.nodes.any { it.id == watchId })
-                discoveredCapabilities += capabilityInfo.name
+        repeating(interval = scanRepeatInterval) {
+            val discoveredCapabilities = mutableListOf<String>()
+            capabilities.forEach { capability ->
+                // Get capability info
+                val capabilityInfo = capabilityClient
+                    .getCapability(capability, CapabilityClient.FILTER_ALL)
+                    .await()
+                // If node is found with same ID as watch, emit capability
+                if (capabilityInfo.nodes.any { it.id == watchId })
+                    discoveredCapabilities += capabilityInfo.name
+            }
+            emit(discoveredCapabilities)
         }
-        emit(discoveredCapabilities)
     }
 
     @ExperimentalCoroutinesApi
-    override fun getStatusFor(watchId: String): Flow<Status> = callbackFlow {
+    override fun getStatusFor(watchId: String): Flow<Status> = flow {
         // Start with CONNECTING
-        send(Status.CONNECTING)
-
-        // Create our listener
-        val listener = CapabilityClient.OnCapabilityChangedListener { info: CapabilityInfo ->
-            getStatusFromCapabilityInfo(watchId, info)
-        }
-        // Add the listener
-        capabilityClient.addListener(listener, appCapability)
-
-        // Update capabilities now
-        val capabilityInfo = capabilityClient
-            .getCapability(appCapability, CapabilityClient.FILTER_ALL).await()
-        getStatusFromCapabilityInfo(watchId, capabilityInfo)
+        emit(Status.CONNECTING)
 
-        // On finish, remove the listener
-        awaitClose {
-            capabilityClient.removeListener(listener)
-        }
-    }
-
-    @ExperimentalCoroutinesApi
-    private fun ProducerScope<Status>.getStatusFromCapabilityInfo(
-        watchId: String,
-        info: CapabilityInfo
-    ) {
-        // If watch is found in capable nodes list, check if it's connected
-        if (info.nodes.any { it.id == watchId }) {
-            try {
-                // runBlocking should be safe here, since we're within a Flow
-                val connectedNodes = runBlocking { nodeClient.connectedNodes.await() }
-                // Got connected nodes, check if it contains our desired node
-                val node = connectedNodes.firstOrNull { it.id == watchId }
-                val status = node?.let {
-                    if (node.isNearby) Status.CONNECTED_NEARBY
-                    else Status.CONNECTED
-                } ?: Status.DISCONNECTED
-                trySend(status)
-            } catch (e: CancellationException) {
-                // Failed, send error
-                trySend(Status.ERROR)
+        // Get status at a set interval
+        repeating(interval = scanRepeatInterval) {
+            val capabilityInfo = capabilityClient
+                .getCapability(appCapability, CapabilityClient.FILTER_ALL).await()
+            if (capabilityInfo.nodes.any { it.id == watchId }) {
+                try {
+                    // runBlocking should be safe here, since we're within a Flow
+                    val connectedNodes = nodeClient.connectedNodes.await()
+                    // Got connected nodes, check if it contains our desired node
+                    val node = connectedNodes.firstOrNull { it.id == watchId }
+                    val status = node?.let {
+                        if (node.isNearby) Status.CONNECTED_NEARBY
+                        else Status.CONNECTED
+                    } ?: Status.DISCONNECTED
+                    emit(status)
+                } catch (e: CancellationException) {
+                    // Failed, send error
+                    emit(Status.ERROR)
+                }
+            } else {
+                // No watch in capable nodes, app is missing
+                emit(Status.MISSING_APP)
             }
-        } else {
-            // No watch in capable nodes, app is missing
-            trySend(Status.MISSING_APP)
         }
     }
 }