Fire async request in parallel but get result in order using rxjs

Question

For example:

Get 5 pages in parrallel using jquery ajax. When page2 return, do nothing. When page1 return, do something with page1 and page2.

// assume there is some operator that can do this, 
// then it might look like this?
Rx.Observable.range(1, 5).
someOperator(function(page) {
  return Rx.Observable.defer( () => $.get(page) );
}).scan(function(preVal, curItem) {
  preVal.push(curItem);
  return preVal;
}, []);

Answer 1

There exists the forkJoin operator which will run all observable sequences in parallel and collect their last elements. (quoted from the documentation). But if you use that one, you will have to wait for all 5 promises to resolve, or one of the 5 to be in error. It is the close equivalent to RSVP.all or jQuery.when. So that would not allow you to do something once you have the second. I mention it anyways in case it might be useful to you in another case.

Another possibility is to use concatMap which will allow you to receive the resolved promises in order. However, I don't have it clear that they will be launched in parallel, the second promise should start only when the first one has resolved.

Last option I can think about is to use merge(2), that should run two promises in parallel, and at anytime they will only be two promises being 'launched'.

Now if you don't use defer, and you use concatMap, I believe you should have all AJAX request started, and still the right ordering. So you could write :

.concatMap(function(page) {
  return $.get(page);
})

Relevant documentation:

• merge(maxConcurrency)
• concatMap
• forkJoin

Answer 2

concatMap keeps the order, but processes elements in sequence.

mergeMap does not keep the order, but it runs in parallel.

I've created the operator mergeMapAsync below to make process elements (e.g. page downloads) in parallel but emit in order. It even supports throttling (e.g. download max. 6 pages in parallel).

Rx.Observable.prototype.mergeMapAsync = mergeMapAsync;
function mergeMapAsync(func, concurrent) {
    return new Rx.Observable(observer => {
        let outputIndex = 0;
        const inputLen = this.array ? this.array.length : this.source.array.length;
        const responses = new Array(inputLen);

        const merged = this.map((value, index) => ({ value, index })) // Add index to input value.
            .mergeMap(value => {
                return Rx.Observable.fromPromise(new Promise(resolve => {
                    console.log(`${now()}: Call func for ${value.value}`);  
                    // Return func retVal and index.
                    func(value.value).then(retVal => {
                        resolve({ value: retVal, index: value.index });
                    });
                }));
            }, concurrent);

        const mergeObserver = {
            next: (x) => {
                console.log(`${now()}: Promise returned for ${x.value}`);
                responses[x.index] = x.value;

                // Emit in order using outputIndex.
                for (let i = outputIndex, len = responses.length; i < len; i++) {
                    if (typeof responses[i] !== "undefined") {
                        observer.next(responses[i]);
                        outputIndex = i + 1;
                    } else {
                        break;
                    }
                }
            },
            error: (err) => observer.error(err),
            complete: () => observer.complete()
        };
        return merged.subscribe(mergeObserver);
    });
};

// ----------------------------------------
const CONCURRENT = 3;
var start = Date.now();
var now = () => Date.now() - start;

const array = ["a", "b", "c", "d", "e"];
Rx.Observable.from(array)
    .mergeMapAsync(value => getData(value), CONCURRENT)
    .finally(() => console.log(`${now()}: End`))
    .subscribe(value => {
        console.log(`${now()}: ${value}`); // getData
    });

function getData(input) {
    const delayMin = 500; // ms
    const delayMax = 2000; // ms

    return new Promise(resolve => {
        setTimeout(() => resolve(`${input}+`), Math.floor(Math.random() * delayMax) + delayMin);
    });
}

<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width">
  <title>mergeMapAsync</title>
  <script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.8/Rx.min.js"></script>
</head>
<body>

</body>
</html>