将 promise 函数转换为 observable答案

【问题标题】：Convert a promise function to an observable将 promise 函数转换为 observable
【发布时间】：2021-09-21 23:41:52
【问题描述】：

我有一段代码使用并返回一个承诺。但是，我想将其转换为可观察的。请指教。

原码：

export const uploadMultipleFilesToAzure = (
  uploadData: Omit<UploadMultipleToAzure, 'id'>[],
  handleProgress: (
    loadedBytes: number,
    fileData: UploadMultipleToAzure['fileData'],
    uploadId: Upload['id']
  ) => void,
  individualCallback: Function
): Promise<BlockBlobUploadHeaders[]> => {
  const PIPELINE: Pipeline = newPipeline(new AnonymousCredential(), {
    retryOptions: { maxTries: 4 }, // Retry options
    keepAliveOptions: {
      // Keep alive is enabled by default, disable keep alive by setting false
      enable: false,
    },
  });

  const promises: Promise<BlobUploadCommonResponse>[] = [];
  forEach(uploadData, (uploadItem) => {
    let blockBlobClient: BlockBlobClient = new BlockBlobClient(
      uploadItem.BlobURL,
      PIPELINE
    );
    promises.push(
      (
        blockBlobClient.uploadData(uploadItem.fileData as Blob, {
          blockSize:
            (uploadItem.fileData as Blob).size > 1024 * 1024 * 32
              ? 1024 * 1024 * 4
              : 1024 * 512,
          maxSingleShotSize: 1024 * 512,
          concurrency: 20, // 20 concurrency,
          onProgress: (ev: TransferProgressEvent) =>
            handleProgress(
              ev.loadedBytes / (uploadItem.fileData as Blob).size,
              uploadItem.fileData,
              uploadItem.id
            ),
        }) as any
      ).then(() => {
        individualCallback(uploadItem);
      })
    );
  });
  return Promise.all(promises);
};

【问题讨论】：

我不清楚你的问题。您是否希望将 Promise 转换为 Observable？如果是这样，from(promise) 有效：stackoverflow.com/questions/39319279/…。或者您是否希望使用基于 Observable 的方法而不是 Promise 来完全重写此代码？如果是这样，您是否有关于流程的具体问题？
是的，我想用 observables 重写它。我知道 forkJoin 或 combineLatest 可以解决问题，但不确定如何进一步进行。
没有一对一的更改来将代码从基于 Promise 的代码更改为基于 Observable 的代码。相反，您需要在排放和通知方面重新考虑它。你能在 Stackblitz 中构建一个 SMALL 演示（而不是你的整个应用程序）来展示你迄今为止所做的尝试吗？

标签： typescript promise rxjs observable

【解决方案1】：

在我看来，这就像一个简单的 forkJoin 情况。

简化，你有：

const uploadMultipleFilesToAzure = (uploads: any[]) =>
  forkJoin(uploads.map(upload => 
    upload.somethingReturningPromise(prms)
  );

这只是简单地将 Promises.all 替换为 forkJoin。 forkJoin 可以将 Promise 作为输入，并会发出带有结果数组的单个事件。

如果你有内部的“然后”这将更改为 xMap - 让我们假设 mergeMap。你现在需要 from() 将 promise 转换为 observable：

const uploadMultipleFilesToAzure = (uploads: any[]) =>
  forkJoin(uploads.map(upload => 
    from(upload.somethingReturningPromise(prms)).pipe(
      mergeMap( ()=> individualCallback(uploadItem);
    )
  );

最后，如果你使用 forkJoin 或 from() ，那么当这个函数（创建 observable）被调用时，promise 将被调用，而不是当 observable 被订阅时。出于这个原因，通常最好使用 defer() 代替：

  const uploadMultipleFilesToAzure = (uploads: any[]) =>
    forkJoin(uploads.map(upload => 
      defer(() => upload.somethingReturningPromise(prms)).pipe(
        mergeMap( ()=> individualCallback(uploadItem);
      )
    );

【讨论】：