From a4de6180ef9e31fb9592d5e415bbaae0edda15bf Mon Sep 17 00:00:00 2001 From: Jacek Centkowski Date: Wed, 11 Oct 2023 15:50:00 +0200 Subject: [PATCH] feat: introduce `intersperse` function The intersperse function behaves as `List.mkString` e.g.: * called with an `inject` element only it produces Source.fromValues("f", "b").intersperse(", ") // ("f", ", ", "b") * called with `start`, `inject` and `end` it produces Source.fromValues("f", "b").intersperse("[", ", ", "]") // ("[", "f", ", ", "b", "]") --- .../main/scala/ox/channels/SourceOps.scala | 63 ++++++++++++++++++- .../channels/SourceOpsIntersperseTest.scala | 39 ++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) create mode 100644 core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index 6494193b..de7f9d87 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -2,7 +2,7 @@ package ox.channels import ox.* -import java.util.concurrent.{ArrayBlockingQueue, ConcurrentLinkedQueue, CountDownLatch, LinkedBlockingQueue, Semaphore} +import java.util.concurrent.{CountDownLatch, Semaphore} import scala.collection.mutable import scala.concurrent.duration.FiniteDuration @@ -34,6 +34,67 @@ trait SourceOps[+T] { this: Source[T] => } c2 + /** Intersperses this source with provided element and forwards it to the returned channel. + * + * @param inject + * An element to be injected between the stream elements. + * @return + * A source, onto which elements will be injected. + * @example + * {{{ + * import ox.* + * import ox.channels.Source + * + * scoped { + * Source.empty[String].intersperse(", ").toList // List() + * Source.fromValues("foo").intersperse(", ").toList // List(foo) + * Source.fromValues("foo", "bar").intersperse(", ").toList // List(foo, ", ", bar) + * } + * }}} + */ + def intersperse[U >: T](inject: U)(using Ox, StageCapacity): Source[U] = + intersperse(None, inject, None) + + /** Intersperses this source with start, end and provided elements and forwards it to the returned channel. + * + * @param start + * An element to be prepended to the stream. + * @param inject + * An element to be injected between the stream elements. + * @param end + * An element to be appended to the end of the stream. + * @return + * A source, onto which elements will be injected. + * @example + * {{{ + * import ox.* + * import ox.channels.Source + * + * scoped { + * Source.empty[String].intersperse("[", ", ", "]").toList // List([, ]) + * Source.fromValues("foo").intersperse("[", ", ", "]").toList // List([, foo, ]) + * Source.fromValues("foo", "bar").intersperse("[", ", ", "]").toList // List([, foo, ", ", bar, ]) + * } + * }}} + */ + def intersperse[U >: T](start: U, inject: U, end: U)(using Ox, StageCapacity): Source[U] = + intersperse(Some(start), inject, Some(end)) + + private def intersperse[U >: T](start: Option[U], inject: U, end: Option[U])(using Ox, StageCapacity): Source[U] = + val c = StageCapacity.newChannel[U] + forkDaemon { + start.foreach(c.send) + var firstEmitted = false + repeatWhile { + receive() match + case ChannelClosed.Done => end.foreach(c.send); c.done(); false + case ChannelClosed.Error(e) => c.error(e); false + case v: U @unchecked if !firstEmitted => firstEmitted = true; c.send(v); true + case v: U @unchecked => c.send(inject); c.send(v); true + } + } + c + /** Applies the given mapping function `f` to each element received from this source, and sends the results to the returned channel. At * most `parallelism` invocations of `f` are run in parallel. * diff --git a/core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala b/core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala new file mode 100644 index 00000000..6a8a30cc --- /dev/null +++ b/core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala @@ -0,0 +1,39 @@ +package ox.channels + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import ox.* + +class SourceOpsIntersperseTest extends AnyFlatSpec with Matchers { + behavior of "Source.intersperse" + + it should "intersperse with inject only over an empty source" in supervised { + val s = Source.empty[String] + s.intersperse(", ").toList shouldBe List.empty + } + + it should "intersperse with inject only over a source with one element" in supervised { + val s = Source.fromValues("foo") + s.intersperse(", ").toList shouldBe List("foo") + } + + it should "intersperse with inject only over a source with multiple elements" in supervised { + val s = Source.fromValues("foo", "bar") + s.intersperse(", ").toList shouldBe List("foo", ", ", "bar") + } + + it should "intersperse with start, inject and end over an empty source" in supervised { + val s = Source.empty[String] + s.intersperse("[", ", ", "]").toList shouldBe List("[", "]") + } + + it should "intersperse with start, inject and end over a source with one element" in supervised { + val s = Source.fromValues("foo") + s.intersperse("[", ", ", "]").toList shouldBe List("[", "foo", "]") + } + + it should "intersperse with start, inject and end over a source with multiple elements" in supervised { + val s = Source.fromValues("foo", "bar") + s.intersperse("[", ", ", "]").toList shouldBe List("[", "foo", ", ", "bar", "]") + } +}