add debouncedBy extension

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 0.22.5
 
+* `Coroutines`:
+  * Add extension `Flow.debouncedBy`
+
 ## 0.22.4
 
 * `Versions`:

coroutines/src/commonMain/kotlin/dev/inmo/micro_utils/coroutines/FlowDebouncedBy.kt

@@ -0,0 +1,40 @@
+package dev.inmo.micro_utils.coroutines
+
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.async
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.*
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
+import kotlin.jvm.JvmInline
+import kotlin.time.Duration
+
+@JvmInline
+private value class DebouncedByData<T>(
+    val millisToData: Pair<Long, T>
+)
+
+fun <T> Flow<T>.debouncedBy(timeout: (T) -> Long, markerFactory: (T) -> Any?): Flow<T> = channelFlow {
+    val jobs = mutableMapOf<Any?, Job>()
+    val mutex = Mutex()
+    subscribe(this) {
+        mutex.withLock {
+            val marker = markerFactory(it)
+            lateinit var job: Job
+            job = async {
+                delay(timeout(it))
+                mutex.withLock {
+                    if (jobs[marker] === job) {
+                        this@channelFlow.send(it)
+                        jobs.remove(marker)
+                    }
+                }
+            }
+            jobs[marker] ?.cancel()
+            jobs[marker] = job
+        }
+    }
+}
+
+fun <T> Flow<T>.debouncedBy(timeout: Long, markerFactory: (T) -> Any?): Flow<T> = debouncedBy({ timeout }, markerFactory)
+fun <T> Flow<T>.debouncedBy(timeout: Duration, markerFactory: (T) -> Any?): Flow<T> = debouncedBy({ timeout.inWholeMilliseconds }, markerFactory)

coroutines/src/commonTest/kotlin/DebouncedByTests.kt

@@ -0,0 +1,42 @@
+import dev.inmo.micro_utils.coroutines.debouncedBy
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertTrue
+
+class DebouncedByTests {
+    @Test
+    fun testThatParallelDebouncingWorksCorrectly() = runTest {
+        val dataToMarkerFactories = listOf(
+            1 to 0,
+            2 to 1,
+            3 to 2,
+            4 to 0,
+            5 to 1,
+            6 to 2,
+            7 to 0,
+            8 to 1,
+            9 to 2,
+        )
+
+        val collected = mutableListOf<Int>()
+
+        dataToMarkerFactories.asFlow().debouncedBy(10L) {
+            it.second
+        }.collect {
+            when (it.second) {
+                0 -> assertEquals(7, it.first)
+                1 -> assertEquals(8, it.first)
+                2 -> assertEquals(9, it.first)
+                else -> error("wtf")
+            }
+            collected.add(it.first)
+        }
+
+        val expectedList = listOf(7, 8, 9)
+        assertEquals(expectedList, collected)
+        assertTrue { collected.containsAll(expectedList) }
+        assertTrue { expectedList.containsAll(collected) }
+    }
+}