Flow.conflate operator

Author: elizarov

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

@@ -812,6 +812,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun combineLatest (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function5;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun combineLatest (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function6;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun combineLatest (Lkotlinx/coroutines/flow/Flow;[Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function0;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun conflate (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun count (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun count (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun debounce (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;

kotlinx-coroutines-core/common/src/flow/operators/Context.kt

@@ -94,6 +94,11 @@ import kotlin.jvm.*
  * }
  * ```
  *
+ * ### Conflation
+ *
+ * Usage of this function with [capacity] of [Channel.CONFLATED][Channel.CONFLATED] is provided as a shortcut via
+ * [conflate] operator. See its documentation for details.
+ *
  * @param capacity type/capacity of the buffer between coroutines. Allowed values are the same as in `Channel(...)`
  * factory function: [BUFFERED][Channel.BUFFERED] (by default), [CONFLATED][Channel.CONFLATED],
  * [RENDEZVOUS][Channel.RENDEZVOUS], [UNLIMITED][Channel.UNLIMITED] or a non-negative value indicating
@@ -110,9 +115,40 @@ public fun <T> Flow<T>.buffer(capacity: Int = BUFFERED): Flow<T> {
         ChannelFlowOperatorImpl(this, capacity = capacity)
 }
 
-// todo: conflate would be a useful operator only when Channel.CONFLATE is changed to always deliver the last send value
-//@FlowPreview
-//public fun <T> Flow<T>.conflate(): Flow<T> = buffer(CONFLATED)
+/**
+ * Conflates flow emissions via conflated channel and runs collector in a separate coroutine.
+ * The effect of this is that emitter is never suspended due to a slow collector, but collector
+ * always gets the most recent value emitted.
+ *
+ * For example, consider the flow that emits integers from 1 to 30 with 100 ms delay between them:
+ *
+ * ```
+ * val flow = flow {
+ *     for (i in 1..30) {
+ *         delay(100)
+ *         emit(i)
+ *     }
+ * }
+ * ```
+ *
+ * Applying `conflate()` operator to it allows a collector that delays 1 second on each element to get
+ * integers 1, 10, 20, 30:
+ *
+ * ```
+ * val result = flow.conflate().onEach { delay(1000) }.toList()
+ * assertEquals(listOf(1, 10, 20, 30), result)
+ * ```
+ *
+ * Note that `conflate` operator is a shortcut for [buffer] with `capacity` of [Channel.CONFLATED][Channel.CONFLATED].
+ *
+ * ### Operator fusion
+ *
+ * Adjacent applications of `conflate`/[buffer], [channelFlow], [flowOn], [produceIn], and [broadcastIn] are
+ * always fused so that only one properly configured channel is used for execution.
+ * **Conflation takes precedence over `buffer()` calls with any other capacity.**
+ */
+@FlowPreview
+public fun <T> Flow<T>.conflate(): Flow<T> = buffer(CONFLATED)
 
 /**
  * The operator that changes the context where this flow is executed to the given [context].

kotlinx-coroutines-core/common/test/flow/operators/BufferTest.kt

@@ -164,7 +164,6 @@ class BufferTest : TestBase() {
         }
 
     @Test
-    @Ignore // todo: conflated behavior is pretty useless right now, because closing channel overwrites last value
     fun testConflate() = runTest {
         expect(1)
         // emit all and conflate / then collect first & last

kotlinx-coroutines-core/common/test/flow/operators/ConflateTest.kt

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.flow.operators
+
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.*
+import kotlin.test.*
+
+class ConflateTest : TestBase() {
+    @Test // from example
+    fun testExample() = withVirtualTime {
+        expect(1)
+        val flow = flow {
+            for (i in 1..30) {
+                delay(100)
+                emit(i)
+            }
+        }
+        val result = flow.conflate().onEach {
+            delay(1000)
+        }.toList()
+        assertEquals(listOf(1, 10, 20, 30), result)
+        finish(2)
+    }
+}