HTTP-запрос очереди RxJS

Question

Я пытаюсь реализовать сохранение в последовательности в Angular. Таким образом, запрос «сохранить» должен дождаться окончания предыдущего. Я должен ждать, потому что предыдущий вызов возвращает ETag.

Я думал о какой-то "карте" в RxJS, но я не знаю, как дождаться окончания предыдущей.

Основная идея показана в: https://stackblitz.com/edit/typescript-mmcqhy?file=index.ts

Answer 1

Одним из способов может быть использование concatMap, а затем tap для внутренних журналов, но это действительно зависит от того, какое поведение вы ожидаете.

let save = function(arg){
  console.log("start request")
  return ajax('https://httpbin.org/get').pipe(
    tap(...),
  );
}

callQueue$.pipe(
  concatMap(i => save(i)),
)
.subscribe(result => {
  ...
});

Ваша обновленная демка: https://stackblitz.com/edit/typescript-13jrrk?file=index.ts

Answer 2

Вот решение с конкурентностью mergeMap - https://stackblitz.com/edit/typescript-y4nu3i?file=index.ts

callQueue$
.pipe(
    mergeMap(result => save(result), 1)
    )
  .subscribe(
    function (data) {
      console.log("finished OK request")
    }
  );

Answer 3

Я сделал это с помощью concat:

import { concat } from 'rxjs/observable/concat';

const source = concat(
    this.myService.saveStuff({ test: 'swag' }), //returns an observable
    this.myService.saveStuff({ test: 'yolo' })
);

source.subscribe(
    response => {
        console.log('Called after each successful save');
    },
    () => {
        console.log('Called when an error happens')
    },
    () => {
        console.log('called when all observables are complete')
    });