Как реализовать Очередь для вызова функций в rxjs?

Question

У меня есть кнопка, которая вызывает функцию сохранения, которая вызывает функцию сохранения в службе.

    <button (click)="save()">save</button>

    save() {
     this.someService.save().subscribe(r => {
       console.log({ r });
     });
    }

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

После завершения текущего метода сохранения в службе затем извлеките следующий из очереди и снова вызовите, чтобы сохранить метод из службы и т. Д., Пока в очереди ничего не останется.

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

* примечание: функция save в SomeService Я не могу изменить.

    save() {
     this.someService.addToQueue(true).subscribe(r => {
       console.log({ r });
     });
    }

@Injectable({
  providedIn: "root"
})
export class SomeService {
  queue = new Subject();

  constructor() {
    this.queue.pipe(concatMap(a => this.save(a)));
  }

  addToQueue(saveAndExit) {
    this.queue.next(saveAndExit);

    return this.queue;
  }

  save(saveAndExit) {
    return Observable.create(obs => {
      setTimeout(() => {
        obs.next({ data: true });
      }, 5000);
    });
  }
}

Демонстрация по стеку

Answer 1

Проблема с concatMap в том, что он не подписывается на следующую внутреннюю Наблюдаемую, пока не завершится предыдущая.Если вы выполните это:

  save(saveAndExit) {
    return Observable.create(obs => {
      setTimeout(() => {
        obs.next({ data: true });
        obs.complete()
      }, 5000);
    });
  }

Ваша проблема будет решена, но, поскольку вы не можете изменить эту функцию, вам необходимо выполнить ее вручную, используя f.ex.take(1) или first():

this.queue.pipe(concatMap(a => this.save(a).pipe(take(1)) ));