RxJS Subscribe with two arguments

Question

I have two observables which I want to combine and in subscribe use either both arguments or only one. I tried .ForkJoin, .merge, .concat but could not achieve the behaviour I'm looking for.

Example:

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return obs1.concat(obs2);
}

Then when using this function:

service.save().subscribe((first, second) => {
    console.log(first); // int e.g. 1000
    console.log(second); // Boolean, e.g. true
});

or

service.save().subscribe((first) => {
    console.log(first); // int e.g. 1000
});

Is there a possibility to get exactly that behaviour?

Hope someone can help!

EDIT:

In my specific use case obs1<int> and obs2<bool> are two different post requests: obs1<int> is the actual save function and obs2<bool> checks if an other service is running.

The value of obs1<int> is needed to reload the page once the request is completed and the value of obs2<bool> is needed to display a message if the service is running - independant of obs1<int>.

So if obs2<bool> emits before obs1<int>, that's not a problem, the message gets display before reload. But if obs1<int> emits before obs2<bool>, the page gets reloaded and the message may not be displayed anymore.

I'm telling this because with the given answers there are different behaviours whether the values get emitted before or after onComplete of the other observable and this can impact the use case.

Answer 1

There are several operators that accomplish this:

CombineLatest

This operator will combine the latest values emitted by both observables, as shown in the marble diagram:

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return combineLatest(obs1, obs2);
}

save().subscribe((val1, val2) => { 
   // logic
});

Zip

The Zip operator will wait for both observables to emit values before emitting one.

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return zip(obs1, obs2);
}

save().subscribe((vals) => { 
   // Note Vals = [val1, val2]
   // Logic
});

Or if you want to use destructuring with the array

save().subscribe(([val1, val2]) => {
   // Logic
});

WithLatestFrom

The WithLatestFrom emits the combination of the last values emitted by the observables, note this operator skips any values that do not have a corresponding value from the other observable.

save: obs1.pipe(withLatestFrom(secondSource))

save().subscribe(([val1, val2]) => {
    // Logic
});

Answer 2

You can use forkJoin for this purpose. Call them parallely and then if either of them is present then do something.

let numberSource = Rx.Observable.of(100);
let booleanSource = Rx.Observable.of(true);

Rx.Observable.forkJoin(
  numberSource,
  booleanSource
).subscribe( ([numberResp, booleanResp]) => {
  if (numberResp) {
    console.log(numberResp);
    // do something
  } else if (booleanResp) {
    console.log(booleanResp);
    // do something
  }
});

Answer 3

You may use the zip static method instead of concat operator.

save(): Observable<any> {
   return zip(obs1, obs2);
}

Then you should be able to do like the following:

service.save().subscribe((x) => {
    console.log(x[0]); // int e.g. 1000
    console.log(x[1]); // Boolean, e.g. true
});

Answer 4

The exact operator to use depends on the specific details of what you are trying to solve.

A valid option is to use combineLatest - Docs:

obs1$: Observable<int>;
obs2$: Observable<Boolean>;
combined$ = combineLatest(obs1$, obs2$);

combined$.subscribe(([obs1, obs2]) => {
  console.log(obs1);
  console.log(obs2);
})

Answer 5

Concat emits two events through the stream, one after the other has completed, this is not what you're after.

Merge will emit both events in the same manner, but in the order that they actually end up completing, also not what you're after.

What you want is the value of both items in the same stream event. forkJoin and zip and combineLatest will do this, where you're getting tripped up is that they all emit an array of the values that you're not accessing properly in subscribe.

zip emits every time all items zipped together emit, in sequence, so if observable 1 emits 1,2,3, and observable two emits 4,5; the emissions from zip will be [1,4], [2,5].

combineLatest will emit everytime either emits so you'll get soemthing like [1,4],[2,4],[2,5],[3,5] (depending on the exact emission order).

finally forkJoin only emits one time, once every item inside it has actually completed,a and then completes itself. This is likely what you want more than anything since you seem to be "saving". if either of those example streams don't complete, forkJoin will never emit, but if they both complete after their final value, forkjoin will only give one emission: [2,5]. I prefer this as it is the "safest" operation in that it guarantees all streams are completing properly and not creating memory leaks. And usually when "saving", you only expect one emission, so it is more explicit as well. When ever you see forkJoin, you know you're dealing with a single emission stream.

I would do it like this, personally:

obs1: Observable<int>;
obs2: Observable<Boolean>;

save(): Observable<any> {
   return forkJoin(obs1, obs2);
}

service.save().subscribe(([first, second]) => {
    console.log(first); // int e.g. 1000
    console.log(second); // Boolean, e.g. true
});

Typescript provides syntax like this to access the items in an array of a known length, but there is no way to truly create multiple arguments in a subscribe success function, as it's interface only accepts a single argument.