3.13.0-M8 Stream.pauseWhen changed behavior when using mergeHaltL

[fede0664]

It seems that Stream.pauseWhen changed behavior (stream1 not pausing when pauseSignal is true) on 3.13.0-M8
This code runs ok on 3.13.0-M7 (but not on 3.13.0-M8), stream1 pauses when pauseSignal is true:

stream1
  .pauseWhen(pauseSignal)
  .mergeHaltL(stream2)

On 3.13.0-M8 this code needs to be changed to the following in order to pause stream1:

stream1
  .pauseWhen(pauseSignal)
  .noneTerminate
  .mergeAndAwaitDownstream(stream2.map(Some(_)))
  .unNoneTerminate

Is this the expected behavior since the merge of #3610?

[AdamChlupacek]

Thank you for the report.

This behaviour is expected. merge now runs the streams in parallel without waiting. From your example, I assume that something downstream is setting the pauseSignal; however, merge always pulls up to one additional chunk from the source stream. As such, merge can cache up to one chunk from stream1 before the pauseSignal starts preventing emission.

If you do not want to pull data from stream1 before the work downstream is finished, it is correct to use mergeAndAwaitDownstream.

Also, note that pauseSignal currently does not interrupt when it is already waiting for data. When it emits any data, it then checks if it can emit more; if so, it will start waiting for more data from the source stream even if the pauseSignal is set to true immediately after that check.

[fede0664]

Understood. Thanks for the explanation.
Is it possible to implement mergeHaltLAndAwaitDownstream and other variants for mergeAndAwaitDownstream?

[AdamChlupacek]

I agree that these two use cases, even though they are almost the same, have a different usage and hence would be beneficial to have same combinators.

@mpilquist if you are fine with extending the library surface, I can implement these.

[mpilquist]

yep