Add debounce selector

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

@@ -16,7 +16,7 @@ import kotlin.time.*
 
 /**
  * Returns a flow that mirrors the original flow, but filters out values
- * that are followed by the newer values within the given [timeout][timeoutMillis].
+ * that are followed by the newer values within the given [timeoutMillis].
  * The latest value is always emitted.
  *
  * Example:
@@ -37,39 +37,50 @@ import kotlin.time.*
  *
  * Note that the resulting flow does not emit anything as long as the original flow emits
  * items faster than every [timeoutMillis] milliseconds.
+ * @param timeoutMillis must be positive
  */
 @FlowPreview
 public fun <T> Flow<T>.debounce(timeoutMillis: Long): Flow<T> {
     require(timeoutMillis > 0) { "Debounce timeout should be positive" }
-    return scopedFlow { downstream ->
-        // Actually Any, KT-30796
-        val values = produce<Any?>(capacity = Channel.CONFLATED) {
-            collect { value -> send(value ?: NULL) }
-        }
-        var lastValue: Any? = null
-        while (lastValue !== DONE) {
-            select<Unit> {
-                // Should be receiveOrClosed when boxing issues are fixed
-                values.onReceiveOrNull {
-                    if (it == null) {
-                        if (lastValue != null) downstream.emit(NULL.unbox(lastValue))
-                        lastValue = DONE
-                    } else {
-                        lastValue = it
-                    }
-                }
+    return debounceInternal { timeoutMillis }
+}
 
-                lastValue?.let { value ->
-                    // set timeout when lastValue != null
-                    onTimeout(timeoutMillis) {
-                        lastValue = null // Consume the value
-                        downstream.emit(NULL.unbox(value))
-                    }
-                }
-            }
-        }
+/**
+ * A variation of [debounce] that allows specifying the timeout value dynamically.
+ *
+ * Example:
+ * ```
+ * flow {
+ *     emit(1)
+ *     delay(100)
+ *     emit(2)
+ *     delay(100)
+ *     emit(3)
+ *     delay(1000)
+ *     emit(4)
+ *     delay(100)
+ *     emit(5)
+ *     delay(100)
+ *     emit(6)
+ * }.debounce {
+ *     if (it == 4) {
+ *         0L
+ *     } else {
+ *         300L
+ *     }
+ * }
+ * ```
+ * produces `3, 4, 6`.
+ *
+ * @param timeoutMillisSelector [T] is the emitted value and the return value is timeout in milliseconds. 0ms timeout is allowed.
+ */
+@FlowPreview
+public fun <T> Flow<T>.debounce(timeoutMillisSelector: (T) -> Long): Flow<T> =
+    debounceInternal { emittedItem ->
+        val timeoutMillis = timeoutMillisSelector(emittedItem)
+        require(timeoutMillis >= 0) { "Debounce timeout should not be negative" }
+        timeoutMillis
     }
-}
 
 /**
  * Returns a flow that mirrors the original flow, but filters out values
@@ -99,6 +110,38 @@ public fun <T> Flow<T>.debounce(timeoutMillis: Long): Flow<T> {
 @FlowPreview
 public fun <T> Flow<T>.debounce(timeout: Duration): Flow<T> = debounce(timeout.toDelayMillis())
 
+private fun <T> Flow<T>.debounceInternal(timeoutMillisSelector: (T) -> Long) : Flow<T> =
+    scopedFlow { downstream ->
+        // Actually Any, KT-30796
+        val values = produce<Any?>(capacity = Channel.CONFLATED) {
+            collect { value -> send(value ?: NULL) }
+        }
+        var lastValue: Any? = null
+        while (lastValue !== DONE) {
+            select<Unit> {
+                // Should be receiveOrClosed when boxing issues are fixed
+                values.onReceiveOrNull {
+                    if (it == null) {
+                        if (lastValue != null) downstream.emit(NULL.unbox(lastValue))
+                        lastValue = DONE
+                    } else {
+                        lastValue = it
+                    }
+                }
+
+                lastValue?.let { value ->
+                    val unboxedValue: T = NULL.unbox(value)
+                    val timeoutMillis = timeoutMillisSelector(unboxedValue)
+                    // set timeout when lastValue != null
+                    onTimeout(timeoutMillis) {
+                        lastValue = null // Consume the value
+                        downstream.emit(unboxedValue)
+                    }
+                }
+            }
+        }
+    }
+
 /**
  * Returns a flow that emits only the latest value emitted by the original flow during the given sampling [period][periodMillis].
  *

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

@@ -11,7 +11,7 @@ import kotlin.time.*
 
 class DebounceTest : TestBase() {
     @Test
-    public fun testBasic() = withVirtualTime {
+    fun testBasic() = withVirtualTime {
         expect(1)
         val flow = flow {
             expect(3)
@@ -175,7 +175,6 @@ class DebounceTest : TestBase() {
             expect(2)
             yield()
             throw TestException()
-            it
         }
 
         assertFailsWith<TestException>(flow)
@@ -193,7 +192,6 @@ class DebounceTest : TestBase() {
             expect(2)
             yield()
             throw TestException()
-            it
         }
 
         assertFailsWith<TestException>(flow)
@@ -202,7 +200,7 @@ class DebounceTest : TestBase() {
 
     @ExperimentalTime
     @Test
-    public fun testDurationBasic() = withVirtualTime {
+    fun testDurationBasic() = withVirtualTime {
         expect(1)
         val flow = flow {
             expect(3)
@@ -223,4 +221,53 @@ class DebounceTest : TestBase() {
         assertEquals(listOf("A", "D", "E"), result)
         finish(5)
     }
+
+    @Test
+    fun testDebounceSelector() = withVirtualTime {
+        expect(1)
+        val flow = flow {
+            expect(3)
+            emit("A")
+            delay(1500)
+            emit("B")
+            delay(500)
+            emit("C")
+            delay(1)
+            emit("D")
+            delay(800)
+            emit("E")
+            expect(4)
+        }
+
+        expect(2)
+        val result = flow.debounce {
+            if (it == "C") {
+                0
+            } else {
+                1000
+            }
+        }.toList()
+
+        assertEquals(listOf("A", "C", "E"), result)
+        finish(5)
+    }
+
+    @Test
+    fun testZeroDebounceTime() = runTest {
+        assertFailsWith<IllegalArgumentException> {
+            flowOf<Int>().debounce(0)
+        }
+    }
+
+    @Test
+    fun testNegativeDebounceTimeSelector() = withVirtualTime {
+        expect(1)
+        val flow = flow {
+            expect(2)
+            emit("A")
+            delay(1)
+        }
+        assertFailsWith<IllegalArgumentException>(flow.debounce { -1 })
+        finish(3)
+    }
 }