等待链式 observables 完成

Question

在使用 RxJS5 实现我想要的东西时遇到了一些麻烦 - 我有一个简单的 Observables 链，从 Rx.Observable.interval 开始：

const Rx = require('rxjs');

var i = 0;

const obs = Rx.Observable.interval(100)
    .flatMap(function () {
        return Rx.Observable.timer(Math.ceil(500*Math.random()))
            .map(function(val){
                console.log(' => These should all log first => ', val);
                return i++;
            });
    })
    .take(5)
    .merge()  // this doesn't seem to do what I want to do
    .map(function (val) {
        console.log('all done = > ', val);
    });

obs.subscribe();

上面记录了这个：

 => These should all log first =>  0
all done = >  0
 => These should all log first =>  0
all done = >  1
 => These should all log first =>  0
all done = >  2
 => These should all log first =>  0
all done = >  3
 => These should all log first =>  0
all done = >  4

我正在寻找记录：

 => These should all log first =>  0
 => These should all log first =>  0
 => These should all log first =>  0
 => These should all log first =>  0
 => These should all log first =>  0

all done = >  [0,1,2,3,4]

很明显，我们并没有等待所有的 timer observables 完成，因为你会看到“all done！”记录了很多次，穿插着“这些都应该先记录”。

我怎样才能得到我正在寻找的输出？

通常情况下，我们可以为此使用zip，但zip 的API 不适合这个用例，因为我们没有同时将所有计时器可观察对象放在一个地方！

如果我的问题不够清楚，这里是我想做的类比，我们阻止所有回调，直到我们任意完成并收集所有结果：

const async = require('async');
var i = 0;

async.forever(function(cb){

    process.nextTick(function(){
       console.log('These should all log first');
       const err = i++ === 5;
       cb(err, i);
    });

}, function done(err, results){
    // let's pretend results contains all the i values
    console.log('all done');
});

Answer 1

请注意，这基本上会产生我期望看到的日志记录顺序，但我认为这不是正确/最佳的方法：

   const {Observable} = require('rxjs');

    const obs = Observable.interval(100)
        .flatMap(function () {
          return Observable.timer(Math.ceil(500*Math.random()))
              .map(function(val){
                  console.log(' => These should all log first => ', val);
              });
        })
        .take(5)
        .takeLast()  // <<<<<<<<<<
        .map(function () {
            console.log('all done');
        })
        .take(1)    // <<<<<<<<<<

   obs.subscribe();

我认为还有更好的方法来实现这一点。

Answer 2

@jonsharpe 给了我这个答案，它基本上是有效的。这个问题真的可以归结为Rx.Observable.interval，我们可以摆脱Rx.Observable.timer 映射。这是我们的基本答案：

const Rx = require('rxjs');

const obs = Rx.Observable.interval(100)
    .take(5)
    .map(function(v){
        console.log(v);
        return v;
    })
    .reduce(function (prev, curr) {
        return prev.concat(curr);
    },[])
    .last(function (results) {
        return results;
    })
    .map(function(v){
        console.log(v);
    });


obs.subscribe();

不过，我会非常感兴趣的一种方式来做到这一点没有减少。

Answer 3

所以要求是：

• 让顶层区间运行 N 次。
• 每个时间间隔都应该返回一个 Y 计时器 observables 数组。
• Next 应该追踪 N 次，每次都使用一个 Y 可观察的数组。

这样做。这有点天真，但它确实有效。

    let timerArrLength = [ 4, 2, 3 ];
    let svc = Rx.Observable.interval(1000)
        .take(timerArrLength.length)
        .map(function ( index ) {
            let arr = [];
            for ( let i = 0; i < timerArrLength [ index ]; i++ ) {
                arr.push ( Rx.Observable.timer ( 1000 ) );
            }
            return arr;
        });


        svc.subscribe(
            function onNext(v){
                console.log('=> v =>',v);
            },
            function onError(e){
                console.error(e);
            },
            function onComplete(){
                console.log('complete');
            }
        );