RxJS эквивалент Async.js mapLimit

Question

Async.js mapLimit и его семейство функций <name>Limit в основном работают как семафор: они позволяют запускать ограниченное количество задач одновременно, в то время как дополнительные входящие задачи добавляются в очередь. Очередь становится производителем (холодным? Подключенным?). Запуск задачи сбрасывает элемент из очереди, как только появляется место (одна из его задач заканчивается).

Таким образом, ограниченное количество одновременных задач всегда активны.

Как достичь аналогичной функциональности в RxJS?

Answer 1

Сочетание defer и flatMapWithMaxConcurrent является RxJs способ сделать это:

// returns a promise 
function runSomeJob(input) { ... } 

function runSomeJobObservable(input) { 
    return Rx.Observable.defer(function() { 
     return runSomeJob(input); 
    }); 
} 

var inputStream = // some Rx.Observable 

// only allow 5 jobs to run concurrently 
var outputStream = inputStream 
    .flatMapWithMaxConcurrent(5, runSomeJobObservable); 

ouputStream.subscribe(...);