Add sample operator to Flow, which can sample a Flow with the emissions of another Flow

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -1105,7 +1105,7 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun retryWhen (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function4;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun runningFold (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun runningReduce (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
-	public static final fun sample (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
+	public static final fun sample (Lkotlinx/coroutines/flow/Flow;Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun sample-HG0u8IE (Lkotlinx/coroutines/flow/Flow;J)Lkotlinx/coroutines/flow/Flow;
 	public static final fun scan (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun scanFold (Lkotlinx/coroutines/flow/Flow;Ljava/lang/Object;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;

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

@@ -203,7 +203,7 @@ public fun <T> Flow<T>.debounce(timeout: (T) -> Duration): Flow<T> =
         timeout(emittedItem).toDelayMillis()
     }
 
-private fun <T> Flow<T>.debounceInternal(timeoutMillisSelector: (T) -> Long) : Flow<T> =
+private fun <T> Flow<T>.debounceInternal(timeoutMillisSelector: (T) -> Long): Flow<T> =
     scopedFlow { downstream ->
         // Produce the values using the default (rendezvous) channel
         val values = produce {
@@ -273,30 +273,76 @@ private fun <T> Flow<T>.debounceInternal(timeoutMillisSelector: (T) -> Long) : F
  */
 @FlowPreview
 public fun <T> Flow<T>.sample(periodMillis: Long): Flow<T> {
-    require(periodMillis > 0) { "Sample period should be positive" }
+    return sample(flow {
+        delay(periodMillis)
+        while (true) {
+            emit(Unit)
+            delay(periodMillis)
+        }
+    })
+}
+
+/**
+ * Returns a flow that emits only the latest value emitted by the original flow only when the [sampler] emits.
+ *
+ * Example:
+ * ```
+ * flow {
+ *     repeat(10) {
+ *         emit(it)
+ *         delay(50)
+ *     }
+ * }.sampleBy(flow {
+ *     repeat(10) {
+ *         delay(100)
+ *         emit(it)
+ *     }
+ * })
+ *
+ * ```
+ * produces `0, 2, 4, 6, 8`.
+ *
+ * Note that the latest element is not emitted if it does not fit into the sampling window.
+ */
+public fun <T, R> Flow<T>.sample(sampler: Flow<R>): Flow<T> {
     return scopedFlow { downstream ->
         val values = produce(capacity = Channel.CONFLATED) {
             collect { value -> send(value ?: NULL) }
         }
+
+        val samplerProducer = produce(capacity = 0) {
+            sampler.collect { value ->
+                send(value)
+            }
+        }
         var lastValue: Any? = null
-        val ticker = fixedPeriodTicker(periodMillis)
         while (lastValue !== DONE) {
             select<Unit> {
                 values.onReceiveCatching { result ->
                     result
                         .onSuccess { lastValue = it }
                         .onFailure {
                             it?.let { throw it }
-                            ticker.cancel(ChildCancelledException())
+                            samplerProducer.cancel(ChildCancelledException())
                             lastValue = DONE
                         }
                 }
 
-                // todo: shall be start sampling only when an element arrives or sample aways as here?
-                ticker.onReceive {
-                    val value = lastValue ?: return@onReceive
-                    lastValue = null // Consume the value
-                    downstream.emit(NULL.unbox(value))
+                samplerProducer.onReceiveCatching { samplerResult ->
+                    samplerResult
+                        .onSuccess { sampledValue ->
+                            if (sampledValue != null) {
+                                val value = lastValue ?: return@onSuccess
+                                lastValue = null // Consume the value
+                                downstream.emit(NULL.unbox(value))
+                            } else {
+                                lastValue = DONE
+                            }
+                        }
+                        .onFailure {
+                            lastValue = DONE
+                        }
+
                 }
             }
         }
@@ -306,7 +352,10 @@ public fun <T> Flow<T>.sample(periodMillis: Long): Flow<T> {
 /*
  * TODO this design (and design of the corresponding operator) depends on #540
  */
-internal fun CoroutineScope.fixedPeriodTicker(delayMillis: Long, initialDelayMillis: Long = delayMillis): ReceiveChannel<Unit> {
+internal fun CoroutineScope.fixedPeriodTicker(
+    delayMillis: Long,
+    initialDelayMillis: Long = delayMillis
+): ReceiveChannel<Unit> {
     require(delayMillis >= 0) { "Expected non-negative delay, but has $delayMillis ms" }
     require(initialDelayMillis >= 0) { "Expected non-negative initial delay, but has $initialDelayMillis ms" }
     return produce(capacity = 0) {