解决可观察到无法正常工作的多个承诺

Question

我正在使用firebase存储，并且正在尝试通过函数调用加载所有资产。获取资产URL的唯一方法是呼叫getDownloadurl返回承诺。我需要为每项资产打电话给它，但由于某种原因，我无法等待所有承诺要等待。

我认为从Mergemap返回承诺会让它等待所有人，但事实并非如此。

我看了很多有关承诺和RXJ的问题，但我似乎无法弄清楚该代码出了什么问题。

getAssets() {

    return this.authService.user$.pipe(
      first(),
      switchMap(user => defer(() => from(this.afs.storage.ref(`${user.uid}/assets`).listAll()))),
      switchMap(assets => from(assets.items).pipe(
        mergeMap(async (asset) => {
        
          return new Promise((res, rej) => {
          
            asset.getDownloadURL().then(url => {
              
              const _asset = {
                name: asset.name,
                url,
              };
  
              this.assets.push(_asset);

              res(_asset);
            })
            .catch((e) => rej(e));
          });
        }),
      )),
      map(() => this.assets),
    );
  }

  ...

  this.getAssets().subscribe(assets => console.log(assets)); // this runs before all asset's url has been resolved

Answer 1

概述

mergemap所有内部可观察物。它旋转N内部可观察的管道，并并行运行，并将所有值串出相同的耦合在管道底部（在这种情况下您的订阅语句）作为单个排放。因此，为什么this.getAssets（）。订阅（Assets =＆gt; console.log（Assets））在所有并行内部Mergemap Pipes完成其单个计算之前运行，因为MERGEMAP不会在发出之前等待所有这些（在它们完成时它会一一发射）。如果要等待可观察到的管道完成，则需要使用forkjoin。

---

fork join

forkjoin您有一组可观察的物品，只关心每个的最终发射值。一个常见的用例是，如果您希望在页面加载（或其他某些事件）上发出多个请求，并且仅在收到所有人的响应时才想采取行动。通过这种方式，它类似于您可能使用Promise.All的方式。

---

解决方案

getAssets(): Observable<Asset[]> {
  return this.authService.user$.pipe(
    // first() will deliver an EmptyError to the observer's error callback if the
    // observable completes before any next notification was sent. If you don't
    // want this behavior, use take(1) instead.
    first(),
    // Switch to users firebase asset stream.
    switchMap(user => {
      // You might have to tweak this part. I'm not exactly sure what 
      // listAll() returns. I guessed that it returns a promise with
      // firebase asset metadata.
      return from(this.afs.storage.ref(`${user.uid}/assets`).listAll());
    }),
    // Map to objects that contain method to get image url.
    map(firebaseAssetMetadata => firebaseAssetMetadata?.items ?? []),
    // Switch to parallel getDownloadUrl streams.
    switchMap(assets => {
      // Not an rxjs map, a regular list map. Returns a list of getAssetUrlPipes.
      const parallelGetAssetUrlPipes = assets.map(asset => {
        return from(asset.getDownloadUrl()).pipe(
          map(url => { name: asset.name, url })
        );
      });
      // 1) Listen to all parallel pipes.
      // 2) Wait until they've all completed.
      // 3) Merge all parallel data into a list.
      // 4) Then move list down the pipe.
      return forkJoin(parallelGetAssetUrlPipes);
    }),
    // Outputs all parallel pipe data as a single emission in list form.
    // Set local variable to users asset data.
    tap(assetObjects => this.assets = assetObjects)
  );
}

// Outputs the list of user asset data.
this.getAssets().subscribe(console.log);

祝您好运，并享受您的瑞典肉丸！

Answer 2

const { from } = rxjs
const { mergeMap } = rxjs.operators

const assets = [1,2,3,4,5]

function getUrl (index) {
  return new Promise((res) => {
    setTimeout(() => res(`http://example.com/${index}`), Math.random() * 3 + 1000)
  })
}

// add param2 1 for mergeMap === concatMap
from(assets).pipe(
  mergeMap(asset => {
    return getUrl(asset)
  }, 1)
).subscribe(console.log)

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/7.5.5/rxjs.umd.min.js"></script>

使用concatmap一个一一运行。