Управление памятью с помощью 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.