Как создать наблюдаемую в RxJS, которая ожидает 2 наблюдаемых (последовательно), затем действует, а затем снова ожидает эти 2 в последовательности

Question

Я борюсь с конкретным сценарием, для которого я надеялся использовать RxJS:

Есть две наблюдаемые, которые должны сработать последовательно для значения, которое будет зарегистрировано:

1. Наблюдаемая, представляющая пользователя, нажимающего кнопку

2. Наблюдаемая, представляющая устройство, возвращающее следующее значение

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

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

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

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

Что было бы лучшим решением для этогодело?

import { Observable, Subject, forkJoin } from 'rxjs';
import { concat, repeat, switchMap } from 'rxjs/operators';

let user: Subject<any> = new Subject<any>();
let device: Subject<number> = new Subject<number>();

user.pipe(concat(device)).subscribe(() => {
    console.log("both called");
});


setInterval(() => {
    setTimeout(() => {
        user.complete();
        console.log("user action executed");
    }, 2000);

    setTimeout(() => {
        device.next(new Date().getSeconds());
        console.log("device value sent");
        device.complete();
    }, 5000);
}, 10000);

Answer 1

exhaustMap позволит вам прослушивать другой поток, игнорируя при этом последующие события из исходного потока:

const { fromEvent, timer, Subject } = rxjs;
const { exhaustMap, take } = rxjs.operators;

// mock device$ will output every 3 seconds
const device$ = new Subject();
timer(0, 3000).subscribe(device$);
device$.subscribe(console.log); // output device ticking

// listen to button clicks
const btn = document.getElementById('btn');
const click$ = fromEvent(btn, 'click');
click$.pipe(
  // when a button is clicked -- we start listening
  // for the next event on the device$
  // until then we don't react to click$
  exhaustMap(() => device$.pipe(take(1)))
).subscribe(value => {
  console.log('Clicked at ' + value);
});

<script src="https://unpkg.com/rxjs@6.4.0/bundles/rxjs.umd.min.js"></script>

<button id="btn">Clicker</button>

Вот иллюстрация того, что происходит

Играть с exhaustMap оператор здесь .

Надеюсь, это поможет

Answer 2

Вы можете переключить карту и вернуть другую наблюдаемую.

const { of, fromEvent } = rxjs;
const { switchMap, delay } = rxjs.operators;

fromEvent(document.getElementById('btn'), 'click').pipe(
    switchMap(e => of('some value').pipe(delay(1000)))
  )
  .subscribe(val => { console.log(val) });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"></script>
<button id="btn">Click</button>