Rx.Subject 丢失事件

Question

谁能解释这 3 个变体之间的区别？

http://jsfiddle.net/8vx2g3fr/2/

1. 首先按预期工作，处理所有事件。
2. 但是第二个输了最后一个事件（3）
3. 第三次输掉第二次事件（2）

您能否帮助我了解问题所在以及如何使第三个变体处理所有事件？

1

let bs = new Rx.Subject();
bs
    .subscribe(v=>{
        console.log("in", v);
        if (v % 2 == 0) {
            setTimeout(()=>{
                console.log(" out", v, "->" , v + 1);
                bs.next(v+1);
            }, 0);
        }
    });

bs.next(0);
bs.next(2);

输出：

in 0
in 2
 out 0 -> 1
in 1
 out 2 -> 3
in 3

2

let bs2 = new Rx.Subject();
bs2
    .subscribe(v=>{
        console.log("in", v);
        if (v % 2 == 0) {            
            Rx.Observable.interval(0).take(1)
                .map(()=>{console.log(" out", v, "->" , v + 1);return v+1;})
                .subscribe(bs2);
        }
    });

bs2.next(0);
bs2.next(2);

输出：

in 0
in 2
 out 0 -> 1
in 1
 out 2 -> 3

3

let bs3 = new Rx.Subject();
bs3
    .switchMap(v=>{
        console.log("in", v);
        if (v % 2 == 0) {            
            return Rx.Observable.interval(0).take(1)
                .map(()=>{console.log(" out", v, "->" , v + 1);return v+1;});
        }

    return Rx.Observable.empty();     
    }).subscribe(bs3);

bs3.next(0);
bs3.next(2);

输出：

in 0
in 2
 out 2 -> 3
in 3

Answer 1

这一切实际上都是预期的行为。

令人困惑的是，当您多次重复使用 Subject 和诸如 take() 之类的运算符时会发生什么。

运算符take(1) 只接受一个值并发送complete 通知。由于.subscribe(bs2)，Subject 收到此通知。现在是最重要的部分。
当Subject 收到complete 或error 通知时，它会将自己标记为已停止。这意味着它永远不会汇出任何项目或通知，这在 Rx 中是正确且预期的行为。通知complete 或error 必须是最后的发射。

所以Subject 由第一个take(1) 完成，它由值0 触发（bs2.next(0) 调用）。

然后，当值 2 触发 Observable.interval(0).take(1) 的第二次运行时，Subject 会接收到它，但它会自动被忽略，因为 Subject 已标记为已停止。

你第三个demo的过程完全一样。

您可以在Subject.ts的源代码中看到它：

• https://github.com/ReactiveX/rxjs/blob/master/src/Subject.ts#L86

• https://github.com/ReactiveX/rxjs/blob/master/src/Subject.ts#L56