奇怪的 takeUntil 在 rxjs 6 中的行为

Question

首先停止启动按钮，然后点击restore01按钮和restore02按钮恢复启动按钮，

那么错误是当点击停止按钮01或停止按钮02时停止按钮应该停止，但不是

注意console.log，启动按钮在启动时停止，然后单击restore01和restore02，然后启动按钮起作用，现在bug像冬天一样来了，当我在控制台中单击stop01按钮或stop02按钮时。 log it show stop stream Array [ "r", "" ] 但是开始按钮仍然有效，没有停止，你能想象一下我看到这种情况的感觉吗？天哪

online demo

代码很糟糕

const startDom = document.querySelector("#start")
const numberDom = document.querySelector("#number")
const restoreDom01 = document.querySelector("#restore01")
const stopDom01 = document.querySelector("#stop01")
const restoreDom02 = document.querySelector("#restore02")
const stopDom02 = document.querySelector("#stop02")

let init01 = false
let init02 = false

const origin01$ = new rxjs.Subject()
const origin02$ = new rxjs.Subject()

const restore01$ = rxjs.fromEvent(restoreDom01, "click")
restore01$.subscribe(() => origin01$.next("r"))
const stop01$ = rxjs.fromEvent(stopDom01, "click")
stop01$.subscribe(() => origin01$.next(""))

const restore02$ = rxjs.fromEvent(restoreDom02, "click")
restore02$.subscribe(() => origin02$.next("r"))
const stop02$ = rxjs.fromEvent(stopDom02, "click")
stop02$.subscribe(() => origin02$.next(""))

const [stopReg$, right$] = rxjs
.combineLatest(origin01$, origin02$)
.pipe(rxjs.operators.partition(arr => arr.some(isEmpty)));
const restoreReg$ = right$.pipe(
  rxjs.operators.take(1),
  rxjs.operators.repeatWhen(() => stopReg$)
);


const start$ = rxjs.fromEvent(startDom, "click");
start$
  .pipe(
  rxjs.operators.mapTo(1),
  rxjs.operators.scan((acc, cur) => acc + cur, 0),
  rxjs.operators.takeUntil(stopReg$),
  rxjs.operators.repeatWhen(() => restoreReg$)
)
  .subscribe(x => (numberDom.innerHTML = x));


stopReg$.subscribe(x => console.log("stop stream", x));
restoreReg$.subscribe(x => console.log("restore stream", x));

origin01$.next("")
origin02$.next("")

function isEmpty(n) {
  return n === "";
}

可能不是很清楚，所以我告诉你步骤

1. 点击2个恢复按钮，一个一个

2. 点击开始按钮，点击后会看到数字加一

3. 点击停止按钮，好的，这是个问题，现在开始按钮应该停止，但不是，你仍然可以添加一个点击开始按钮，但是在控制台你可以看到停止流被订阅，好奇怪

I also have other demo with the same problem using react and rxjs

Answer 1

您的问题可能与您使用combineLatest 和Subject 的方式有关。

combineLatest(origin01$, origin02$)

其中origin01$ 和origin02$ 都是Subject。主体不会自然而然地持有旧值来发射，它们本质上只是“事件发射器”，它们会将它们的发射广播给订阅者，仅此而已，在发射之后他们不再持有它。

看起来当origin01$ 触发时，combineLatest 必须等待另一个 Subject 再次触发才能发出信号。您的stopReg$ 流可能没有像您期望的那样触发，请使用tap 进行检查，如下所示：

rxjs.operators.repeatWhen(() => 
  stopReg$.pipe(
    rxjs.operators.tap(x => console.log('got to here'))
  )
)

BehaviorSubject 可以在这里提供帮助。这是RxJS docs的描述：

Subject 的一种变体，它需要一个初始值并发出它的 订阅时的当前值。

如果您将原点Subject 改为BehaviorSubject，您可能会得到您希望的预期结果。我试过这个：

const origin01$ = new rxjs.BehaviorSubject("")
const origin02$ = new rxjs.BehaviorSubject("")

在我只点击一个“停止”按钮后，我无法看到“开始”扫描流在我点击它时继续对值求和。

我相信这是可行的，因为在 combineLatest 中，因为它是 BehaviorSubject，它已经能够获取最新值，而不必等待来自另一个 Subject 的额外发射。