Have withLatestFrom wait until all sources have produced one value

Question

I'm making use of the withLatestFrom operator in RxJS in the normal way:

var combined = source1.withLatestFrom(source2, source3);

...to actively collect the most recent emission from source2 and source3 and to emit all three value only when source1 emits.

But I cannot guarantee that source2 or source3 will have produced values before source1 produces a value. Instead I need to wait until all three sources produce at least one value each before letting withLatestFrom do its thing.

The contract needs to be: if source1 emits then combined will always eventually emit when the other sources finally produce. If source1 emits multiple times while waiting for the other sources we can use the latest value and discard the previous values. Edit: as a marble diagram:

--1------------2---- (source)
----a-----b--------- (other1)
------x-----y------- (other2)
------1ax------2by--


--1------------2---- (source)
------a---b--------- (other1)
--x---------y------- (other2)
------1ax------2by--


------1--------2---- (source)
----a-----b--------- (other1)
--x---------y------- (other2)
------1ax------2by--

I can make a custom operator for this, but I want to make sure I'm not missing an obvious way to do this using the vanilla operators. It feels almost like I want combineLatest for the initial emit and then to switch to withLatestFrom from then on but I haven't been able to figure out how to do that.

Edit: Full code example from final solution:

var Dispatcher = new Rx.Subject();
var source1 = Dispatcher.filter(x => x === 'foo');
var source2 = Dispatcher.filter(x => x === 'bar');
var source3 = Dispatcher.filter(x => x === 'baz');

var combined = source1.publish(function(s1) {
    return source2.publish(function(s2) {
        return source3.publish(function(s3) {
            var cL = s1.combineLatest(s2, s3).take(1).do(() => console.log('cL'));
            var wLF = s1.skip(1).withLatestFrom(s2, s3).do(() => console.log('wLF'));

            return Rx.Observable.merge(cL, wLF);
        });
    });
});

var sub1 = combined.subscribe(x => console.log('x', x));

// These can arrive in any order
// and we can get multiple values from any one.
Dispatcher.onNext('foo');
Dispatcher.onNext('bar');
Dispatcher.onNext('foo');
Dispatcher.onNext('baz');

// combineLatest triggers once we have all values.
// cL
// x ["foo", "bar", "baz"]

// withLatestFrom takes over from there.
Dispatcher.onNext('foo');
Dispatcher.onNext('bar');
Dispatcher.onNext('foo');
// wLF
// x ["foo", "bar", "baz"]
// wLF
// x ["foo", "bar", "baz"]

Answer 1

I wasn't quite satisfied with the accepted answer, so I ended up finding another solution. Many ways to skin a cat!

My use-case involves just two streams - a "requests" stream and a "tokens" stream. I want requests to fire as soon as they are received, using the whatever the latest token is. If there is no token yet, then it should wait until the first token appears, and then fire off all the pending requests.

I wasn't quite satisfied with the accepted answer, so I ended up finding another solution. Essentially I split the request stream into two parts - before and after first token arrives. I buffer the first part, and then re-release everything in one go once I know that the token stream is non-empty.

const first = token$.first()

Rx.Observable.merge(
  request$.buffer(first).mergeAll(),
  request$.skipUntil(first)
)
  .withLatestFrom(token$)

See it live here: https://rxviz.com/v/VOK2GEoX

Answer 2

I had similar requirements but for just 2 observables. I ended up using switchMap+first:

observable1
 .switchMap(() => observable2.first(), (a, b) => [a, b])
 .subscribe(([a, b]) => {...}));

So it:

• waits until both observables emit some value
• pulls the value from second observable only if the first one has changed (unlike combineLatest)
• doesn't hang subscribed on second observable (because of .first())

In my case, second observable is a ReplaySubject. I'm not sure if it will work with other observable types.

I think that:

• flatMap would probably work too
• it might be possible to extend this approach to handle more than 2 observables

I was surprised that withLatestFrom will not wait on second observable.

Answer 3

Use combineLatest and filter to remove tuples before first full set is found then set a variable to stop filtering. The variable can be within the scope of a wrapping defer to do things properly (support resubscription). Here it is in java (but the same operators exist in RxJs):

Observable.defer(
    boolean emittedOne = false;
    return Observable.combineLatest(s1, s2, s3, selector)
        .filter(x -> {
            if (emittedOne) 
               return true;
            else {
               if (hasAll(x)) {
                   emittedOne = true;
                   return true;
               } else 
                   return false; 
            }
        });
)

Answer 4

I wanted a version where tokens are fetched regularly - and where I want to retry the main data post on (network) failure. I found shareReplay to be the key. The first mergeWith creates a "muted" stream, which causes the first token to be fetched immediately, not when the first action arrives. In the unlikely event that the first token will still not be available in time, the logic also has a startWith with an invalid value. This causes the retry logic to pause and try again. (Some/map is just a Maybe-monad):

Some(fetchToken$.pipe(shareReplay({refCount: false, bufferSize: 1})))
    .map(fetchToken$ =>
        actions$.pipe(
            // This line is just for starting the loadToken loop immediately, not waiting until first write arrives.
            mergeWith(fetchToken$.pipe(map(() => true), catchError(() => of(false)), tap(x => loggers.info(`New token received, success: ${x}`)), mergeMap(() => of()))),
            concatMap(action =>
                of(action).pipe(
                    withLatestFrom(fetchToken$.pipe(startWith(""))),
                    mergeMap(([x, token]) => (!token ? throwError(() => "Token not ready") : of([x, token] as const))),
                    mergeMap(([{sessionId, visitId, events, eventIds}, token]) => writer(sessionId, visitId, events, token).pipe(map(() => <ISessionEventIdPair>{sessionId, eventIds}))),
                    retryWhen(errors =>
                        errors.pipe(
                            tap(err => loggers.warn(`Error writing data to WG; ${err?.message || err}`)),
                            mergeMap((_error: any, attemptIdx) => (attemptIdx >= retryPolicy.retryCount ? throwError(() => Error("It's enough now, already")) : of(attemptIdx))), // error?.response?.status  (int, response code) error.code === "ENOTFOUND" / isAxiosError: true / response === undefined
                            delayWhen(attempt => timer(attempt < 2 ? retryPolicy.shortRetry : retryPolicy.longRetry, scheduler))
                        )
                    )
                )
            ),
        )
    )

Thanks to everyone on this question-page for good inputs.