Wait on RxJs.Subscriptions to finish before resuming

Question

In my Angular 2 app I need to make a series of http request. I have two services, A & B, that each make request, A.get() and B.get(), that gets data from the API and stores them in the their service. These two can be called at the same time, however I have a third request doSomething() that depends on the results of A.get() and B.get(). Since both A.get() and B.get() are storing their responses locally their return value ends up be the RxJs Subscription. Like so:

class A{
  public data;
  public get(){
    return api.call(params).subscribe((response)=>{ this.data = response.json();})
  }
}

class B{
  public data;
  public get(){
    return api.call(params).subscribe((response)=>{ this.data = response.json();})
  }
}

My component looks something like this:

class MyComponent{
  constructor(private a: A, private b: B){
    a.get();
    b.get();
    this.doSomething(a.data, b.data);
  }
  doSomething(aData, bData){
    ...
  }
}

My problem is doSomething() is failing because a.get() and b.get() have completed there http request yet. I need a way to hold of calling doSomething() until my other calls have completed. I've search all over but haven't had any luck with this issue. RxJs documentation gives a few ways you can merge Observables but that is not what I have in this case.

Answer 1

Well, what you want can be achieved with this:

class A {

    private data;

    public get(): Observable<any> {
        // this is a very primitive caching just to show concept
        if (this.data) {
            return Observable.of(this.data);
        }
        // In real life it would be something like this:
        //  let call = this.http.get(...).map(r => r.json())
        let call = Observable.of("some value A");
        return call.do(s => this.data = s);
    }

}

class B {

    private data;

    public get(): Observable<any> {
        if (this.data) {
            return Observable.of(this.data);
        }
        let call = Observable.of("some value B");
        return call.do(s => this.data = s);
    }

}

class MyComponent {
    constructor(private a: A, private b: B) {
        Observable
            .zip(this.a.get(), this.b.get(), (a: any, b: any) => { return { a: a, b: b } })
            .subscribe((r) => {
                this.doSomething(r.a, r.b);
            });
    }
    doSomething(aData, bData) {
        console.log("aData", aData);
        console.log("bData", bData);
    }
}

This is a modification of your code. As you see there's no need to subscribe to observables inside the service components. There's no even need to subscribe to them separately outside of service components. We can directly receive the ultimate result in the single final subscribe() just by combining observables in a certain way.

UPDATE

What is the difference between do() and subscribe().

Simplifying a bit (skipping hot observables), observable pipe will not start to do anything until you subscribe to it. do() is just one of many operators designed to have some "side effects" in the chain of your observables, for example it can output some intermediate results in the middle of observable pipe to console for debugging purposes. This is the main difference. So, you can get something in subscribe() without do() in the middle, but you will not get anything in do() without subscribe().

Answer 2

Use Observable.forkJoin to easily solve your problem by moving your subscription to the component. It takes an array of cold observables, and after subscribing, you receive an array of the results. Something like this:

Observable.forkJoin([ 
    a.get(),
    b.get()
]).subscribe(
    results => {
        doSomething(results[0], results[1]);
    },
    err => {
        //handle error
    }
);

The callback in the subscription will not get called until both requests come in.

If either a.get() or b.get() fails, the error method will get called