The abstract problem
Is there any way to consume the result of a mergeMap
in the original order of the outer observable, while still allowing the inner observables to run in parallel?
More detailed explanation
Let's look at two merge-mapping operators:
-
...which takes a mapping callback, and a number of how many inner observables may run concurrently:
of(1, 2, 3, 4, 5, 6).pipe( mergeMap(number => api.get('/double', { number }), 3) );
See it here in action: https://codepen.io/JosephSilber/pen/YzwVYNb?editors=1010
This will fire off 3 parallel requests for
1
,2
and3
, respectively. As soon as one of those requests completes, it'll fire off another request for4
. And so on and so forth, always maintaining 3 concurrent requests, until all of the values have been processed.However, since previous requests may complete before subsequent requests, the values produced may be out of order. So instead of:
[2, 4, 6, 8, 10, 12]
...we may actually get:
[4, 2, 8, 10, 6, 12] // or any other permutation
-
...enter
concatMap
. This operator ensures that the observables are all concatenated in the original order, so that:of(1, 2, 3, 4, 5, 6).pipe( concatMap(number => api.get('/double', { number })) );
...will always produce:
[2, 4, 6, 8, 10, 12]
See it here in action: https://codepen.io/JosephSilber/pen/OJMmzpy?editors=1010
This is what we want, but now the requests won't run in parallel. As the documentation says:
concatMap
is equivalent tomergeMap
withconcurrency
parameter set to1
.
So back to the question: is it possible to get the benefits of mergeMap
, whereby a given amount of requests can be run in parallel, while still having the mapped values be emitted in the original order?
My concrete problem
The above described the problem in abstract. It is sometimes easier to reason about a problem when you know the actual problem at hand, so here goes:
I have a list of orders that have to be shipped:
const orderNumbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10];
I have a
shipOrder
method that actually ships the orders. It returns aPromise
:const shipOrder = orderNumber => api.shipOrder(orderNumber);
The API can only process up to 5 order shipments simultaneously, so I'm using
mergeMap
to handle that:from(orderNumbers).pipe( mergeMap(orderNumber => shipOrder(orderNumber), 5) );
After an order is shipped, we need to print its shipping label. I have a
printShippingLabel
function that, given the order number of a shipped order, will print its shipping label. So I subscribe to our observable, and print the shipping labels as the values come in:from(orderNumbers) .pipe(mergeMap(orderNumber => shipOrder(orderNumber), 5)) .pipe(orderNumber => printShippingLabel(orderNumber));
This works, but now the shipping labels are printed out of order, since
mergeMap
emits values based on whenshipOrder
completes its request. What I want is for the labels to print in the same order as the original list.
Is that possible?
Visualization
See here for a visualization of the problem: https://codepen.io/JosephSilber/pen/YzwVYZb?editors=1010
You can see that earlier orders are being printed before later orders are even shipped.