Merge pull request #33 from boswelja/incoming-messages-flow

Replace message listener with a Flow

build.gradle.kts

@@ -6,7 +6,7 @@ buildscript {
         mavenCentral()
     }
     dependencies {
-        classpath("com.android.tools.build:gradle:7.1.0-alpha01")
+        classpath("com.android.tools.build:gradle:7.1.0-alpha02")
         classpath("org.jetbrains.kotlin:kotlin-gradle-plugin:$kotlinVersion")
     }
 }

buildSrc/src/main/kotlin/Publishing.kt

@@ -12,9 +12,9 @@ object Publishing {
     val version: String?
         get() = System.getenv("VERSION")
 
-    private val ossrhUsername: String?
+    val ossrhUsername: String?
         get() = System.getenv("OSSRH_USERNAME")
-    private val ossrhPassword: String?
+    val ossrhPassword: String?
         get() = System.getenv("OSSRH_PASSWORD")
 
     const val groupId = "io.github.boswelja.watchconnection"

core/src/main/kotlin/com/boswelja/watchconnection/core/Message.kt

@@ -0,0 +1,38 @@
+package com.boswelja.watchconnection.core
+
+import java.util.UUID
+
+/**
+ * A data class containing information about a received message.
+ * @param sourceWatchId The [Watch.id] of the watch that sent the message.
+ * @param message The message itself.
+ * @param data Any data that may have been included with the message, or null if there is none.
+ */
+data class Message(
+    val sourceWatchId: UUID,
+    val message: String,
+    val data: ByteArray?
+) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+
+        other as Message
+
+        if (sourceWatchId != other.sourceWatchId) return false
+        if (message != other.message) return false
+        if (data != null) {
+            if (other.data == null) return false
+            if (!data.contentEquals(other.data)) return false
+        } else if (other.data != null) return false
+
+        return true
+    }
+
+    override fun hashCode(): Int {
+        var result = sourceWatchId.hashCode()
+        result = 31 * result + message.hashCode()
+        result = 31 * result + (data?.contentHashCode() ?: 0)
+        return result
+    }
+}

core/src/main/kotlin/com/boswelja/watchconnection/core/MessageReceiver.kt

@@ -18,15 +18,11 @@ abstract class MessageReceiver : BroadcastReceiver() {
     /**
      * Called when a message has been received. While this is a suspendable function, the limits
      * of [BroadcastReceiver.goAsync] still apply.
-     * @param sourceWatchId The [Watch.id] of the watch that sent the message.
-     * @param message The message that was received.
-     * @param data The data included with the message, or null if there was no data.
+     * @param message The [Message] that was received.
      */
     abstract suspend fun onMessageReceived(
         context: Context,
-        sourceWatchId: UUID,
-        message: String,
-        data: ByteArray?
+        message: Message
     )
 
     final override fun onReceive(context: Context?, intent: Intent?) {
@@ -43,7 +39,10 @@ abstract class MessageReceiver : BroadcastReceiver() {
                 val data = intent.getByteArrayExtra(EXTRA_DATA)
 
                 // Pass it on to user code
-                onMessageReceived(context, watchId, message, data)
+                onMessageReceived(
+                    context,
+                    Message(watchId, message, data)
+                )
 
                 // Let the BroadcastReceiver know we're done
                 pendingResult.finish()

core/src/main/kotlin/com/boswelja/watchconnection/core/WatchPlatform.kt

@@ -13,6 +13,12 @@ interface WatchPlatform {
      */
     val platformIdentifier: String
 
+    /**
+     * A [Flow] of [Message]s received by this platform. This should not emit anything unless there
+     * are collectors attached.
+     */
+    fun incomingMessages(): Flow<Message>
+
     /**
      * A flow of all available watches for this platform.
      */
@@ -50,17 +56,4 @@ interface WatchPlatform {
         message: String,
         data: ByteArray? = null
     ): Boolean
-
-    /**
-     * Adds a new [MessageListener].
-     * @param listener The [MessageListener] to register.
-     */
-    fun addMessageListener(listener: MessageListener)
-
-    /**
-     * Removes a [MessageListener]. This will do nothing if the provided listener is not
-     * registered.
-     * @param listener The [MessageListener] to unregister.
-     */
-    fun removeMessageListener(listener: MessageListener)
 }

core/src/main/kotlin/com/boswelja/watchconnection/core/WatchPlatformManager.kt

@@ -3,6 +3,7 @@ package com.boswelja.watchconnection.core
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.combine
+import kotlinx.coroutines.flow.merge
 
 /**
  * A class to simplify handling multiple [WatchPlatform].
@@ -49,6 +50,13 @@ class WatchPlatformManager(
             list
         }
 
+    /**
+     * A [Flow] of [Message]s received by all [WatchPlatform]s.
+     */
+    @ExperimentalCoroutinesApi
+    fun incomingMessages(): Flow<Message> =
+        connectionHandlers.values.map { it.incomingMessages() }.merge()
+
     /**
      * Send a message to a [Watch].
      * @param to The [Watch] to send the message to.
@@ -77,22 +85,4 @@ class WatchPlatformManager(
     fun getStatusFor(watch: Watch): Flow<Status>? {
         return connectionHandlers[watch.platform]?.getStatusFor(watch.platformId)
     }
-
-    /**
-     * Adds a [MessageListener] to all platforms.
-     */
-    fun addMessageListener(messageListener: MessageListener) {
-        connectionHandlers.values.forEach {
-            it.addMessageListener(messageListener)
-        }
-    }
-
-    /**
-     * Removes a [MessageListener] from all platforms.
-     */
-    fun removeMessageListener(messageListener: MessageListener) {
-        connectionHandlers.values.forEach {
-            it.removeMessageListener(messageListener)
-        }
-    }
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,4 +1,4 @@
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.0.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.1-rc-2-bin.zip
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStorePath=wrapper/dists

test/src/test/kotlin/com/boswelja/watchconnection/core/ConcreteMessageReceiver.kt

@@ -1,14 +1,11 @@
 package com.boswelja.watchconnection.core
 
 import android.content.Context
-import java.util.UUID
 
 class ConcreteMessageReceiver : MessageReceiver() {
 
     override suspend fun onMessageReceived(
         context: Context,
-        sourceWatchId: UUID,
-        message: String,
-        data: ByteArray?
+        message: Message
     ) { }
 }

test/src/test/kotlin/com/boswelja/watchconnection/core/DummyPlatform.kt

@@ -6,28 +6,22 @@ import kotlinx.coroutines.flow.flowOf
 
 class DummyPlatform(
     override val platformIdentifier: String,
-    val allWatches: Array<Watch>,
-    val watchesWithApp: Array<Watch>
+    val allWatches: List<Watch>,
+    val watchesWithApp: List<Watch>
 ) : WatchPlatform {
 
-    override fun allWatches(): Flow<Array<Watch>> = flowOf(allWatches)
+    override fun incomingMessages(): Flow<Message> = flow { }
 
-    override fun watchesWithApp(): Flow<Array<Watch>> = flowOf(watchesWithApp)
+    override fun allWatches(): Flow<List<Watch>> = flowOf(allWatches)
 
-    override fun getCapabilitiesFor(watchId: String): Flow<Array<String>> = flow { }
+    override fun watchesWithApp(): Flow<List<Watch>> = flowOf(watchesWithApp)
+
+    override fun getCapabilitiesFor(watchId: String): Flow<List<String>> = flow { }
 
     override fun getStatusFor(watchId: String): Flow<Status> = flow { }
 
     override suspend fun sendMessage(watchId: String, message: String, data: ByteArray?): Boolean {
         // Do nothing
         return false
     }
-
-    override fun addMessageListener(listener: MessageListener) {
-        // Do nothing
-    }
-
-    override fun removeMessageListener(listener: MessageListener) {
-        // Do nothing
-    }
 }

test/src/test/kotlin/com/boswelja/watchconnection/core/MessageReceiverTest.kt

@@ -43,27 +43,12 @@ class MessageReceiverTest {
         messageReceiver.onReceive(context, intent)
         verify(inverse = true, timeout = 50) { messageReceiver.goAsync() }
         coVerify(inverse = true, timeout = 50) {
-            messageReceiver.onMessageReceived(any(), any(), any(), any())
+            messageReceiver.onMessageReceived(any(), any())
         }
     }
 
     @Test
-    fun `onReceive passes variables to onMessageReceived if message has no data`() {
-        val id = UUID.randomUUID()
-        val message = "message"
-
-        Intent(ACTION_MESSAGE_RECEIVED).apply {
-            putExtra(EXTRA_WATCH_ID, id.toString())
-            putExtra(EXTRA_MESSAGE, message)
-        }.also { intent ->
-            messageReceiver.onReceive(context, intent)
-        }
-
-        coVerify(timeout = 50) { messageReceiver.onMessageReceived(context, id, message, null) }
-    }
-
-    @Test
-    fun `onReceive passes variables to onMessageReceived if message has data`() {
+    fun `onReceive passes variables to onMessageReceived`() {
         val id = UUID.randomUUID()
         val message = "message"
         val data = Random.nextBytes(10)
@@ -76,6 +61,7 @@ class MessageReceiverTest {
             messageReceiver.onReceive(context, intent)
         }
 
-        coVerify(timeout = 50) { messageReceiver.onMessageReceived(context, id, message, data) }
+        val expectedMessage = Message(id, message, data)
+        coVerify(timeout = 50) { messageReceiver.onMessageReceived(context, expectedMessage) }
     }
 }

test/src/test/kotlin/com/boswelja/watchconnection/core/WatchPlatformManagerTest.kt

@@ -5,7 +5,6 @@ import androidx.test.ext.junit.runners.AndroidJUnit4
 import io.mockk.coVerify
 import io.mockk.spyk
 import io.mockk.verify
-import java.util.UUID
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.runBlocking
 import org.junit.Before
@@ -37,7 +36,7 @@ class WatchPlatformManagerTest {
 
         dummyPlatformIds.forEach { platform ->
             // Generate dummy watches
-            var allWatches = emptyArray<Watch>()
+            val allWatches = mutableListOf<Watch>()
             (1..dummyWatchCountPerPlatform + 1).forEach { count ->
                 allWatches += Watch(
                     name = "Watch $count",
@@ -47,7 +46,7 @@ class WatchPlatformManagerTest {
             }
 
             // Generate dummy watches with app
-            var watchesWithApp = emptyArray<Watch>()
+            val watchesWithApp = mutableListOf<Watch>()
             (1..dummyWatchWithAppCountPerPlatform + 1).forEach { count ->
                 watchesWithApp += Watch(
                     name = "Watch $count",
@@ -102,34 +101,4 @@ class WatchPlatformManagerTest {
             coVerify { platform.getCapabilitiesFor(watch.platformId) }
         }
     }
-
-    @Test
-    fun `registerMessageListener adds the message listener to all platforms`() {
-        val messageListener = object : MessageListener {
-            override fun onMessageReceived(
-                sourceWatchId: UUID,
-                message: String,
-                data: ByteArray?
-            ) { }
-        }
-        platformManager.addMessageListener(messageListener)
-        dummyPlatforms.forEach { platform ->
-            verify { platform.addMessageListener(messageListener) }
-        }
-    }
-
-    @Test
-    fun `unregisterMessageListener removes the message listener from all platforms`() {
-        val messageListener = object : MessageListener {
-            override fun onMessageReceived(
-                sourceWatchId: UUID,
-                message: String,
-                data: ByteArray?
-            ) { }
-        }
-        platformManager.removeMessageListener(messageListener)
-        dummyPlatforms.forEach { platform ->
-            verify { platform.removeMessageListener(messageListener) }
-        }
-    }
 }

test/src/test/kotlin/com/boswelja/watchconnection/tizen/TizenPlatformTest.kt

@@ -4,15 +4,13 @@ import android.os.Build
 import androidx.arch.core.executor.testing.InstantTaskExecutorRule
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
-import com.boswelja.watchconnection.core.MessageListener
 import com.samsung.android.sdk.accessory.SAAgentV2
 import io.mockk.MockKAnnotations
 import io.mockk.coVerify
 import io.mockk.every
 import io.mockk.impl.annotations.RelaxedMockK
 import io.mockk.mockkStatic
 import io.mockk.verify
-import java.util.UUID
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.runBlocking
@@ -126,30 +124,4 @@ class TizenPlatformTest {
         connectionHandler.sendMessage(id, message, data)
         coVerify { accessoryAgent.sendMessage(id, message, data) }
     }
-
-    @Test
-    fun `registerMessageListener calls agent`() {
-        val listener = object : MessageListener {
-            override fun onMessageReceived(
-                sourceWatchId: UUID,
-                message: String,
-                data: ByteArray?
-            ) { }
-        }
-        connectionHandler.addMessageListener(listener)
-        verify { accessoryAgent.registerMessageListener(listener) }
-    }
-
-    @Test
-    fun `unregisterMessageListener calls agent`() {
-        val listener = object : MessageListener {
-            override fun onMessageReceived(
-                sourceWatchId: UUID,
-                message: String,
-                data: ByteArray?
-            ) { }
-        }
-        connectionHandler.removeMessageListener(listener)
-        verify { accessoryAgent.unregisterMessageListener(listener) }
-    }
 }