Added simple flowemitter test with relevant comments for Emitter

chat-android/src/main/java/com/ably/chat/FlowEmitter.kt

@@ -8,12 +8,18 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.takeWhile
 import kotlinx.coroutines.launch
 
+/**
+ * Interface implementation should work in both java and kotlin
+ */
 interface Emitter<V> {
     fun emit(value: V)
     fun on(block: suspend CoroutineScope.(V) -> Unit): Subscription
     fun offAll()
 }
 
+/**
+ * Interface implementation should work in both java and kotlin
+ */
 interface EventEmitter<K, V> {
     fun emit(event: K, value: V)
     fun on(event: K, block: suspend CoroutineScope.(V) -> Unit): Subscription
@@ -39,6 +45,9 @@ open class FlowEmitter<V>(scope: CoroutineScope = CoroutineScope(Dispatchers.Def
             }
         }
         return Subscription {
+            /**
+             * Seems only sensible way to cancel collector, since job.cancel cancels ongoing job
+             */
             keepCollecting.set(false)
         }
     }

chat-android/src/test/java/com/ably/chat/FlowEmitterTest.kt

@@ -0,0 +1,34 @@
+package com.ably.chat
+
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert
+import org.junit.Test
+
+class FlowEmitterTest {
+    @Test
+    fun `should be able to emit and listen to event`() = runTest {
+        val flowEmitter = FlowEmitter<String>()
+        val receivedValues = mutableListOf<String>()
+
+        flowEmitter.emit("1")
+
+        val subscription = flowEmitter.on { received: String ->
+            receivedValues.add(received)
+        }
+
+        flowEmitter.emit("2")
+        flowEmitter.emit("3")
+        flowEmitter.emit("4")
+
+        subscription.unsubscribe()
+
+        flowEmitter.emit("5")
+        flowEmitter.emit("7")
+
+        assertWaiter { receivedValues.size == 3 }.join()
+
+        // Assertion fails because receivedValues returns empty values
+        // Seems collector works independently of emitter and can be late processing values
+        Assert.assertEquals(listOf("2", "3", "4"), receivedValues)
+    }
+}

chat-android/src/test/java/com/ably/chat/TestUtils.kt

@@ -4,6 +4,12 @@ import com.google.gson.JsonElement
 import io.ably.lib.types.AsyncHttpPaginatedResponse
 import io.mockk.every
 import io.mockk.mockk
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withTimeout
 
 fun buildAsyncHttpPaginatedResponse(items: List<JsonElement>): AsyncHttpPaginatedResponse {
     val response = mockk<AsyncHttpPaginatedResponse>()
@@ -49,3 +55,16 @@ fun mockOccupancyApiResponse(realtimeClientMock: RealtimeClient, response: JsonE
         )
     }
 }
+
+suspend fun assertWaiter(timeoutInMs: Long = 10000, block : () -> Boolean): Job {
+    // Need to create coroutineScope because delay doesn't work in runTest default CoroutineScope
+    val scope = CoroutineScope(Dispatchers.Default)
+    return scope.launch {
+        withTimeout(timeoutInMs) {
+            do {
+                val success = block()
+                delay(100)
+            } while (!success)
+        }
+    }
+}