rxjs: объединение двух потоков, но отмена первого после того, как второй отправил событие

Question

Я уверен, что смогу найти ответ на мою проблему в документации по rxjs, но я никуда не денусь ...

Проблема: У меня есть два потока, один из которых важнее другого. Оба предоставляют одинаковую информацию. Но в случае, если поток A впервые предоставляет значение, я хочу полностью закрыть поток B, потому что он теперь устарел.

Это что-то вроде кеша: введите значение из кеша, но как только оно станет действительным, обновите кеш и больше не читайте его.

Я пробовал несколько комбинаций для двух потоков, но я не могу отменить B после публикации A ...

Какой rx-оператор может предоставить мне необходимую функциональность?

Answer 1

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

const interval1$ = interval(1000);
const interval2$ = interval(800);

const source1 = interval1$.pipe(
  map(x => { return { id: 1, value: x } }),
  take(5)
);
const source2 = interval2$.pipe(
  map(x => { return { id: 2, value: x } }),
  take(5)
);

const s = merge(source1, source2);

const subscribe = s.pipe(take(1)).subscribe(x => console.log(x));

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

Answer 2

Похоже, вы можете просто использовать merge и takeUntil:

merge(
  streamA$.pipe(
    takeUntil(streamB$)
  ),
  streamB$
)

Когда streamB$ испускает что-либо, streamA$ будет игнорироваться, потому что takeUntil завершит цепочку.