RxJS：Observables 之间的可重复竞赛

Question

我有两个独立但重叠的用户交互流程，它们都触发相同的操作。如果一个流程已启动，则在第一个流程完成之前，另一个流程不应启动。我可能想多了，但我找不到符合我要求的模式。

具体例子：

我想打开上下文菜单，以便在具有两个独立交互流的文本区域中进行选择：

通过鼠标：鼠标按下 -> 选择已更改 -> 鼠标弹起
通过键盘：选择已更改 -> 超时（500 毫秒）

如您所见，两个流程在“选择已更改”部分重叠，但第一个和最终条件不同。用键盘选择时，没有明确的结束条件，所以我依赖定时器。用鼠标选择时是明确的结束条件（鼠标松开），所以我想立即显示菜单。两者都被定义为单独的 Observables，它们在结束条件下完成。

为了区分两者，我想等待第一个发出的 observable，然后只听那个 observable 直到它完成，然后重新开始。因此，当出现鼠标按下事件时，我想等待鼠标弹起事件触发。当没有鼠标按下的选择更改时，我开始超时。

我试过的东西：

exhaust/exhaustMap: 这在技术上是我想要的（在考虑下一个之前跟随一个 Observable 到底）但是那些不带 Observables 列表但是发出可观察量的函数。不过，我的可观察对象是固定的。我不知道如何连接它来做我想做的事。

race: 我可以传入一个 observables 列表，然后首先发出的任何东西都会被跟踪到最后。但是，它不会在完成时重新订阅。 race() 简单地完成就是这样。我试过这样的事情：

const openMenu$ = of(true).pipe(
  startWith(race(a$, b$)),
  exhaust(),
  repeat()
);

或

const openMenu$ = of(true).pipe(
  exhaustMap(() => race(a$, b$)),
  exhaust(),
  repeat()
);

或

const openMenu$ = race(a$, b$).pipe(
  repeat()
);

但所有这些只是一遍又一遍地重复第一个 Observable。我找不到重新开始比赛的方法。

有什么想法吗？还是我完全错了？

Answer 1

一个简单的race应该是可重复的。

import { defer, timer, race } from 'rxjs';
import { map } from 'rxjs/operators';

// an observable that will emit `0` after a random time less than 3 seconds
const randomTime$ = defer(() => timer(Math.random() * 3000));

// Will randomly emit "a" after a random time less than 3 seconds
const a$ = randomTime$.pipe(map(() => 'a'));

// Will randomly emit "b" after a random time less than 3 seconds
const b$ = randomTime$.pipe(map(() => 'b'));

// Race a$ and b$
const source$ = race(a$, b$).pipe(
  // repeat the race 5 times.
  random(5)
);

source$.subscribe(console.log);

上面的代码中不需要 startWith 和 exhaust。 race(a$, b$).pipe(repeat()) 应有尽有。

我会检查第一个 Observable 是否总是在第二个 Observable 之前发出。也许它有它自己的 startWith 或某种保证它“赢得”比赛的同步发射。如果您与两个同步可观察对象赛跑，第一个总是获胜。

RxJS：Observables 之间的可重复竞赛

RxJS: repeatable race between Observables

rxjs

rxjs6