RxJS 在回调中合并 Observable

Question

我需要一些帮助来让函数返回一个可观察对象。我有一个函数（我们称之为mainFunction），它调用一个异步函数（我们称之为getAsyncWithCallback）并执行一个回调。回调的一部分是一个异步调用，它返回一个 observable（我们称之为getData）。

getAsyncWithCallback 是库的一部分，我无法更改它。

这就是我所拥有的：

mainFunction(){
    getAsyncWithCallback(() => {
        let myObservable = getData();
    });
}

这就是我想做的：

mainFunction().subscribe((data) => { 
    // data is the data from getData()
});

实现这一点的最佳方法是什么，并在内部链接错误和完成？

Answer 1

我的解决方案：

mainFunction(): Observable<any> {

    return Rx.Observable.create(observer => {
        getAsyncWithCallback((): void => {
            getData().subscribe(observer);
        })
    })
}

mainFunction().subscribe((data) => {
    console.log(data);
})); 

Answer 2

您将希望从mainFunction 返回一个 Observable。

如果 getData 返回一些不是 Observable 的其他类型的对象，您将使用 Observable.create 构造自己的 Observable：

function mainFunction() {
  return Observable.create((obs) => {
    getAsyncWithCallback(() => {
      getData()
        .subscribe((data) => { obs.next(data); })
        .catch((err) => { obs.error(err) });
    });
  });
}

mainFunction().subscribe(data => console.log(data));

来自文档：

create 将 onSubscription 函数转换为实际的 Observable。每当有人订阅该 Observable 时，都会使用 Observer 实例作为第一个也是唯一的参数调用该函数

用一个值调用 next 会将该值发送给观察者。

大多数时候您不需要使用 create，因为现有的操作符允许您为大多数用例创建 Observable。话虽如此，create 是一种低级机制，如果您有非常特定的需求，您可以创建任何 Observable。

还有Observable.from，它将从数组、类数组对象、迭代器、类 Observable 对象和 Promise 创建 Observable。