Improve display and management of backup progress.

core-util-jvm/src/main/java/org/signal/core/util/FlowExtensions.kt

@@ -7,7 +7,9 @@ package org.signal.core.util
 
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.flow.conflate
+import kotlinx.coroutines.flow.filterNot
 import kotlinx.coroutines.flow.onEach
 import kotlin.time.Duration
 
@@ -16,9 +18,20 @@ import kotlin.time.Duration
  *
  * You can think of this like debouncing, but with "checkpoints" so that even if you have a constant stream of values,
  * you'll still get an emission every [timeout] (unlike debouncing, which will only emit once the stream settles down).
+ *
+ * You can specify an optional [emitImmediately] function that will indicate whether an emission should skip throttling and
+ * be emitted immediately. This lambda should be stateless, as it may be called multiple times for each item.
  */
-fun <T> Flow<T>.throttleLatest(timeout: Duration): Flow<T> {
-  return this
-    .conflate()
-    .onEach { delay(timeout) }
+fun <T> Flow<T>.throttleLatest(timeout: Duration, emitImmediately: (T) -> Boolean = { false }): Flow<T> {
+  val rootFlow = this
+  return channelFlow {
+    rootFlow
+      .onEach { if (emitImmediately(it)) send(it) }
+      .filterNot { emitImmediately(it) }
+      .conflate()
+      .collect {
+        send(it)
+        delay(timeout)
+      }
+  }
 }

core-util-jvm/src/test/java/org/signal/core/util/FlowExtensionsTests.kt

@@ -60,4 +60,20 @@ class FlowExtensionsTests {
 
     assertEquals(listOf(1, 5, 10, 15, 20, 25, 30), output)
   }
+
+  @Test
+  fun `throttleLatest - respects skipThrottle`() = runTest {
+    val testFlow = flow {
+      for (i in 1..30) {
+        emit(i)
+        delay(10)
+      }
+    }
+
+    val output = testFlow
+      .throttleLatest(50.milliseconds) { it in setOf(2, 3, 4, 26, 27, 28) }
+      .toList()
+
+    assertEquals(listOf(1, 2, 3, 4, 5, 10, 15, 20, 25, 26, 27, 28, 30), output)
+  }
 }