如何链接 Observable 和 Promise（Async/Await、RxJS、Observable）答案

【问题标题】：How to chain Observable and Promise (Async/Await, RxJS, Observable)如何链接 Observable 和 Promise（Async/Await、RxJS、Observable）
【发布时间】：2018-02-05 19:10:15
【问题描述】：

我想知道如何在 typescript 中使用 await/async 和 RxJS 做类似的事情：

async function bar(): Promise<boolean>{ ... }

function boo(): Observable<any> { ... }

function baz(): Observable<any> {
  return Observable
         .fromPromise(() => bar())
         .map(() => boo())
}

总而言之，我有多个 baz 流，稍后我将使用 zip 链接它们。在每个 baz() 中，我需要做一些异步检查。但没有任何附加内容...... Observable.fromPromise 正在按预期工作。嘘没有。就像当我订阅流时（zip(baz1(), baz2())），boo 永远不会执行。

我错过了什么吗？这个故事中的最佳实践是什么？

编辑：

这是第一部分的代码：

syncResource(urlEndpoint: string, storeCollectionKey: string): Observable<any> {
    return Observable
        .fromPromise(this.shouldSyncResource(urlEndpoint, storeCollectionKey))
        .map((shouldUpdateResource) => {
            if(!shouldUpdateResource){ return Observable.empty(); }
            // here i can get "shouldUpdateResource" value

            return this._Remote.get(urlEndpoint, this._getRequestOptions())
                .map((response: Response) => response.json())
                .map((collection) => {
                    collection.forEach((item) => {
                        this._Stores[storeCollectionKey].update(item);
                    });
                });
        });
}

这是第二部分：

Observable.zip(
   syncResource('/poultries', 'poultriesStoreKey'),
   syncResource('/buildings', 'buildingsStoreKey')
).subscribe(() => console.log('Syncing done'))

【问题讨论】：

你需要传递一个Promise给fromPromise，而不是一个返回Promise的函数，见reactivex.io/rxjs/class/es6/…
谢谢马丁，但即使我用 .fromPromise(new Promise((resolve, reject) => resolve(true))) “强制”它也不起作用，我可以得到承诺任何情况下的价值回报
你订阅了baz()返回的Observable吗？
是的，稍后在我的代码中我使用 .zip(baz(), baz()).subscribe()
stackoverflow.com/a/43885089/2398593

标签： typescript promise rxjs

【解决方案1】：

好的，解决了。我需要在代码的第二部分使用 mergeMap 和 return a value ：

syncResource(urlEndpoint: string, storeCollectionKey: string): Observable<any> {
    return Observable
        .fromPromise(this.shouldSyncResource(urlEndpoint, storeCollectionKey))
        .mergeMap((shouldUpdateResource, index) => {
            if(!shouldUpdateResource){ return Observable.empty(); }

            return this._Remote.get(urlEndpoint, this._getRequestOptions())
                .do((response: Response) => this.rememberResponseHashIfPresent(response.headers.get('x-hash'), storeCollectionKey))
                .map((response: Response) => {
                    console.log(response.json());
                    return response.json();
                })
                .map((collection) => {
                    collection.forEach((item) => this._Stores[storeCollectionKey].update(item));
                    return true;
                });
        });
}

【讨论】：