Add haltStrategy option to zipLatest family of Stream combinators

[Avaq]

What is the problem this feature would solve?

I use zipLatestWith extensively in my FRP to combine two reactive states into one. Sometimes, I'm zipping a finite data stream with an infinite reactive state stream (such as the ones created by SubscriptionRef). In those cases, I want my data stream to control when the resulting zipped stream ends.

What is the feature you are proposing to solve the problem?

  const data = Stream.make(1, 2, 3, 4, 5).pipe(Stream.schedule(Schedule.spaced('1 second')));
  const state = yield* SubscriptionRef.make(2);

  return Stream.zipLatestWith(
    data,
    state.changes,
    (x, y) => x * y,
    {
      haltStrategy: HaltStrategy.Left
//    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
//      This is what I'm proposing.
    }
  );

What alternatives have you considered?

I've tried doing some wacky stuff, like:

const data = Stream.make(1, 2, 3, 4, 5).pipe(Stream.schedule(Schedule.spaced('1 second')));
const state = yield* SubscriptionRef.make(2);
const sharedData = yield* Stream.share(data);

return Stream.zipLatestWith(
  sharedData,
  state.changes.pipe(Stream.haltWhen(
    Stream.runDrain(sharedData)
  )),
  (x, y) => x * y,
);

...and some equally hacky variations, but none of it works.

[Avaq]

Oh and I'm not sure if there are other combinators like the merge* and zipLatest* families, where two streams are combined in parallel, in real-time. But if there are, these combinators would all be good candidates for a haltStrategy option I think.

Maybe actually flatMap with switch or concurrency is one? Where a haltingStrategy could be used to determine whether the returned stream halts if the input ends, or halts when the stream returned from the mapper ends, or either or both.

[Avaq]

Generalizing even further: Any situation where a Failure signal can short-circuit the combination of two streams would be a situation where the user may want to pass a haltingStrategy option to also potentially short-circuit when one of the input streams end.