javascript - Rx.Subject 丢失事件

Question

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

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

1. 首先作为 expect，处理所有事件。
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 已标记为已停止。

第三个演示中的过程完全相同。

在Subject.ts中的源码中可以看到:

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

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