Rx.Subject 失去事件

Question

任何人都可以解释这 3 个变体之间的区别吗？

首先作为excpect，处理所有事件。
但是第二个输掉了最后一个事件 (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中的源代码中看到它：

Rx.Subject 失去事件

Rx.Subject loses events

javascript

rxjs

rxjs5

ngrx

1

2

3