Switch remaining primary observables

1 files changed, 68 insertions, 4 deletions

diff --git a/app/src/test/kotlin/com/pitchedapps/frost/views/FrostContentViewAsyncTest.kt b/app/src/test/kotlin/com/pitchedapps/frost/utils/CoroutineTest.kt
index 13c40aa3..f930e529 100644
--- a/app/src/test/kotlin/com/pitchedapps/frost/views/FrostContentViewAsyncTest.kt
+++ b/app/src/test/kotlin/com/pitchedapps/frost/utils/CoroutineTest.kt
@@ -1,12 +1,14 @@
-package com.pitchedapps.frost.views
+package com.pitchedapps.frost.utils
 
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.BroadcastChannel
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.count
+import kotlinx.coroutines.delay
 import kotlinx.coroutines.joinAll
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
@@ -19,10 +21,10 @@ import kotlin.test.assertFalse
 import kotlin.test.assertTrue
 
 /**
- * Collection of tests around the view thread logic
+ * Collection of tests around coroutines
  */
 @UseExperimental(ExperimentalCoroutinesApi::class)
-class FrostContentViewAsyncTest {
+class CoroutineTest {
 
     /**
      * Hooks onto the refresh channel for one true -> false cycle.
@@ -39,7 +41,7 @@ class FrostContentViewAsyncTest {
         }
     }
 
-    private suspend fun <T> listen(channel: ReceiveChannel<T>, shouldEnd: (T) -> Boolean = { false }): List<T> =
+    private suspend fun <T> listen(channel: ReceiveChannel<T>, shouldEnd: suspend (T) -> Boolean = { false }): List<T> =
         withContext(Dispatchers.IO) {
             val data = mutableListOf<T>()
             for (c in channel) {
@@ -127,4 +129,66 @@ class FrostContentViewAsyncTest {
             assertEquals(4, receiver2.count(), "Not all events received")
         }
     }
+
+    /**
+     * Not a true throttle, but for things like fetching header badges, we want to avoid simultaneous fetches.
+     * As a result, I want to test that the usage of offer along with a rendezvous channel will work as I expect.
+     * Events should be consumed when there is no pending consumer on previous elements.
+     */
+    @Test
+    fun throttledChannel() {
+        val channel = Channel<Int>(Channel.RENDEZVOUS)
+        runBlocking {
+            val deferred = async {
+                listen(channel) {
+                    // Throttle consumer
+                    delay(10)
+                    return@listen false
+                }
+            }
+            (0..100).forEach {
+                channel.offer(it)
+                delay(1)
+            }
+            channel.close()
+            val received = deferred.await()
+            assertTrue(
+                received.size < 20,
+                "Received data should be throttled; expected that around 1/10th of all events are consumed"
+            )
+            println(received)
+        }
+    }
+
+    @Test
+    fun uniqueOnly() {
+        val channel = BroadcastChannel<Int>(100)
+        runBlocking {
+            val fullReceiver = channel.openSubscription()
+            val uniqueReceiver = channel.openSubscription().uniqueOnly(this)
+
+            val fullDeferred = async { listen(fullReceiver) }
+            val uniqueDeferred = async { listen(uniqueReceiver) }
+
+            listOf(0, 1, 2, 3, 3, 3, 4, 3, 5, 5, 1).forEach {
+                channel.offer(it)
+            }
+            channel.close()
+
+            val fullData = fullDeferred.await()
+            val uniqueData = uniqueDeferred.await()
+
+            assertEquals(
+                listOf(0, 1, 2, 3, 3, 3, 4, 3, 5, 5, 1),
+                fullData,
+                "Full receiver should get all channel events"
+            )
+            assertEquals(
+                listOf(0, 1, 2, 3, 4, 3, 5, 1),
+                uniqueData,
+                "Unique receiver should not have two consecutive events that are equal"
+            )
+
+        }
+    }
 }
\ No newline at end of file