concatMap не может обработать все элементы в массиве

Question

Я обрабатываю массив файлов.Для каждого файла мне нужно запросить правильный URL-адрес для загрузки, а затем загрузить этот файл в правильную конечную точку.

concatMap должен испускаться только после завершения предыдущего значения.Таким образом, файлы должны обрабатываться последовательно.

Вот мой код:

filesChanged(e) {

const files_raw = e.srcElement.files;

of(files_raw)
.pipe(
  flatMap((file) => file),
  tap((file) => console.log("Firing 1", file)),
  concatMap((file) => {
    return this.ms.fetchUploadURL().pipe(withLatestFrom(of(file)));
  }),
  tap((file) => console.log("Firing 2", file)),
  concatMap((values) => {
    return this.ms.uploadDirectToS3(values[0], values[1])
      .pipe(tap(value => console.log('From request', value)), withLatestFrom(of(values[0])));
  })
)
.subscribe(event => {
  console.log('In Subscribe', event);
});
}

Если я попытаюсь загрузить 4 файла, этот код выведет на консоль следующий вывод: Firing 1 Firing 1 Firing 1 Firing 1 Firing 2 From request <value> In Subscribe <value>

Другими словами, только первый файл обрабатывается и загружается.Все остальные файлы, похоже, игнорируются.Почему это так?

Разве не должно начинаться извлечение второго URL-адреса после завершения первого?

Кроме того, если я изменю concatMap на mergeMap, то он работает, но обрабатывает все файлы одновременно, а не последовательно.Я думал, что передача опции параллелизма в mergeMap может помочь, но действует аналогично concatMap.

Answer 1

Прежде всего я попытался смоделировать ваш случай здесь: https://stackblitz.com/edit/rxjs-ijfky6?devtoolsheight=60. Обратите внимание, что я использую mergeMap вместо flatMap, но они точно такие же.

В соответствии с таким моделированиеми для логики, которую я вижу реализованной, то, что вы получаете, имеет смысл, и вы, вероятно, видите только 1 появление «Увольнения 2», потому что во втором вызове uploadDirectToS3 возникает некоторая ошибка.Возможно, вам следует добавить функцию ошибки к subscribe, как в приведенном выше моделировании.

Сказал, что, я думаю, есть несколько исправлений, которые вы должны добавить в свой код, чтобы получить то, что вы хотите.

Давайте начнем с простого.Вы можете заменить

of(files_raw)
.pipe(
  flatMap((file) => file),

на from(files_raw).

Теперь давайте введем ядро ​​точки.Вы хотите получить url для использования через функцию this.ms.fetchUploadURL() (кстати, меня удивляет, что такая функция не имеет никакого параметра, предполагая, что она всегда возвращает одно и то же значение, но здесь это не имеет значения).И затем вы хотите вызвать this.ms.uploadDirectToS3, передав в качестве параметров url и file .Если это так, то сначала вы должны позвонить this.ms.fetchUploadURL(), а затем через оператора switchMap переключить на другое наблюдаемое, возвращаемое this.ms.uploadDirectToS3.Эти операции должны выполняться в пределах concatMap, чтобы их можно было выполнять последовательно.

Конечный код может выглядеть примерно так:

from(files_raw)
.pipe(
  tap((file) => console.log("Firing 1", file)),
  concatMap((file) => {
    return fetchUploadURL().pipe(
      tap((file) => console.log("Firing 2", file)),
      switchMap(url => uploadDirectToS3(url, file))
    );
  }),
)
.subscribe(
  d => console.log('subscribe', d),
  err => console.error(err)
)

Здесь симуляция нового кода https://stackblitz.com/edit/rxjs-whip1l?devtoolsheight=60