RxJS merge when previous stream receives first value

0
votes

Short explanation of what I want:

I have an array of observables and I want the results of all of them to come through the same stream. Initially I want the first observable to be activated, and as soon as an observable receives its first value, I want the next observable to be activated.

Long explanation of what I want:

The solution I am looking for lies somewhere between the RxJS creation operators “merge” and “concat".

I have an array of observables. Each observable will have several emissions over time. I want to queue these observables after each other so that initially only the first observable is activated (this is similar to how “concat” works).

Then, as soon as the first observable receives its first value I want the second observable to be activated. As soon as the second observable receives its first value I want the third observable to be activated, and so on. (This is different from “concat”. “concat” waits for the previous observable to be finished, but in my use case I want to wait for the previous observable to have received it’s first value)

There will be only one resulting stream that emits the values from all the activated observables (this is similar to how “merge” works).

I do not think there is a specific RxJS operator for this, but I hope a solution can be found by mixing multiple operators.

2
javascriptrxjs

2 Answers

1
votes

as soon as the first observable receives its first value

Observables don't receive values, subscribers do; this distinction is important here, because it matters at which point we subscribe to each observable. For example, if we don't think of this in terms of subscribers, then

merge(
  a$,
  a$.pipe(first(), switchMapTo(b$)),
  b$.pipe(first(), switchMapTo(c$)),
)

would be a solution – but my assumption here is that this isn't what you meant. It does get us closer to a solution, though. What we'd need instead is this:

merge(
  a$,
  a$.pipe(first(), switchMapTo(b$)),
  a$.pipe(first(), switchMapTo(b$.pipe(first(), switchMapTo(c$)))),
  // …
)

Of course we need to find a generic solution for this. A naive approach would simply be to "recurse":

const activatedMerge = (sources) => {
  const activatedSources = sources.reduce((result, source, idx) => {
    result.push(idx === 0 ? source : result[idx-1].pipe(first(), switchMapTo(source)));
    return result;
  }, []);

  return merge(...activatedSources);
};

activatedMerge([a$, b$, c$]).subscribe(console.log);

There might well be a better solution than this, unfortunately I ran out of time. Hope this helps anyway.

1
votes

Cascade Merge

This should do what you describe, it just recursively merges a new stream each time a first value is seen from the previous stream.

function cascadeMerge<T>(...observables: Observable<T>[]): Observable<T>{
  if(observables.length < 1) return EMPTY;
  return observables[0].pipe(
    mergeMap((v,i) => {
      if(i === 0 && observables.length > 1){
        return concat(
          of(v),
          cascadeMerge(...observables.slice(1))
        )
      }
      return of(v);
    })
  );
}

The benefit of this over something like:

merge(
  a$,
  a$.pipe(first(), switchMapTo(b$))
)

is that you only subscribe to each stream once. You don't have to worry about hot vs cold observables or if you're multicasting.

cascadeMerge(a$, b$, c$).subscribe(console.log);