feat: introduce `intersperse` function #18

geminicaprograms · 2023-10-11T13:58:13Z

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", "]")

core/src/main/scala/ox/channels/SourceOps.scala

rucek · 2023-10-12T07:44:41Z

core/src/main/scala/ox/channels/SourceOps.scala

+    *   An element to be injected between the stream elements.
+    * @return
+    *   A source, onto which elements will be injected.
+    * @example


suggestion: Please add the necessary imports to the example (probably import ox.* and import ox.channels.Source) so that it's copy-paste'able (also in the other variant of intersperse)

🥇 Updated examples and confirmed in Scala Worksheet file

rucek · 2023-10-12T07:48:19Z

core/src/main/scala/ox/channels/SourceOps.scala

+    val c = StageCapacity.newChannel[U]
+    forkDaemon {
+      var firstEmitted = false
+      start.foreach(c.send(_))


nitpick: Could be just start.foreach(c.send), also applies to case ChannelClosed.Done

question: In the two other places we only continue processing when send succeeds, i.e. when send(...).isValue is true - shouldn't this be the case here as well?

Could be just start.foreach(c.send), also applies to case ChannelClosed.Done

👍 method reference rulez - fixed

question: In the two other places we only continue processing...

That is necessary to properly implement the situation when sink cannot accept anything anymore, right? Fixed.

rucek · 2023-10-12T07:53:03Z

core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala

+class SourceOpsIntersperseTest extends AnyFlatSpec with Matchers {
+  behavior of "Source.intersperse"
+
+  it should "intersperse with inject only over an empty source" in {


nitpick: You can avoid one level of the curly braces by writing

it should "..." in scoped { ... }

rucek · 2023-10-12T07:56:51Z

core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala

+      s.receive() shouldBe "foo"
+      s.receive() shouldBe ", "
+      s.receive() shouldBe "bar"
+      s.receive() shouldBe ChannelClosed.Done


suggestion: If you want to receive all elements from the source, you can use s.toList and make a single assertion on the list instead of aseerting individual receives.

adamw · 2023-10-12T13:12:37Z

core/src/main/scala/ox/channels/SourceOps.scala

+  private def intersperse[U >: T](start: Option[U], inject: U, end: Option[U])(using Ox, StageCapacity): Source[U] =
+    val c = StageCapacity.newChannel[U]
+    forkDaemon {
+      if (start.map(c.send(_).isValue).getOrElse(true))


we just created the c channel, it can't bet done/errored (we're the only ones sending to it)

That would be true as long as the downstream has no way to close the channel (which is the current state). So the question is whether we plan to change this at some point soon - if so, this code would already handle such case, otherwise it's YAGNI.

Since the "repeat while send succeeds" pattern originates from treating SourceOps.map as a reference implementation. Thus if we decide to keep the downstream unable to close the channel, we should probably also update some of the recently added combinators so that they don't use isValue.

@rucek yes, that's a good question, but I think it's the current design's assumption, that the streams are one-way - data always flows downstream. There's no backchannel which would allow propagating information. There are two ways of implementing such a backchannel, if needed:

(a) when the downstream receives an error/done, simply close the scope. This will close all intermediate components. My suspicion is that this will cover 90% of the use-cases, but I might be proven wrong of course. But so far I didn't come up with a compelling use-case to actually have the backchannel, in presence of structured concurrency & scopes
(b) for the other 10%, you can implement a backchannel using a channel. Then, in you original source or some intermediate component you can do select(backchannel.receiveClause, s.sendClause(x)) to send data while being informed if anything is sent on the backchannel

The docs kind of try to communicate the same as above, but maybe they need extending :)

adamw · 2023-10-12T13:13:10Z

core/src/main/scala/ox/channels/SourceOps.scala

+          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).isValue


same here, we know that the channel can't be closed

rucek · 2023-10-16T07:16:31Z

core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala

+    val c = Channel[String]()
+    fork {
+      c.done()
+    }


nitpick: This is equivalent to Source.empty[String]

rucek · 2023-10-16T07:17:46Z

core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala

+    val c = Channel[String]()
+    fork {
+      c.send("foo")
+      c.done()
+    }


nitpick: This is equivalent to Source.fromValues("foo"), you could also use fromValues in the other test cases.

geminicaprograms · 2023-10-17T11:19:19Z

core/src/main/scala/ox/channels/SourceOps.scala

@@ -34,6 +34,67 @@ trait SourceOps[+T] { this: Source[T] =>
    }
    c2

+  /** Intersperses stream with provided element.


nitpick: Updated function description to match the other operators

core/src/main/scala/ox/channels/SourceOps.scala

core/src/test/scala/ox/channels/SourceOpsIntersperseTest.scala

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", "]")

geminicaprograms requested review from adamw and rucek October 11, 2023 13:58

geminicaprograms self-assigned this Oct 11, 2023

geminicaprograms commented Oct 11, 2023

View reviewed changes

core/src/main/scala/ox/channels/SourceOps.scala Outdated Show resolved Hide resolved

geminicaprograms requested a review from kciesielski October 11, 2023 14:24

geminicaprograms force-pushed the feat_intersperse branch from 04e57f4 to d77e952 Compare October 12, 2023 06:27

kciesielski approved these changes Oct 12, 2023

View reviewed changes

rucek reviewed Oct 12, 2023

View reviewed changes

geminicaprograms force-pushed the feat_intersperse branch from d77e952 to 767a962 Compare October 12, 2023 10:47

geminicaprograms requested review from rucek and kciesielski October 12, 2023 10:49

adamw reviewed Oct 12, 2023

View reviewed changes

rucek mentioned this pull request Oct 12, 2023

Add Source.mapStateful and Source.mapStatefulConcat combinators #15

Merged

rucek reviewed Oct 16, 2023

View reviewed changes

geminicaprograms force-pushed the feat_intersperse branch from 767a962 to d16440d Compare October 16, 2023 11:48

geminicaprograms requested review from rucek and adamw October 16, 2023 11:49

rucek approved these changes Oct 17, 2023

View reviewed changes

geminicaprograms commented Oct 17, 2023

View reviewed changes

geminicaprograms force-pushed the feat_intersperse branch from d16440d to a4de618 Compare October 17, 2023 11:20

geminicaprograms requested a review from rucek October 17, 2023 11:20

adamw merged commit 23be5b5 into master Oct 17, 2023
4 checks passed

adamw deleted the feat_intersperse branch February 21, 2024 15:21

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: introduce `intersperse` function #18

feat: introduce `intersperse` function #18

geminicaprograms commented Oct 11, 2023

rucek Oct 12, 2023

geminicaprograms Oct 12, 2023

rucek Oct 12, 2023

rucek Oct 12, 2023

geminicaprograms Oct 12, 2023

geminicaprograms Oct 12, 2023

rucek Oct 12, 2023

geminicaprograms Oct 12, 2023

rucek Oct 12, 2023

geminicaprograms Oct 12, 2023

adamw Oct 12, 2023

rucek Oct 12, 2023

adamw Oct 12, 2023

geminicaprograms Oct 16, 2023

adamw Oct 12, 2023

geminicaprograms Oct 17, 2023

rucek Oct 16, 2023

geminicaprograms Oct 16, 2023

rucek Oct 16, 2023

geminicaprograms Oct 16, 2023

geminicaprograms Oct 17, 2023

feat: introduce intersperse function #18

feat: introduce intersperse function #18

Conversation

geminicaprograms commented Oct 11, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

feat: introduce `intersperse` function #18

feat: introduce `intersperse` function #18