在恢复之前等待 RxJs.Subscriptions 完成答案

【问题标题】：Wait on RxJs.Subscriptions to finish before resuming在恢复之前等待 RxJs.Subscriptions 完成
【发布时间】：2018-02-13 13:35:54
【问题描述】：

在我的 Angular 2 应用程序中，我需要发出一系列 http 请求。我有两个服务，A 和 B，每个都发出请求，A.get() 和 B.get()，它们从 API 获取数据并将它们存储在他们的服务中。这两个可以同时调用，但是我有第三个请求doSomething()，这取决于A.get() 和B.get() 的结果。由于A.get() 和B.get() 都在本地存储它们的响应，它们的返回值最终是RxJs 订阅。像这样：

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

我的组件看起来像这样：

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

我的问题是doSomething() 失败，因为a.get() 和b.get() 已经完成了http 请求。我需要一种方法来保持呼叫doSomething()，直到我的其他呼叫完成。我已经搜索了所有内容，但对这个问题没有任何运气。 RxJs 文档提供了几种可以合并 Observables 的方法，但这不是我在这种情况下所拥有的。

【问题讨论】：

标签： angular rxjs

【解决方案1】：

嗯，你想要的可以用这个来实现：

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

这是对您的代码的修改。如您所见，无需订阅服务组件内的可观察对象。甚至不需要在服务组件之外单独订阅它们。我们可以在单个最终的 subscribe() 中直接接收最终结果，只需通过某种方式组合 observables。

更新

do() 和 subscribe() 有什么区别。

稍微简化一下（跳过热的可观察对象），可观察管道在您订阅它之前不会开始做任何事情。 do() 只是众多旨在在您的可观察对象链中产生一些“副作用”的运算符之一，例如，它可以在可观察管道的中间输出一些中间结果到控制台以进行调试。这是主要区别。所以，你可以在 subscribe() 中不带 do() 在中间得到一些东西，但你不会在 do() 中不带 subscribe() 得到任何东西。

【讨论】：

感谢@Alexander，效果很好！。一个后续问题，.do 和 .subscribe 有什么区别。在 Observable 上调用 .do 会丢失任何功能吗？
我没有得到这部分： if (this.data) { return Observable.of(this.data); }
这意味着，当它被定义（有值）返回它时，另一个 observable 会给它一个值。

【解决方案2】：

使用Observable.forkJoin 将您的订阅移至组件，轻松解决您的问题。它需要一组冷可观察对象，订阅后，您会收到一组结果。像这样的：

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

在两个请求都进来之前，订阅中的回调不会被调用。

如果a.get() 或b.get() 失败，则会调用错误方法

【讨论】：