diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index 6984efc4..216c14e3 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -159,6 +159,25 @@ trait SourceOps[+T] { this: Source[T] => */ def drop(n: Int)(using Ox, StageCapacity): Source[T] = transform(_.drop(n)) + /** Drops elements from the source as long as predicate `f` is satisfied (returns `true`). Note that once the predicate `f` is not + * satisfied (returns `false`) no elements are dropped from the source even if they could still satisfy it. + * + * @param f + * A predicate function. + * @example + * {{{ + * import ox.* + * import ox.channels.Source + * + * scoped { + * Source.empty[Int].dropWhile(_ > 3).toList // List() + * Source.fromValues(1, 2, 3).dropWhile(_ < 3).toList // List(3) + * Source.fromValues(1, 2, 1).dropWhile(_ < 2).toList // List(2, 1) + * } + * }}} + */ + def dropWhile(f: T => Boolean)(using Ox, StageCapacity): Source[T] = transform(_.dropWhile(f)) + def filter(f: T => Boolean)(using Ox, StageCapacity): Source[T] = transform(_.filter(f)) def transform[U](f: Iterator[T] => Iterator[U])(using Ox, StageCapacity): Source[U] = diff --git a/core/src/test/scala/ox/channels/SourceOpsDropWhileTest.scala b/core/src/test/scala/ox/channels/SourceOpsDropWhileTest.scala new file mode 100644 index 00000000..f52e5403 --- /dev/null +++ b/core/src/test/scala/ox/channels/SourceOpsDropWhileTest.scala @@ -0,0 +1,34 @@ +package ox.channels + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import ox.* + +class SourceOpsDropWhileTest extends AnyFlatSpec with Matchers { + behavior of "Source.drop" + + it should "not drop from the empty source" in scoped { + val s = Source.empty[Int] + s.dropWhile(_ > 0).toList shouldBe List.empty + } + + it should "drop elements from the source while predicate is true" in scoped { + val s = Source.fromValues(1, 2, 3) + s.dropWhile(_ < 3).toList shouldBe List(3) + } + + it should "drop elements from the source until predicate is true and then emit subsequent ones" in scoped { + val s = Source.fromValues(1, 2, 3, 2) + s.dropWhile(_ < 3).toList shouldBe List(3, 2) + } + + it should "not drop elements from the source if predicate is false" in scoped { + val s = Source.fromValues(1, 2, 3) + s.dropWhile(_ > 3).toList shouldBe List(1, 2, 3) + } + + it should "not drop elements from the source when predicate is false for first or more elements" in scoped { + val s = Source.fromValues(1, 4, 5) + s.dropWhile(_ > 3).toList shouldBe List(1, 4, 5) + } +}