From 151b52bcd246ecb577792224a3d31ec2af3dd67e Mon Sep 17 00:00:00 2001 From: Jacek Kunicki Date: Thu, 5 Oct 2023 09:13:50 +0200 Subject: [PATCH 1/5] Add Source.interleave combinator --- .../main/scala/ox/channels/SourceOps.scala | 61 +++++++++++++++++++ .../scala/ox/channels/SourceOpsTest.scala | 45 ++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index 1fa5b4fd..8277b55b 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -168,6 +168,67 @@ trait SourceOps[+T] { this: Source[T] => // + /** Sends a given number of elements (determined byc `segmentSize`) from this source to the returned channel, then sends the same number + * of elements from the `other` source and repeats. If one of the sources is closed before the other, the remaining elements from the + * open one are sent to the returned channel. The order of elements in both sources is preserved. + * + * Must be run within a scope, since a child fork is created which receives from both sources and sends to the resulting channel. + * + * @param other + * The source whose elements will be interleaved with the elements of this source. + * @param segmentSize + * The number of elements sent from each source before switching to the other one. Default is 1. + * @return + * A source to which the interleaved elements from both sources would be sent. + * @example + * {{{ + * scala> + * import ox.* + * import ox.channels.Source + * + * scoped { + * val s1 = Source.fromValues(1, 2, 3, 4) + * val s2 = Source.fromValues(10, 20, 30, 40) + * s1.interleave(s2, segmentSize = 2).toList + * } + * + * scala> val res0: List[Int] = List(1, 2, 10, 20, 3, 4, 30, 40) + * }}} + */ + def interleave[U >: T](other: Source[U], segmentSize: Int = 1)(using Ox, StageCapacity): Source[U] = + val c = StageCapacity.newChannel[U] + var source: Source[U] = this + var counter = 0 + var neitherCompleted = true + + def switchSource(): Unit = { + if (source == this) source = other else source = this + counter = 0 + } + + forkDaemon { + repeatWhile { + source.receive() match + case ChannelClosed.Done => + // if one source has completed, switch to the other one, otherwise (i.e. if both sources have completed) complete the resulting source + if (neitherCompleted) { + neitherCompleted = false + switchSource() + true + } else { + c.done() + false + } + case ChannelClosed.Error(r) => c.error(r); false + case value: U @unchecked => + counter += 1 + // after reaching segmentSize, only switch to the other source if it hasn't completed yet + if (counter == segmentSize && neitherCompleted) switchSource() + c.send(value).isValue + } + } + c + /** Invokes the given function for each received element. Blocks until the channel is done. * @throws ChannelClosedException * when there is an upstream error. diff --git a/core/src/test/scala/ox/channels/SourceOpsTest.scala b/core/src/test/scala/ox/channels/SourceOpsTest.scala index e48d1b6f..85ded9a4 100644 --- a/core/src/test/scala/ox/channels/SourceOpsTest.scala +++ b/core/src/test/scala/ox/channels/SourceOpsTest.scala @@ -209,6 +209,51 @@ class SourceOpsTest extends AnyFlatSpec with Matchers with Eventually { } } + it should "interleave with an empty source" in scoped { + val c1 = Source.fromValues(1, 2, 3) + val c2 = Source.fromValues() + + val s1 = c1.interleave(c2) + + s1.toList shouldBe List(1, 2, 3) + } + + it should "interleave two sources with default segment size" in scoped { + val c1 = Source.fromValues(1, 3, 5) + val c2 = Source.fromValues(2, 4, 6) + + val s = c1.interleave(c2) + + s.toList shouldBe List(1, 2, 3, 4, 5, 6) + } + + it should "interleave two sources with default segment size and different lengths" in scoped { + val c1 = Source.fromValues(1, 3, 5) + val c2 = Source.fromValues(2, 4, 6, 8, 10, 12) + + val s = c1.interleave(c2) + + s.toList shouldBe List(1, 2, 3, 4, 5, 6, 8, 10, 12) + } + + it should "interleave two sources with custom segment size" in scoped { + val c1 = Source.fromValues(1, 2, 3, 4) + val c2 = Source.fromValues(10, 20, 30, 40) + + val s = c1.interleave(c2, segmentSize = 2) + + s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40) + } + + it should "interleave two sources with custom segment size and different lengths" in scoped { + val c1 = Source.fromValues(1, 2, 3, 4, 5, 6, 7) + val c2 = Source.fromValues(10, 20, 30, 40) + + val s = c1.interleave(c2, segmentSize = 2) + + s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40, 5, 6, 7) + } + it should "merge two sources" in { scoped { val c1 = Source.fromValues(1, 2, 3) From 07072e5ed42fc94b2c08878b524b2c8267fbe12b Mon Sep 17 00:00:00 2001 From: Jacek Kunicki Date: Thu, 5 Oct 2023 10:25:43 +0200 Subject: [PATCH 2/5] Move Source.interleave tests to separate class --- .../ox/channels/SourceOpsInterleaveTest.scala | 55 +++++++++++++++++++ .../scala/ox/channels/SourceOpsTest.scala | 45 --------------- 2 files changed, 55 insertions(+), 45 deletions(-) create mode 100644 core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala diff --git a/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala b/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala new file mode 100644 index 00000000..3b23462f --- /dev/null +++ b/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala @@ -0,0 +1,55 @@ +package ox.channels + +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers +import ox.* + +class SourceOpsInterleaveTest extends AnyFlatSpec with Matchers { + + behavior of "Source.interleave" + + it should "interleave with an empty source" in scoped { + val c1 = Source.fromValues(1, 2, 3) + val c2 = Source.fromValues() + + val s1 = c1.interleave(c2) + + s1.toList shouldBe List(1, 2, 3) + } + + it should "interleave two sources with default segment size" in scoped { + val c1 = Source.fromValues(1, 3, 5) + val c2 = Source.fromValues(2, 4, 6) + + val s = c1.interleave(c2) + + s.toList shouldBe List(1, 2, 3, 4, 5, 6) + } + + it should "interleave two sources with default segment size and different lengths" in scoped { + val c1 = Source.fromValues(1, 3, 5) + val c2 = Source.fromValues(2, 4, 6, 8, 10, 12) + + val s = c1.interleave(c2) + + s.toList shouldBe List(1, 2, 3, 4, 5, 6, 8, 10, 12) + } + + it should "interleave two sources with custom segment size" in scoped { + val c1 = Source.fromValues(1, 2, 3, 4) + val c2 = Source.fromValues(10, 20, 30, 40) + + val s = c1.interleave(c2, segmentSize = 2) + + s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40) + } + + it should "interleave two sources with custom segment size and different lengths" in scoped { + val c1 = Source.fromValues(1, 2, 3, 4, 5, 6, 7) + val c2 = Source.fromValues(10, 20, 30, 40) + + val s = c1.interleave(c2, segmentSize = 2) + + s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40, 5, 6, 7) + } +} diff --git a/core/src/test/scala/ox/channels/SourceOpsTest.scala b/core/src/test/scala/ox/channels/SourceOpsTest.scala index 85ded9a4..e48d1b6f 100644 --- a/core/src/test/scala/ox/channels/SourceOpsTest.scala +++ b/core/src/test/scala/ox/channels/SourceOpsTest.scala @@ -209,51 +209,6 @@ class SourceOpsTest extends AnyFlatSpec with Matchers with Eventually { } } - it should "interleave with an empty source" in scoped { - val c1 = Source.fromValues(1, 2, 3) - val c2 = Source.fromValues() - - val s1 = c1.interleave(c2) - - s1.toList shouldBe List(1, 2, 3) - } - - it should "interleave two sources with default segment size" in scoped { - val c1 = Source.fromValues(1, 3, 5) - val c2 = Source.fromValues(2, 4, 6) - - val s = c1.interleave(c2) - - s.toList shouldBe List(1, 2, 3, 4, 5, 6) - } - - it should "interleave two sources with default segment size and different lengths" in scoped { - val c1 = Source.fromValues(1, 3, 5) - val c2 = Source.fromValues(2, 4, 6, 8, 10, 12) - - val s = c1.interleave(c2) - - s.toList shouldBe List(1, 2, 3, 4, 5, 6, 8, 10, 12) - } - - it should "interleave two sources with custom segment size" in scoped { - val c1 = Source.fromValues(1, 2, 3, 4) - val c2 = Source.fromValues(10, 20, 30, 40) - - val s = c1.interleave(c2, segmentSize = 2) - - s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40) - } - - it should "interleave two sources with custom segment size and different lengths" in scoped { - val c1 = Source.fromValues(1, 2, 3, 4, 5, 6, 7) - val c2 = Source.fromValues(10, 20, 30, 40) - - val s = c1.interleave(c2, segmentSize = 2) - - s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40, 5, 6, 7) - } - it should "merge two sources" in { scoped { val c1 = Source.fromValues(1, 2, 3) From d393463ddf4a7c0585b62fdb082d061231a7d61f Mon Sep 17 00:00:00 2001 From: Jacek Kunicki Date: Thu, 5 Oct 2023 11:09:26 +0200 Subject: [PATCH 3/5] Move internal state to fork --- core/src/main/scala/ox/channels/SourceOps.scala | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index 8277b55b..ccda0740 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -197,16 +197,17 @@ trait SourceOps[+T] { this: Source[T] => */ def interleave[U >: T](other: Source[U], segmentSize: Int = 1)(using Ox, StageCapacity): Source[U] = val c = StageCapacity.newChannel[U] - var source: Source[U] = this - var counter = 0 - var neitherCompleted = true - - def switchSource(): Unit = { - if (source == this) source = other else source = this - counter = 0 - } forkDaemon { + var source: Source[U] = this + var counter = 0 + var neitherCompleted = true + + def switchSource(): Unit = { + if (source == this) source = other else source = this + counter = 0 + } + repeatWhile { source.receive() match case ChannelClosed.Done => From a5336ec39a25dcc842fa8802d53e38a95ca3e0dc Mon Sep 17 00:00:00 2001 From: Jacek Kunicki Date: Thu, 5 Oct 2023 11:12:36 +0200 Subject: [PATCH 4/5] Update example for Source.interleave --- core/src/main/scala/ox/channels/SourceOps.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index ccda0740..d6d3f2a1 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -187,12 +187,12 @@ trait SourceOps[+T] { this: Source[T] => * import ox.channels.Source * * scoped { - * val s1 = Source.fromValues(1, 2, 3, 4) + * val s1 = Source.fromValues(1, 2, 3, 4, 5, 6, 7) * val s2 = Source.fromValues(10, 20, 30, 40) * s1.interleave(s2, segmentSize = 2).toList * } * - * scala> val res0: List[Int] = List(1, 2, 10, 20, 3, 4, 30, 40) + * scala> val res0: List[Int] = List(1, 2, 10, 20, 3, 4, 30, 40, 5, 6, 7) * }}} */ def interleave[U >: T](other: Source[U], segmentSize: Int = 1)(using Ox, StageCapacity): Source[U] = From 42a453d1b8b0390497ce257ff5dfe2e93fc664da Mon Sep 17 00:00:00 2001 From: Jacek Kunicki Date: Thu, 5 Oct 2023 12:25:25 +0200 Subject: [PATCH 5/5] Add eagerComplete flag to Source.interleave --- .../main/scala/ox/channels/SourceOps.scala | 17 ++++++++---- .../ox/channels/SourceOpsInterleaveTest.scala | 27 +++++++++++++++++++ 2 files changed, 39 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/ox/channels/SourceOps.scala b/core/src/main/scala/ox/channels/SourceOps.scala index d6d3f2a1..7d113f64 100644 --- a/core/src/main/scala/ox/channels/SourceOps.scala +++ b/core/src/main/scala/ox/channels/SourceOps.scala @@ -169,8 +169,10 @@ trait SourceOps[+T] { this: Source[T] => // /** Sends a given number of elements (determined byc `segmentSize`) from this source to the returned channel, then sends the same number - * of elements from the `other` source and repeats. If one of the sources is closed before the other, the remaining elements from the - * open one are sent to the returned channel. The order of elements in both sources is preserved. + * of elements from the `other` source and repeats. The order of elements in both sources is preserved. + * + * If one of the sources is closed before the other, the behavior depends on the `eagerCancel` flag. When set to `true`, the other source + * is cancelled immediately, otherwise the remaining elements from the other source are sent to the returned channel. * * Must be run within a scope, since a child fork is created which receives from both sources and sends to the resulting channel. * @@ -178,6 +180,9 @@ trait SourceOps[+T] { this: Source[T] => * The source whose elements will be interleaved with the elements of this source. * @param segmentSize * The number of elements sent from each source before switching to the other one. Default is 1. + * @param eagerComplete + * If `true`, the returned channel is completed as soon as either of the sources completes. If 'false`, the remaining elements of the + * non-completed source are sent downstream. * @return * A source to which the interleaved elements from both sources would be sent. * @example @@ -195,7 +200,7 @@ trait SourceOps[+T] { this: Source[T] => * scala> val res0: List[Int] = List(1, 2, 10, 20, 3, 4, 30, 40, 5, 6, 7) * }}} */ - def interleave[U >: T](other: Source[U], segmentSize: Int = 1)(using Ox, StageCapacity): Source[U] = + def interleave[U >: T](other: Source[U], segmentSize: Int = 1, eagerComplete: Boolean = false)(using Ox, StageCapacity): Source[U] = val c = StageCapacity.newChannel[U] forkDaemon { @@ -211,8 +216,10 @@ trait SourceOps[+T] { this: Source[T] => repeatWhile { source.receive() match case ChannelClosed.Done => - // if one source has completed, switch to the other one, otherwise (i.e. if both sources have completed) complete the resulting source - if (neitherCompleted) { + // if one source has completed, either complete the resulting source immediately if eagerComplete is set, or: + // - continue with the other source if it hasn't completed yet, or + // - complete the resulting source if both input sources have completed + if (neitherCompleted && !eagerComplete) { neitherCompleted = false switchSource() true diff --git a/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala b/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala index 3b23462f..8120b3c5 100644 --- a/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala +++ b/core/src/test/scala/ox/channels/SourceOpsInterleaveTest.scala @@ -52,4 +52,31 @@ class SourceOpsInterleaveTest extends AnyFlatSpec with Matchers { s.toList shouldBe List(1, 2, 10, 20, 3, 4, 30, 40, 5, 6, 7) } + + it should "interleave two sources with different lengths and complete eagerly" in scoped { + val c1 = Source.fromValues(1, 3, 5) + val c2 = Source.fromValues(2, 4, 6, 8, 10, 12) + + val s = c1.interleave(c2, eagerComplete = true) + + s.toList shouldBe List(1, 2, 3, 4, 5, 6) + } + + it should "when empty, interleave with a non-empty source and complete eagerly" in scoped { + val c1 = Source.fromValues() + val c2 = Source.fromValues(1, 2, 3) + + val s1 = c1.interleave(c2, eagerComplete = true) + + s1.toList shouldBe empty + } + + it should "interleave with an empty source and complete eagerly" in scoped { + val c1 = Source.fromValues(1, 2, 3) + val c2 = Source.fromValues() + + val s1 = c1.interleave(c2, eagerComplete = true) + + s1.toList shouldBe List(1) + } }