Introducing collectLatest terminal operator

Fixes #1269

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -847,6 +847,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun collect (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun collectIndexed (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun collectLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final synthetic fun combine (Ljava/lang/Iterable;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun combine (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun combine (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function4;)Lkotlinx/coroutines/flow/Flow;
@@ -913,7 +914,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun getDEFAULT_CONCURRENCY ()I
 	public static final fun launchIn (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/CoroutineScope;)Lkotlinx/coroutines/Job;
 	public static final fun map (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun mapLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun mapLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun mapNotNull (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun merge (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun observeOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;

kotlinx-coroutines-core/common/src/flow/internal/Merge.kt

@@ -48,12 +48,17 @@ internal class ChannelFlowMerge<T>(
     // The actual merge implementation with concurrency limit
     private suspend fun mergeImpl(scope: CoroutineScope, collector: ConcurrentFlowCollector<T>) {
         val semaphore = Semaphore(concurrency)
-        @Suppress("UNCHECKED_CAST")
+        val job: Job? = coroutineContext[Job]
         flow.collect { inner ->
-            semaphore.acquire() // Acquire concurrency permit
+            /*
+             * We launch a coroutine on each emitted element and the only potential
+             * suspension point in this collector is `semaphore.acquire` that rarely suspends,
+             * so we manually check for cancellation to propagate it to the upstream in time.
+             */
+            job?.ensureActive()
+            semaphore.acquire()
             scope.launch {
                 try {
-                    scope.ensureActive()
                     inner.collect(collector)
                 } finally {
                     semaphore.release() // Release concurrency permit

kotlinx-coroutines-core/common/src/flow/terminal/Collect.kt

@@ -86,9 +86,45 @@ public suspend inline fun <T> Flow<T>.collectIndexed(crossinline action: suspend
         override suspend fun emit(value: T) = action(checkIndexOverflow(index++), value)
     })
 
+/**
+ * Terminal flow operator that collects the given flow with a provided [action].
+ * The crucial difference from [collect] is that when the original flow emits a new value, [action] block for previous
+ * value is cancelled.
+ * It can be demonstrated by the following example:
+ * ```
+ * flow {
+ *     emit(1)
+ *     delay(50)
+ *     emit(2)
+ * }.collectLatest { value ->
+ *     println("Collecting $value")
+ *     delay(100) // Emulate work
+ *     println("$value collected")
+ * }
+ * ```
+ *
+ * prints "Collecting 1, Collecting 2, 2 collected"
+ */
+@ExperimentalCoroutinesApi
+public suspend fun <T> Flow<T>.collectLatest(action: suspend (value: T) -> Unit) {
+    /*
+     * Implementation note:
+     * buffer(0) is inserted here to fulfil user's expectations in sequential usages, e.g.:
+     * ```
+     * flowOf(1, 2, 3).collectLatest {
+     *     delay(1)
+     *     println(it) // Expect only 3 to be printed
+     * }
+     * ```
+     *
+     * It's not the case for intermediate operators which users mostly use for interactive UI,
+     * where performance of dispatch is more important.
+     */
+    mapLatest(action).buffer(0).collect()
+}
+
 /**
  * Collects all the values from the given [flow] and emits them to the collector.
- * 
  * It is a shorthand for `flow.collect { value -> emit(value) }`.
  */
 @ExperimentalCoroutinesApi

kotlinx-coroutines-core/common/test/flow/terminal/CollectLatestTest.kt

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow
+
+import kotlinx.coroutines.*
+import kotlin.test.*
+
+class CollectLatestTest : TestBase() {
+    @Test
+    fun testNoSuspension() = runTest {
+        flowOf(1, 2, 3).collectLatest {
+            expect(it)
+        }
+        finish(4)
+    }
+
+    @Test
+    fun testSuspension() = runTest {
+        flowOf(1, 2, 3).collectLatest {
+            yield()
+            expect(1)
+        }
+        finish(2)
+    }
+
+    @Test
+    fun testUpstreamErrorSuspension() = runTest({it is TestException}) {
+        try {
+            flow {
+                emit(1)
+                throw TestException()
+            }.collectLatest { expect(1) }
+            expectUnreached()
+        } finally {
+            finish(2)
+        }
+    }
+
+    @Test
+    fun testDownstreamError() = runTest({it is TestException}) {
+        try {
+            flow {
+                emit(1)
+                hang { expect(1) }
+            }.collectLatest {
+                throw TestException()
+            }
+            expectUnreached()
+        } finally {
+            finish(2)
+        }
+
+    }
+}