javascript - 对主题使用 mergeMap() 会忽略我正在合并的可观察对象上的 takeWhile()

Question

我目前正在尝试找出一种暂停可观察的方法。看了 blesh 在 github 上的帖子 https://github.com/ReactiveX/rxjs/issues/1542我认为我走在正确的轨道上。但由于某种原因，一旦我从暂停主题执行 switchMap() ，我的观察者上的 takeWhile() 就会被忽略。

正确结束:

export class CompositionService {
    cursor = -1;
    pauser = new Subject();
    interval;
    init = (slides) => {
        let waitUntil = 0;
        return this.interval = Observable
            .range(0, slides.length)
            .mergeMap((i) => {
                let next = Observable.of(i).delay(waitUntil);
                waitUntil += !!slides[i]["duration"] ? slides[i]["duration"] : 0;
                return next;
            })
            .scan((cursor) => {
                return this.cursor = cursor = slides[cursor + 1] ? cursor + 1 : -1;
            }, this.cursor)
            .map(cursor => slides[cursor])
            .takeWhile((slide) => {
                return !!slide;
            });

    };
    // these methods are not called for this sample
    play = () => {
        this.pauser.next(false);
    };
    pause = () => {
        this.pauser.next(true);
    };
};

当以这种方式调用时，这会起作用:

it("should subscribe to init", (done) => {
    slides.forEach((slide, i) => {
        if (slide.duration) {
            slide.duration = slide.duration / 100;
        }
    });
    composition.init(slides).subscribe(
        (slide) => {
            console.log(slide);
        },
        (err) => {
            console.log("Error: " + err);
        },
        () => {
            done();
        });
});

虽然前面的示例按照宣传的方式工作，但当我添加一些“魔法”时，观察者间隔永远不会结束:

export class CompositionService2 {
    cursor = -1;
    pauser = new Subject();
    interval;
    init = (slides) => {
        let waitUntil = 0;
        this.interval = Observable
            .range(0, slides.length)
            .mergeMap((i) => {
                let next = Observable.of(i).delay(waitUntil);
                waitUntil += !!slides[i]["duration"] ? slides[i]["duration"] : 0;
                return next;
            })
            .scan((cursor) => {
                return this.cursor = cursor = slides[cursor + 1] ? cursor + 1 : -1;
            }, this.cursor)
            .map(cursor => slides[cursor])
            .takeWhile((slide) => {
                return !!slide;
            });
        return this.pauser
            // leaving commented for clarity of the end game
            // .switchMap( paused => paused ? Observable.never() : this.interval );
            // however, not even a straight forward switchMap is yeilding the expected results
            .switchMap( paused => this.interval );            
    };
    play = () => {
        this.pauser.next(false);
    };
    pause = () => {
        this.pauser.next(true);
    };
};

这样调用:

it("should subscribe to init", (done) => {
    slides.forEach((slide, i) => {
        if (slide.duration) {
            slide.duration = slide.duration / 100;
        }
    });
    composition.init(slides).subscribe(
        (slide) => {
            console.log(slide);
        },
        (err) => {
            console.log("Error: " + err);
        },
        () => {
            //I never get here!!!!!
            done();
        });
    // kickstart my heart!
    composition.play();
});

有人知道我在这里做错了什么吗？

Answer 1

您没有完成外部流。在第一个版本中，当 takeWhile 完成流时，您就完成了。但是，一旦将其嵌套在 switchMap 中。您只会完成内部流，因为外部流(Subject)永远不会完成。当它变平时，它对订阅者来说就像一个永无止境的流。

如果你想完成它，你需要在某个时刻终止流，例如:

composition.init(slides)
  .take(3)
  .subscribe(
    (slide) => {
        console.log(slide);
    },
    (err) => {
        console.log("Error: " + err);
    },
    () => {
        //I never get here!!!!!
        done();
    });

我不太相信 Rx 实际上是正确的工具，因为流并不是真正设计为“暂停”的，因为你实际上无法阻止 Observable 继续传播。您可能已经注意到在暂停之间为了存储状态而跳过的圈数，因此考虑使用生成器或其他库(如 IxJS)可能是有意义的。 。但我离题了。