Add Stream#flatMapPar, Stream.flatten and Stream.mergeAll

[iravid]

Resolves #836.

I'm going to add a bunch of more tests that exercise errors, interruption and other issues, but I'd love to get comments on the implementation and possible simplifications.

Pending fixes:

• an error in an inner fiber would enqueue a Take.Fail, which would cause the output stream to fail; however nothing in this case would stop the driver stream
• currently the outer stream ends completely before the inner fibers; which means that an outer stream created using Stream.bracket would release its resources before the inner fibers complete

[regiskuckaertz]

Isn't this redundant with the line below?

[iravid]

supervise is the one that adds a supervisor that interrupts all children; supervised just enables supervision so children are tracked and introspectable by ZIO.children :-)

[regiskuckaertz]

😰 ah yeah, I remember. I have some catching up to do!

[jdegoes]

We probably need better names too... like enableSupervision and cleanupChildren or something...

[iravid]

yeah. I am constantly confused between the two ;-)

[iravid]

Deferring the outer stream's completion to after the inner fibers complete actually seems quite difficult with the current encoding of Fold, because we can't know when the fold has ended.

For example, say we have the following stream:

Stream
  .bracket(acquire)(release) { resource =>
    resource.read
  }
  .mapM { element =>
    process(element).fork
  }
  .foldLeft(List[Fiber[Nothing, Element]]())((fibers, fiber) => fiber :: fibers)
  .flatMap(Fiber.joinAll)

By the time foldLeft returns the list of fibers, the finalizer for the bracket has already run.

If anyone has a clever idea, I'd love to hear it!

[iravid]

I tried some approaches of encoding a "finalization" step into the stream's fold; e.g. something like S => ZIO[R, Nothing, _]. This seems promising and works in the simple combinators (bracket, map) but I haven't explored it further as it has one discouraging facet: when used with a flatMap'd stream, the finalization step would run an arbitrary number of times. Not sure this is a good direction.

Another thought I had: if Stream#fold would return a ZManaged, this could enable the arbitrary manipulation of finalizers.

[jdegoes]

@iravid All right, I began reviewing this today. At a high level, the strategy for flatMapPar seems correct. I'll do a more thorough review tomorrow.

Can you explain this:

currently the outer stream ends completely before the inner fibers; which means that an outer stream created using Stream.bracket would release its resources before the inner fibers complete

Specifically, it's clear this is the case because the data passes through a queue, which will be out-of-sync with the input stream; buffered by some number of elements. But since the child fibers only read from the queue, it should not be an issue; and if it is, we can always avoid exiting the flatMapPar stream until after all the child fibers are done reading from the queue.

Or is there another more fundamental problem that I'm not seeing?

[iravid]

Yeah that’s the fundamental issue. The problem is that I could not find a way to delay the outer stream from completing until the child fibers are empty.

When folding, there’s no way to know that you’re at the last element and join the remaining fibers :-/

[iravid]

Going to revisit these combinators after restructuring ZStream around ZManaged.