Repository: kevincianfarini/cardiologist (kotlin)

Issue:
Consider introducing `RecurringJobMode.Skip`
Currently we allow both concurrent execution of pulses and cancellation of pulses which would otherwise experience backpressure. Another option for coping with backpressure which we can't apply is to skip the current item if the previous one is still executing.

Pull request:
Closes #86.

Patch:
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
index 4e37ddc..801efb5 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/Pulse.kt
@@ -1,5 +1,6 @@
 package io.github.kevincianfarini.cardiologist
 
+import io.github.kevincianfarini.cardiologist.impl.collectCurrent
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.collectLatest
@@ -47,5 +48,8 @@ public value class Pulse internal constructor(private val flow: Flow<Pair<Instan
                 launch { action(scheduled, occurred) }
             }
         }
+        RecurringJobMode.Skip -> flow.collectCurrent { (scheduled, occurred) ->
+            action(scheduled, occurred)
+        }
     }
 }
\ No newline at end of file
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt
index 8d4f094..258a7f6 100644
--- a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/RecurringJobMode.kt
@@ -1,14 +1,22 @@
 package io.github.kevincianfarini.cardiologist
 
 public enum class RecurringJobMode {
+
     /**
      * Schedules recurring jobs concurrently. That is if job `n` is still active when job
      * `n + 1` should begin, then job `n` continues to run concurrently alongside job `n + 1`.
      */
     Concurrent,
+
     /**
      * Schedules recurring jobs sequentially by cancelling an unfinished job if a new one
      * should begin executing.
      */
     CancellingSequential,
+
+    /**
+     * Schedules recurring jobs sequentially by skipping a new job if an unfinished job
+     * is still executing.
+     */
+    Skip,
 }
diff --git a/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/impl/flow.kt b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/impl/flow.kt
new file mode 100644
index 0000000..aebe105
--- /dev/null
+++ b/src/commonMain/kotlin/io/github/kevincianfarini/cardiologist/impl/flow.kt
@@ -0,0 +1,41 @@
+package io.github.kevincianfarini.cardiologist.impl
+
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.launch
+
+/**
+ * Terminal flow operator that collects the given flow with a provided action. The crucial difference from collect is
+ * that when the original flow emits a new value and the action block from the previous value has not yet completed,
+ * then the newly emitted value gets skipped.
+ *
+ * It can be demonstrated by the following example:
+ *
+ * ```kt
+ * flow {
+ *     emit(1)
+ *     delay(50)
+ *     emit(2)
+ * }.collectLatest { value ->
+ *     println("Collecting $value")
+ *     delay(100) // Emulate work
+ *     println("$value collected")
+ * }
+ * ```
+ *
+ * which prints:
+ *
+ * ```
+ * Collecting 1
+ * 1 collected
+ * ```
+ */
+internal suspend fun <T> Flow<T>.collectCurrent(action: suspend (value: T) -> Unit) = coroutineScope {
+    var job: Job? = null
+    collect { value ->
+        if (job?.isActive != true) {
+            job = launch { action(value) }
+        }
+    }
+}
\ No newline at end of file


Test patch:
diff --git a/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/impl/FlowCollectCurrentTest.kt b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/impl/FlowCollectCurrentTest.kt
new file mode 100644
index 0000000..3fc256e
--- /dev/null
+++ b/src/commonTest/kotlin/io/github/kevincianfarini/cardiologist/impl/FlowCollectCurrentTest.kt
@@ -0,0 +1,26 @@
+package io.github.kevincianfarini.cardiologist.impl
+
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class FlowCollectCurrentTest {
+
+    @Test
+    fun skips_elements_when_backpressure_occurs() = runTest {
+        val flow = flow {
+            emit(1)
+            delay(50)
+            emit(2)
+        }
+        val output = buildList {
+            flow.collectCurrent { value ->
+                add(value)
+                delay(100)
+            }
+        }
+        assertEquals(expected = listOf(1), actual = output)
+    }
+}
\ No newline at end of file