Как я могу отложить obs 1 до тех пор, пока obs 2 не испустится в течение x секунд?

Question

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

const {Subject, interval} = rxjs;
const {take} = rxjs.operators;

const subject = new Subject(0)

interval(1000)
  .pipe(take(2))
  // detect that subject has recently fired and delay until
  // 1000 after most recent subject emission, then take only latest
  .subscribe(r => console.log(r))

subject.subscribe(r => console.log('subject emission'))

setTimeout(() => {
  subject.next(1);
}, 500)

setTimeout(() => {
  subject.next(2);
}, 1200)

Фактические результаты:

subject emission
0
subject emission
1

Желаемые результаты

subject emission
subject emission
// one second to be sure no immediate emissions happening
1

Редактировать согласно комментарию @ benshabatnoam.Я использовал интервал, потому что это был быстрый инструмент для достижения, и это было ошибкой.В моем реальном случае это не предсказуемо.Он должен быть представлен obs, который может выстрелить в любое время.Таким образом, фактические желаемые результаты должны быть примерно такими:

// gold obs fires but notices subject is emitting
subject emission
subject emission
// one second to be sure no immediate emissions happening
1
// gold obs fires a few seconds later and sees there are no subject emissions
2

Answer 1

РЕДАКТИРОВАТЬ : Если я правильно вас понял, эта труба выполнит свою работу:

this.obs1.pipe(
  debounceTime(1000),
  switchMap(() => this.obs2)
)

Логика в этой трубе заключается в ожидании секунды, когда obs стреляет (используяоператор debounceTime ), а затем переключите его на прослушивание второй наблюдаемой (используя оператор switchMap ).

Вот демонстрационные примеры для вашего вопроса:

Воспроизведение проблемы

Устранение проблемы ( ОБНОВЛЕНО )

ура

Answer 2

Используйте combineLatest, debounceTime и map:

const main$ = new Subject(0);

main$.subscribe(() => console.log('subject emission'))


// aux$ could be any observable
const aux$ = interval(1000).pipe(
  take(2)
)


setTimeout(() => main$.next(1), 500);
setTimeout(() => main$.next(2), 1200);

combineLatest(main$, aux$).pipe(
  map(([main, _aux]) => main),
  debounceTime(1000), // define your own delay time here
).subscribe(result => console.log(result))

Результат, как и ожидалось.

edit с учетом нового комментария, withLatestFrom это, вероятно, оператор, которого вы ищете:

aux$.pipe(
    withLatestFrom(main$),
    map(([_aux, main]) => main),
    debounceTime(1000),
)