Add distinctUntilChanged operator that uses a comparator function instead of a key extractor.

Author: zach-klippenstein

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

@@ -833,6 +833,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun delayEach (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
 	public static final fun delayFlow (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
 	public static final fun distinctUntilChanged (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun distinctUntilChanged (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun distinctUntilChangedBy (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function1;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun drop (Lkotlinx/coroutines/flow/Flow;I)Lkotlinx/coroutines/flow/Flow;
 	public static final fun dropWhile (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;

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

@@ -18,17 +18,37 @@ import kotlinx.coroutines.flow.unsafeFlow as flow
 @ExperimentalCoroutinesApi
 public fun <T> Flow<T>.distinctUntilChanged(): Flow<T> = distinctUntilChangedBy { it }
 
+/**
+ * Returns flow where all subsequent repetitions of the same value are filtered out, when compared
+ * with each other via the provided [areEquivalent] function.
+ */
+@FlowPreview
+public fun <T> Flow<T>.distinctUntilChanged(areEquivalent: (old: T, new: T) -> Boolean): Flow<T> =
+    distinctUntilChangedBy(keySelector = { it }, areEquivalent = areEquivalent)
+
 /**
  * Returns flow where all subsequent repetitions of the same key are filtered out, where
  * key is extracted with [keySelector] function.
  */
 @FlowPreview
 public fun <T, K> Flow<T>.distinctUntilChangedBy(keySelector: (T) -> K): Flow<T> =
+    distinctUntilChangedBy(keySelector = keySelector, areEquivalent = { old, new -> old == new })
+
+/**
+ * Returns flow where all subsequent repetitions of the same key are filtered out, where
+ * keys are extracted with [keySelector] function and compared with each other via the
+ * provided [areEquivalent] function.
+ */
+private inline fun <T, K> Flow<T>.distinctUntilChangedBy(
+    crossinline keySelector: (T) -> K,
+    crossinline areEquivalent: (old: K, new: K) -> Boolean
+): Flow<T> =
     flow {
         var previousKey: Any? = NULL
         collect { value ->
             val key = keySelector(value)
-            if (previousKey != key) {
+            @Suppress("UNCHECKED_CAST")
+            if (previousKey === NULL || !areEquivalent(previousKey as K, key)) {
                 previousKey = key
                 emit(value)
             }

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

@@ -32,6 +32,28 @@ class DistinctUntilChangedTest : TestBase() {
         assertEquals(4, sum2)
     }
 
+    @Test
+    fun testDistinctUntilChangedAreEquivalent() = runTest {
+        val flow = flow {
+            emit(Box(1))
+            emit(Box(1))
+            emit(Box(2))
+            emit(Box(1))
+        }
+
+        val sum1 = flow.distinctUntilChanged().map { it.i }.sum()
+        val sum2 = flow.distinctUntilChanged { old, new -> old.i == new.i }.map { it.i }.sum()
+        assertEquals(5, sum1)
+        assertEquals(4, sum2)
+    }
+
+    @Test
+    fun testDistinctUntilChangedAreEquivalentSingleValue() = runTest {
+        val flow = flowOf(1)
+        val values = flow.distinctUntilChanged { _, _ -> fail("Expected not to compare single value.") }.toList()
+        assertEquals(listOf(1), values)
+    }
+
     @Test
     fun testThrowingKeySelector() = runTest {
         val flow = flow {
@@ -50,8 +72,26 @@ class DistinctUntilChangedTest : TestBase() {
     }
 
     @Test
-    fun testDistinctUntilChangedNull() = runTest{
-        val flow = flowOf(null, 1, null).distinctUntilChanged()
+    fun testThrowingAreEquivalent() = runTest {
+        val flow = flow {
+            coroutineScope {
+                launch(start = CoroutineStart.ATOMIC) {
+                    hang { expect(3) }
+                }
+                expect(2)
+                emit(1)
+                emit(2)
+            }
+        }.distinctUntilChanged { _, _ -> throw TestException() }
+
+        expect(1)
+        assertFailsWith<TestException>(flow)
+        finish(4)
+    }
+
+    @Test
+    fun testDistinctUntilChangedNull() = runTest {
+        val flow = flowOf(null, 1, null, null).distinctUntilChanged()
         assertEquals(listOf(null, 1, null), flow.toList())
     }
 }