RxJSですべての約束が達成されるまでどのように待つのですか？

Question

私は結果を得てページに表示する約束のリストを持っています。

let directives = ['object-src', 'img-src', 'media-src', 'script-src', 'style-src', 'frame-src', 'child-src']; 

let directivesStream = Rx.Observable.from(directives) 
    .map((d) => { 
    let q = getSummaryCSP({ 
     origin: origin, 
     directive: d, 
     from: moment().subtract(3, 'days').utc().format(), 
     to: moment().utc().format() 
    }); 

    return Rx.Observable.interval(10000) 
     .flatMap(() => Rx.Observable.fromPromise(queries.query(q))); 
    }); 

directivesStream.mergeAll().subscribe((response) => { 
    let flattenedRes = _.sortBy(response.data, (e) => e.event.count); 
    flattenedRes = _.reverse(flattenedRes); 
    this.setState({ 
    cspReport: flattenedRes 
    }); 
}); 

だから、それはやっているもの、それは10000の間隔と約束のストリームを作成しています。しかし、最終的な結果はすべての約束の合算結果であり、その理由は私がsortingとreversingを行っていて、結果をReactに表示しているからです。

私はRxJsのすべてがストリームであることを知っています。だから、私はこれをまったく達成できるかどうか分からないのですか？

私の質問は、すべてのストリームを1つのストリームにマージする必要があるため、すべての約束の結果を使用できるように、最後の10秒に約束のストリームを作成する方法です。

Answer 1

あなたが調査するために、私はいくつかのポインタをうまくあなたの質問を理解しますが、ここにいる場合はわからない：

• toArrayオペレータは：配列にストリームのすべての排出量を収集できるよう
• forkJoin演算子：Promise.allに相当し、すべての約束を並行して実行し、すべての結果を集めます。

私は確信していますあなたの目標を達成することができるそれらの2つのうちの1つを持つ帽子。あなた自身が問題を解決しようとしているのがRxjをよく理解する最良の方法であるかどうかを理解できるようにします。さらに、演算子hereのリストを見つけることができます。私は質問を投稿する前に一見することをお勧めします。

Answer 2

ノードモジュールをReactのクライアントサイドコードにインポートできるので、ノードモジュールAsyncを見ていきます。私はあなたが実行したい実行プロセスはマップと滝の組み合わせであると信じています。

async.waterfall([ 
    myFirstFunction, 
    mySecondFunction, 
    myLastFunction, 
], function (err, result) { 
    // result now equals 'done' 
}); 
function myFirstFunction(callback) { 
    callback(null, 'one', 'two'); 
} 
function mySecondFunction(arg1, arg2, callback) { 
    // arg1 now equals 'one' and arg2 now equals 'two' 
    callback(null, 'three'); 
} 
function myLastFunction(arg1, callback) { 
    // arg1 now equals 'three' 
    callback(null, 'done'); 
}