javascript - RxJS 等待上一次执行完成

Question

好吧，我是 RxJs 的新手，有些事情我想不通。我需要实现图像处理，其中用户一次添加多个图像，并且对于每个图像，除其他外，必须执行以下操作:

1. 在网络 worker 中生成图像缩略图
2. 上传图片和缩略图到服务器

但我不希望一次生成所有缩略图，我希望它们按顺序生成。这是我尝试过的:https://codesandbox.io/s/funny-mountain-tm0jj?expanddevtools=1&fontsize=14

我也把代码贴在这里:

import { of } from "rxjs";
import { map } from "rxjs/operators";

const imageObservable = of(1, 2, 3, 4, 5);

function generateThumbnail(image) {
  console.log(`Generating thumbnail ${image}`);
  return new Promise(resolve => {
    setTimeout(() => {
      console.log(`Finished generating thumbnail ${image}`);
      resolve({
        image,
        thumbnail: `This is a thumbnail of image ${image}`
      });
    }, 1500);
  });
}

async function upload(imagePromise) {
  const image = await imagePromise;
  console.log("Uploading", image);
  return new Promise(resolve => {
    setTimeout(() => {
      console.log("Finished uploading", image);
      resolve();
    }, 1500);
  });
}

imageObservable.pipe(map(generateThumbnail)).subscribe(upload);

我希望 RxJS 等待之前的 generateThumbnail 以执行当前调用(这就是它返回一个 promise 对象的原因)并且还等待之前的 upload 按顺序执行当前的(因此 upload 也返回 promise)。我不知道如何实现。任何 RxJS 高手愿意帮助我吗？

Answer 1

实现以下目标:

1. 应该按顺序生成缩略图顺序很重要 => 使用concatMap
2. 生成缩略图后应立即上传
3. 可以并行上传顺序不重要 => 使用 map

所以最终的代码可能是

imageObservable
  .pipe(
    // preserve orders and wait for previous observable
    concatMap(generateThumbnail), 

    // map the generate
    map(upload)
  )
  .subscribe();

Note: upload does not wait for the generateThumbnail promise since this should be handled by concatMap itself.

async function upload(image) {
  console.log(`[upload] start ${image.image}`);
  return new Promise(resolve => {
    setTimeout(() => {
      console.log(`[upload] done ${image.image}`);
      resolve();
    }, 1000);
  });
}