RxJS Observable.concatMapによるメモリ管理？

Question

私はRxJSが新しく、ライブラリの使い方もまだ学んでいます。

警告：ソース値が無限に到着し、その 対応する内側の観測よりも高速に完了することができた場合、内部観測が待っ無限のバッファに蓄えるように、それはメモリ 問題になりますconcatMapの説明は次の警告を与えます 自分のターンを購読する。

これは、メモリを消費しますが、高速のconcatMapは、ゆっくりとしたconcatMapを提供するため、私にとっては問題です。これは次のように設定されています：

let uploadObs = Observable.range(0, blockCount).concatMap(blockIndex => { 
    // This part is fast and memory intensive. I'd like to use 
    // a bounded buffer here or something similar to control 
    // memory utilization 

    let blockReaderObs = ...; 
    // ... read a block from a large file object in blockReaderObs 
    return blockReaderObs; 
}).concatMap((blockData, index) => { 
    // This part involves a POST so is much slower than reading a 
    // file block 
    let objFromBlockData = someTransformation(blockData); 
    return this.http.post(someUrl, objFromBlockData) 
     .map(transformResponse); 
}); 

RxJSのこの種の問題に対処する正しい方法はありますか？

Answer 1

これは古典的な生産者 - 消費者の問題です。 バックプレッシャ演算子を使用して、処理のために送信される要素の数を制限することができます。 controlled streamsを参照してください。