RxJS ждет завершения предыдущего выполнения

Question

Хорошо, я новичок в RxJ и не могу ничего понять. Мне нужно реализовать обработку изображений, в которой пользователь добавляет несколько изображений одновременно, и для каждого изображения, среди прочего, должны выполняться следующие действия:

1. Создание миниатюры изображения в веб-работнике
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, чтобы выполнить текущий вызов (вот почему он возвращает объект обещания), а также дождался предыдущего upload, чтобы выполнить текущий (так что upload также возвращается обещание). Я понятия не имею, как этого достичь. Любой ниндзя 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();

Примечание: загрузка не ожидает обещания generateThumbnail, поскольку это должно быть обработано самим concatMap.

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

Answer 2

Хотите ли вы, чтобы генерация миниатюр осуществлялась только последовательно или генерация второго миниатюры также должна ждать завершения загрузки первого?

Если вы не хотите ждать загрузки,вам просто нужно объединить их:

import { of, from, } from 'rxjs';
import { concatMap } 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);

imageObservable.pipe(concatMap(image => from(generateThumbnail(image)))).subscribe();

concatMap делает то, что вы хотите, и я только что сделал Observable из вашего Promise с оператором создания from, которыйЯ не думаю, что это необходимо.Но обычно проще работать с Observables вместо Promises при использовании RxJs.

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

Редактировать: Можете ли вы попробовать это?

imageObservable
    .pipe(
        concatMap(image => generateThumbnail(image)),
        concatMap(image => upload(image))
    )
    .subscribe();