RxJS:可重复的可观测值竞赛

Question

我有两个单独但重叠的用户交互流，它们都触发相同的操作。如果一个流已启动，则另一个流在第一个流完成之前不应该启动。我可能想得太多了，但我找不到一个模式来做我想做的事。

具体示例：

我想为文本区域中的选择打开一个上下文菜单，其中包含两个独立的交互流：

鼠标点击Timeout(500ms)

1. ：鼠标下->选择改变->鼠标Up
2. 通过键盘:选择改变了->

如您所见，这两个流在"Selection“部分中重叠，但第一个和最后一个条件不同。当选择用键盘，没有明确的结束条件，所以我依靠一个定时器。当用鼠标选择时，有一个明确的结束条件(鼠标向上)，所以我想立即显示菜单。两者都被定义为在其结束条件下完成的单独的可观测值。

为了区分这两者，我想先等待任何一个可观察到的发射，然后只听这个可观察的，直到它完成，然后重新开始。因此，当有一个鼠标向下事件时，我想等待鼠标向上事件触发。当有一个选择更改没有鼠标关闭，我启动超时。

我尝试过的：

从技术上讲，这就是我想要的(在考虑下一个之前，遵循一个可观察到的，直到结束)，但是那些不需要列出一个可观察的列表，而是一个发射可观测的函数。不过，我的可观测值是固定的。我不知道怎样才能做我想做的事。

比赛：，我可以通过一个可观察到的列表，不管是什么第一次发射，然后一直持续到最后。然而，当它完成时，它不会重新订阅。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()
);

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

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

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())应该是你所需要的。

我会检查第一个可观测到的并不总是在第二个之前发出。也许它有自己的startWith或者某种同步排放来保证它“赢得”比赛。如果您竞赛两个同步可观测值，第一个总是赢。