如何将 Observable<Observable<any>> 扁平化为 Observable<any>？

Question

我正在尝试通过执行以下操作获得 Observable<MyType>：

return this.cache.get<MyType, number>(storageName, version)
    .map(cached => {
        if (cached !== undefined) {
            return cached;
        }
        return this.http.get<MyType>(`MyTypeGetUrl/${version}`)
            .map(m => m)
            .do(result => {
                this.cache.put<MyType>(storageName, result).first().subscribe();
            });
        }).publishReplay(1).refCount();

cache.get returns 一个 Observable<MyType> 和 http.get<T> returns Observable<T>.

我想检查一个对象的缓存，如果它不存在，从服务器获取它，缓存它并return它。

但是，如果对象存在于缓存中，它returns Observable<MyType>，但如果它需要去服务器，它returns Observable<Observable<MyType>> .

我怎样才能将服务器的结果展平，这样我就可以 return 一个 Observable，只在我必须的时候调用服务器？

Answer 1

而不是 map 使用 mergeMap 并将缓存的结果包装到 Observable 中。在这两种情况下，它都会经过 mergeMap，因此它会将 Observable<MyType> 展开为 MyType（在此示例中，您也可以使用 concatMap 或 switchMap） .

return this.cache.get<MyType, number>(storageName, version)
  .mergeMap(cached => {
    if (cached !== undefined) {
      return Observable.of(cached);
    }
    return this.http.get(...);
  })
  ...

Answer 2

您可以像这样使用 mergeMap 和 Observable.of：

return this.cache.get<MyType, number>(storageName, version)
    .mergeMap(cached => {
        if (cached !== undefined) {
            return Observable.of(cached);
        }
        return this.http.get<MyType>(`MyTypeGetUrl/${version}`)
            .map(m => m)
            .do(result => {
                this.cache.put<MyType>(storageName, result).first().subscribe();
            });
        }).publishReplay(1).refCount();

如何将 Observable<Observable<any>> 扁平化为 Observable<any>？

How to flatten an Observable<Observable<any>> to Observable<any>?

observable

rxjs

rxjs5