Как в RxJS обрабатывать асинхронные потоки последовательно?

Question

[Mariik]

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

Для примера вот такой код:

import { Observable, of } from 'rxjs';
import { merge, concat } from 'rxjs/operators';

const emmit$ = Observable.create(function(observer) {
  setTimeout( ()=> observer.next('emmit #1'), 40);
  setTimeout( ()=> observer.next('emmit #2'), 5000);
});

const process$ = Observable.create(function(observer) {
  setTimeout( ()=> observer.next('process #1'), 2500);
  setTimeout( ()=> observer.next('process #2'), 400);
});

const combined$ =  emmit$.pipe(concat(process$));
combined$.subscribe( (val) =>{
  console.log(val);
});

Я ожидаю вот такой вывод:

// process #1
// process #2
// emmit #1
// emmit #2

На практике получается так, словно я подписался только на $emmit.

Answer 1

[bro-dev]

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