Подождите, прикованные наблюдаемые для завершения

Question

Имея некоторые проблемы в достижении того, что я хочу с RxJS5 - У меня есть простая Наблюдаемая цепь, начались с 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] 

это ясно, что мы не ждем всех тим er наблюдаемых до конца, так как вы увидите «все сделано!». многократно заносится в журнал, вкрапленное «Все должны регистрироваться в первую очередь».

Как я могу получить вывод, который я ищу?

Для этого мы можем использовать zip, но API для zip не подходит для этого прецедента, потому что у нас нет всех наблюдателей таймера в одном месте одновременно!

Если мой вопрос не был достаточно ясен, здесь является аналогом того, что я хочу сделать, мы блокируем все обратные вызовы, пока мы произвольно не закончим, и мы собрали все результаты:

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.
• Далее следует проследить 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'); 
      } 
     );