用 observable 替换回调地狱

Question

RxJS github repo 解释了如何从事件或数组创建 observable。我知道如何用 async 或 Promise 替换回调地狱，但我找不到关于如何为执行异步任务的函数创建和返回 observable 的示例。

For example,

x = getData();
y = getMoreData(x);
z = getMoreData(y);
...

getData(function(x){
    getMoreData(x, function(y){
        getMoreData(y, function(z){ 
            ...
        });
    });
});

如何用 observables 替换这个回调地狱？我发现我们可以在RxJS github - creating observable 中调用observer.next() 方法，但无法为这个示例找到一个实现。

Answer 1

您可以使用flatMap 运算符来链接结果。看看这里：RxJS Promise Composition (passing data)。基本上链接承诺与链接flatMap 相同。那就是：

pl().then(p2).then(p3).then(console.log);

类似于：

 Rx.Observable.just()
         .flatMap(p1)
         .flatMap(p2)
         .flatMap(p3);

所以从 Promise 到 Observable 的转换很简单。如果你有一个使用回调而不是 Promise 操作的函数，我可以想到两个选项：

• 尝试使用Rx.Observable.fromCallback 或Rx.Observable.fromNodeCallback
• 将回调包装在您自己的 observable 中。也看看这里：rx.js how to chain observables

例如，function asyncCall (param, cb) 会导致类似：

Rx.Observable.create (function(observer){
  asyncCall(param, function cb(err, data){
    if (err) {observer.onError(err)}
    else {
      observer.onNext(x);
      // The following is supposing your callback is only called once, so there is no more data
      observer.onCompleted();
    }
  }
})

完成后，您可以使用flatMap（或concatMap，如果执行顺序很重要），如前所示。