いくつかのトラブル私はRxJS5で何をしたい達成を持つ

Question

を完了するために連鎖観測待ち - 私はシンプルな観測鎖を有していて、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] 

それは、私たちはすべてのティムを待っていないことは明らかだ：私はこれを記録しているよ

あなたが "すべて完了しました！"何度もログに記録され、「これはすべて最初にログする必要があります」と散在しています。

私が探している出力を得るにはどうすればよいですか？

通常、これにはzipを使用できますが、zipのAPIは、同時に1つの場所にすべてのタイマーオブザーバブルが存在しないため、この使用例には当てはまりません。このこと

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タイマオブザーバブルの配列を返す必要があります。
• 次に、Y個のオブザーバブルの配列をN回トレースする必要があります。

これはそうです。ちょっと素朴ですが、うまくいきます。

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'); 
      } 
     );