Merge pull request #1313 from AllanWang/enhancement/deferred

Enhancement/deferred

5 files changed, 54 insertions, 105 deletions

diff --git a/app/src/test/kotlin/com/pitchedapps/frost/MiscTest.kt b/app/src/test/kotlin/com/pitchedapps/frost/MiscTest.kt
deleted file mode 100644
index 20610b2a..00000000
--- a/app/src/test/kotlin/com/pitchedapps/frost/MiscTest.kt
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Copyright 2018 Allan Wang
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-package com.pitchedapps.frost
-
-import com.pitchedapps.frost.facebook.requests.zip
-import org.junit.Test
-import kotlin.test.assertTrue
-
-/**
- * Created by Allan Wang on 2017-06-14.
- */
-class MiscTest {
-
-    /**
-     * Spin off 15 threads
-     * Pause each for 1 - 2s
-     * Ensure that total zipped process does not take over 5s
-     */
-    @Test
-    fun zip() {
-        val now = System.currentTimeMillis()
-        val base = 1
-        val data: LongArray = (0..15).map { Math.random() + base }
-            .toTypedArray().zip(List<Long>::toLongArray) {
-                Thread.sleep((it * 1000).toLong())
-                System.currentTimeMillis() - now
-            }.blockingGet()
-        println(data.contentToString())
-        assertTrue(
-            data.all { it >= base * 1000 && it < base * 1000 * 5 },
-            "zip did not seem to work on different threads"
-        )
-    }
-}
diff --git a/app/src/test/kotlin/com/pitchedapps/frost/facebook/requests/FbRequestTest.kt b/app/src/test/kotlin/com/pitchedapps/frost/facebook/requests/FbRequestTest.kt
index ec765448..8610436a 100644
--- a/app/src/test/kotlin/com/pitchedapps/frost/facebook/requests/FbRequestTest.kt
+++ b/app/src/test/kotlin/com/pitchedapps/frost/facebook/requests/FbRequestTest.kt
@@ -84,7 +84,7 @@ class FbRequestTest {
         val data = AUTH.getMenuData().invoke()
         assertNotNull(data)
         println(ObjectMapper().writerWithDefaultPrettyPrinter().writeValueAsString(data))
-        assertTrue(data!!.data.isNotEmpty())
+        assertTrue(data.data.isNotEmpty())
         assertTrue(data.footer.hasContent, "Footer may be badly parsed")
         val items = data.flatMapValid()
         assertTrue(items.size > 15, "Something may be badly parsed")
diff --git a/app/src/test/kotlin/com/pitchedapps/frost/internal/Internal.kt b/app/src/test/kotlin/com/pitchedapps/frost/internal/Internal.kt
index 061e7c38..b8d9635a 100644
--- a/app/src/test/kotlin/com/pitchedapps/frost/internal/Internal.kt
+++ b/app/src/test/kotlin/com/pitchedapps/frost/internal/Internal.kt
@@ -22,13 +22,10 @@ import com.pitchedapps.frost.facebook.get
 import com.pitchedapps.frost.facebook.requests.RequestAuth
 import com.pitchedapps.frost.facebook.requests.getAuth
 import com.pitchedapps.frost.utils.frostJsoup
-import io.reactivex.Completable
 import org.junit.Assume
-import org.junit.Test
 import java.io.File
 import java.io.FileInputStream
 import java.util.Properties
-import java.util.concurrent.TimeUnit
 import kotlin.reflect.full.starProjectedType
 import kotlin.test.assertEquals
 import kotlin.test.assertTrue
@@ -97,34 +94,3 @@ fun Any.assertComponentsNotEmpty() {
 fun <T : Comparable<T>> List<T>.assertDescending(tag: String) {
     assertEquals(sortedDescending(), this, "$tag not sorted in descending order")
 }
-
-interface CompletableCallback {
-    fun onComplete()
-    fun onError(message: String)
-}
-
-inline fun concurrentTest(crossinline caller: (callback: CompletableCallback) -> Unit) {
-    val result = Completable.create { emitter ->
-        caller(object : CompletableCallback {
-            override fun onComplete() = emitter.onComplete()
-            override fun onError(message: String) = emitter.onError(Throwable(message))
-        })
-    }.blockingGet(5, TimeUnit.SECONDS)
-    if (result != null)
-        throw RuntimeException("Concurrent fail: ${result.message}")
-}
-
-class InternalTest {
-    @Test
-    fun concurrentTest() = try {
-        concurrentTest { result ->
-            Thread().run {
-                Thread.sleep(100)
-                result.onError("Intentional fail")
-            }
-        }
-        fail("Did not throw exception")
-    } catch (e: Exception) {
-        // pass
-    }
-}
diff --git a/app/src/test/kotlin/com/pitchedapps/frost/kotlin/FlyweightTest.kt b/app/src/test/kotlin/com/pitchedapps/frost/kotlin/FlyweightTest.kt
index 0eee530e..d1d976b6 100644
--- a/app/src/test/kotlin/com/pitchedapps/frost/kotlin/FlyweightTest.kt
+++ b/app/src/test/kotlin/com/pitchedapps/frost/kotlin/FlyweightTest.kt
@@ -16,8 +16,8 @@
  */
 package com.pitchedapps.frost.kotlin
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.async
 import kotlinx.coroutines.runBlocking
 import org.junit.Rule
 import org.junit.rules.Timeout
@@ -25,6 +25,7 @@ import java.util.concurrent.atomic.AtomicInteger
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
 import kotlin.test.assertTrue
 import kotlin.test.fail
 
@@ -42,7 +43,7 @@ class FlyweightTest {
     @BeforeTest
     fun before() {
         callCount = AtomicInteger(0)
-        flyweight = Flyweight(GlobalScope, 100, 200L) {
+        flyweight = Flyweight(GlobalScope, 200L) {
             callCount.incrementAndGet()
             when (it) {
                 LONG_RUNNING_KEY -> Thread.sleep(100000)
@@ -54,7 +55,7 @@ class FlyweightTest {
 
     @Test
     fun basic() {
-        assertEquals(2, runBlocking { flyweight.fetch(1) }, "Invalid result")
+        assertEquals(2, runBlocking { flyweight.fetch(1).await() }, "Invalid result")
         assertEquals(1, callCount.get(), "1 call expected")
     }
 
@@ -62,9 +63,7 @@ class FlyweightTest {
     fun multipleWithOneKey() {
         val results: List<Int> = runBlocking {
             (0..1000).map {
-                flyweight.scope.async {
-                    flyweight.fetch(1)
-                }
+                flyweight.fetch(1)
             }.map { it.await() }
         }
         assertEquals(1, callCount.get(), "1 call expected")
@@ -75,12 +74,12 @@ class FlyweightTest {
     @Test
     fun consecutiveReuse() {
         runBlocking {
-            flyweight.fetch(1)
+            flyweight.fetch(1).await()
             assertEquals(1, callCount.get(), "1 call expected")
-            flyweight.fetch(1)
+            flyweight.fetch(1).await()
             assertEquals(1, callCount.get(), "Reuse expected")
             Thread.sleep(300)
-            flyweight.fetch(1)
+            flyweight.fetch(1).await()
             assertEquals(2, callCount.get(), "Refetch expected")
         }
     }
@@ -88,10 +87,10 @@ class FlyweightTest {
     @Test
     fun invalidate() {
         runBlocking {
-            flyweight.fetch(1)
+            flyweight.fetch(1).await()
             assertEquals(1, callCount.get(), "1 call expected")
             flyweight.invalidate(1)
-            flyweight.fetch(1)
+            flyweight.fetch(1).await()
             assertEquals(2, callCount.get(), "New call expected")
         }
     }
@@ -99,24 +98,19 @@ class FlyweightTest {
     @Test
     fun destroy() {
         runBlocking {
-            val longRunningResult = async { flyweight.fetch(LONG_RUNNING_KEY) }
-            flyweight.fetch(1)
+            val longRunningResult = flyweight.fetch(LONG_RUNNING_KEY)
+            flyweight.fetch(1).await()
             flyweight.cancel()
             try {
-                flyweight.fetch(1)
+                flyweight.fetch(1).await()
                 fail("Flyweight should not be fulfilled after it is destroyed")
-            } catch (e: Exception) {
-                assertEquals("Flyweight is not active", e.message, "Incorrect error found on fetch after destruction")
+            } catch (ignore: CancellationException) {
             }
             try {
+                assertFalse(longRunningResult.isActive, "Long running result should no longer be active")
                 longRunningResult.await()
                 fail("Flyweight should have cancelled previously running requests")
-            } catch (e: Exception) {
-                assertEquals(
-                    "Flyweight cancelled",
-                    e.message,
-                    "Incorrect error found on fetch cancelled by destruction"
-                )
+            } catch (ignore: CancellationException) {
             }
         }
     }
diff --git a/app/src/test/kotlin/com/pitchedapps/frost/utils/CoroutineTest.kt b/app/src/test/kotlin/com/pitchedapps/frost/utils/CoroutineTest.kt
index 72eb6076..e7520794 100644
--- a/app/src/test/kotlin/com/pitchedapps/frost/utils/CoroutineTest.kt
+++ b/app/src/test/kotlin/com/pitchedapps/frost/utils/CoroutineTest.kt
@@ -18,6 +18,7 @@ package com.pitchedapps.frost.utils
 
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.async
 import kotlinx.coroutines.channels.BroadcastChannel
@@ -31,6 +32,7 @@ import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.withContext
 import java.util.concurrent.Executors
 import kotlin.coroutines.EmptyCoroutineContext
+import kotlin.test.Ignore
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
@@ -206,4 +208,39 @@ class CoroutineTest {
             )
         }
     }
+
+    /**
+     * When using [uniqueOnly] for channels with limited capacity,
+     * the duplicates should not count towards the actual capacity
+     */
+    @Ignore("Not yet working as unique only buffered removes the capacity limitation of the channel")
+    @Test
+    fun uniqueOnlyBuffer() {
+        val channel = Channel<Int>(3)
+        runBlocking {
+
+            val deferred = async {
+                listen(channel.uniqueOnly(GlobalScope)) {
+                    // Throttle consumer
+                    delay(50)
+                    return@listen false
+                }
+            }
+
+            listOf(0, 1, 1, 1, 1, 1, 2, 2, 2).forEach {
+                delay(10)
+                channel.offer(it)
+            }
+
+            channel.close()
+
+            val data = deferred.await()
+
+            assertEquals(
+                listOf(0, 1, 2),
+                data,
+                "Unique receiver should not have two consecutive events that are equal"
+            )
+        }
+    }
 }