I have two observables which I want to bine 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 pleted 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.

I have two observables which I want to bine 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 pleted 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.

• javascript
• angular
• typescript
• rxjs

• what version of rxjs are you using? – J. Pichardo Commented May 22, 2018 at 14:42
• 1 rxjs version 6.1.0 – knnhcn Commented May 22, 2018 at 14:51
• Maybe useful - see this RxJs v6 solution using zip & param destructuring on the subscribe function - stackoverflow./a/55993246/1882064 – arcseldon Commented May 5, 2019 at 15:00

5 Answers 5

There are several operators that acplish this:

CombineLatest

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

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

save(): Observable<any> {
   return bineLatest(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 bination 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
});

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
  }
});

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
});

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

A valid option is to use bineLatest - Docs:

obs1$: Observable<int>;
obs2$: Observable<Boolean>;
bined$ = bineLatest(obs1$, obs2$);

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

Concat emits two events through the stream, one after the other has pleted, 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 pleting, also not what you're after.

What you want is the value of both items in the same stream event. forkJoin and zip and bineLatest 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].

bineLatest 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 pleted,a and then pletes itself. This is likely what you want more than anything since you seem to be "saving". if either of those example streams don't plete, forkJoin will never emit, but if they both plete 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 pleting 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.