Should Pekko stream support for expression?

[He-Pin]

Motivation:
Both ZIO and FS2 support this, and which is easier for simple job.

Prons:
Simpler and easier to use.

Cons:
Easy to get unexpected result.

Changes needed:

1. add flatmap === flatmapConcat
2. add foreach === to(Sink.foreach(f), Keep.Right)
3. add withFilter === filter

class SourceForSpec extends StreamSpec with DefaultTimeout {
  "be able to use for" in {
    val sa = Source(List(1, 2, 3, 4, 5))
    val source = for {
      a <- sa if a % 2 == 0
    } yield a

    val graph = for (i <- source) {
      println(i)
    }
    graph.run
  }
}

[mdedetrich]

Im open to exploring this even if its usecase is limited because we are only abstracting over 3 operators (as you have stated). The unexpected result is definitely something we need to look into, especially withFilter can be problematic

[He-Pin]

I just looked into the fs2 , just a simple:

  /** Alias for [[filter]]
    * Implemented to enable filtering in for comprehensions
    */
  def withFilter(f: O => Boolean) = this.filter(f)

[mdedetrich]

For me its more about evaluation semantics but I may just be overthinking things

[jxnu-liguobin]

I like it

[gmethvin]

I don't like the idea of implementing foreach as runForEach. It's strange to have a way to produce a side effect that isn't directly calling a run method. Also you'll typically want to do something with the future returned by runForEach.

[gmethvin]

It doesn't need to be terminal. The user still might want to control the demand from the sink (e.g. using throttle) or keep track of when elements have been processed.

[He-Pin]

Then that will be different with ZIO & FS2

[mdedetrich]

It doesn't need to be terminal. The user still might want to control the demand from the sink (e.g. using throttle) or keep track of when elements have been processed.

I agree with @He-Pin on this point, if we are going to implement foreach it should be terminal (but still lazy), that's the expectation for foreach.

[gmethvin]

I'm assuming in that case we would only implement foreach on Source not on Flow? Or should Flow#foreach return a Sink?

In any case, I feel like we want to think through the semantics of foreach before we implement it solely for the purpose of supporting for comprehensions.

[He-Pin]

@gmethvin @mdedetrich I took a look in zio and fs2

• in ZIO: foreach == runForeach
• in FS2: foreach == return another Stream[T], and needs compile to run.

if we follow the fs2, then we will need(for(...) { ... }).run(...)

[He-Pin]

I took another look at fs2's stream and I see no reason why we can't support it.

If people want more control they can just leave the for synatex

And just reviewed zio too, I will prepare a PR then, there is no harm, the scaldsl should be more scala not javalish as lightbend's akka.

[mdedetrich]

Go ahead, we can discuss the semantics in the PR

[He-Pin]

added in #935