FlowOps: debounce operator (#255)

core/src/main/scala/ox/flow/FlowOps.scala

@@ -86,6 +86,25 @@ class FlowOps[+T]:
         if n != 0 && sampleCounter % n == 0 then emit(t)
     )
 
+  /** Remove subsequent, repeating elements
+    */
+  def debounce: Flow[T] =
+    debounceBy(identity)
+
+  /** Remove subsequent, repeating elements matching 'f'
+    *
+    * @param f
+    *   The function used to compare the previous and current elements
+    */
+  def debounceBy[U](f: T => U): Flow[T] = Flow.usingEmitInline: emit =>
+    var previousElement: Option[U] = None
+    last.run(
+      FlowEmit.fromInline: t =>
+        val currentElement = f(t)
+        if !previousElement.contains(currentElement) then emit(t)
+        previousElement = Some(currentElement)
+    )
+
   /** Applies the given mapping function `f` to each element emitted by this flow, for which the function is defined, and emits the result.
     * If `f` is not defined at an element, the element will be skipped.
     *

core/src/test/scala/ox/flow/FlowOpsDebounceByTest.scala

@@ -0,0 +1,30 @@
+package ox.flow
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import ox.*
+
+class FlowOpsDebounceByTest extends AnyFlatSpec with Matchers:
+  behavior of "debounceBy"
+
+  it should "not debounce if applied on an empty flow" in:
+    val c = Flow.empty[Int]
+    val s = c.debounceBy(_ * 2)
+    s.runToList() shouldBe List.empty
+
+  it should "not debounce if applied on a flow containing only distinct f(value)" in:
+    val c = Flow.fromValues(1 to 10: _*)
+    val s = c.debounceBy(_ * 2)
+    s.runToList() shouldBe (1 to 10)
+
+  it should "debounce if applied on a flow containing repeating f(value)" in:
+    val c = Flow.fromValues(1, 1, 2, 3, 4, 4, 5)
+    val s = c.debounceBy(_ * 2)
+    s.runToList() shouldBe (1 to 5)
+
+  it should "debounce subsequent odd/prime numbers" in:
+    val c = Flow.fromValues(1, 1, 1, 2, 4, 3, 7, 4, 5)
+    val s = c.debounceBy(_ % 2 == 0)
+    s.runToList() shouldBe List(1, 2, 3, 4, 5)
+
+end FlowOpsDebounceByTest

core/src/test/scala/ox/flow/FlowOpsDebounceTest.scala

@@ -0,0 +1,30 @@
+package ox.flow
+
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+import ox.*
+
+class FlowOpsDebounceTest extends AnyFlatSpec with Matchers:
+  behavior of "debounce"
+
+  it should "not debounce if applied on an empty flow" in:
+    val c = Flow.empty[Int]
+    val s = c.debounce
+    s.runToList() shouldBe List.empty
+
+  it should "not debounce if applied on a flow containing only distinct values" in:
+    val c = Flow.fromValues(1 to 10: _*)
+    val s = c.debounce
+    s.runToList() shouldBe (1 to 10)
+
+  it should "debounce if applied on a flow containing only repeating values" in:
+    val c = Flow.fromValues(1, 1, 1, 1, 1)
+    val s = c.debounce
+    s.runToList() shouldBe List(1)
+
+  it should "debounce if applied on a flow containing repeating elements" in:
+    val c = Flow.fromValues(1, 1, 2, 3, 4, 4, 5)
+    val s = c.debounce
+    s.runToList() shouldBe (1 to 5)
+
+end FlowOpsDebounceTest