RxJS 5 Observable源是否可以被链中的另一个源停止？

Question

RxJS 5 Angular 2 RC4应用程序写在Typescript 1.9:

我有两个可观察链.我想,如果在第二个条件满足,第一个立即完成.我的努力似乎不必要地复杂.在下面的示例中,我尝试在发出3个值后停止第一个observable:

source = Observable.interval(1000)
        .do(()=>this.print("*******EMITTING from Source*******"))
        .switchMap(count => {
            if(count<3){ //just pass along the value
                return Observable.create(observer=>{
                    observer.next(count);observer.complete()
                })                
            }
            else{ //abort by issuing a non-productive observable
                return Observable.create(observer=>
                    observer.complete()
                )
            }
        })
this.source.subscribe(count=>this.print('Ouput is '+count);

这是输出:

*******EMITTING from Source*******
Output is 0
*******EMITTING from Source*******
Output is 1
*******EMITTING from Source*******
Output is 2
*******EMITTING from Source*******
*******EMITTING from Source*******
*******EMITTING from Source*******

所以,在功能上我得到了我想要的结果,因为更宽的脚本在三次输出后停止收到通知.但是,我确信有更好的方法.我的问题是:

1. 上游可观测量继续发光.我怎么能阻止它？
2. 我在每次排放时都会在链条上创建一个新的可观察对象.难道我不能只传递前3个值但是在第4个中止或完成链吗？

Answer 1

您可以使用take运算符来执行此操作.take获取前N个事件并完成流.

this.source = Observable.interval(1000)
  .do(()=>this.print("*******EMITTING from Source*******"))
  .take(3);
this.source.subscribe(count=>this.print('Ouput is '+count);

您的示例的流未完成,因为switchMap内部流完成时外部流未完成.switchMap()等于map().switch().在您的示例中,该map部件会发出以下可观察对象:

1. next(0),complete()
2. next(1),complete()
3. next(2),complete()
4. 完成()
5. 完成()
6. 完成()
7. 完成()
8. ......(继续无限地)......

并且该switch部分切换那些可观察的并且一直等待即将到来的可观察量.

编辑

你的例子也可以写成:

source = Observable.interval(1000)
        .do(()=>this.print("*******EMITTING from Source*******"))
        .takeWhile(count => count < 3);

编辑2

关于您的评论,如果您想要在内部流发出时终止流true:

source = Observable.interval(1000)
        .do(()=>this.print("*******EMITTING from Source*******"))
        .switchMap(count => createSomeObservable(count))
        .takeWhile(x => x !== true);