Deferred Promise Queue

Motivation

I recently published a simple npm package to defer the execution of collection of promise based asynchronous calls.

If you ever had to do multiple async calls returning a promise object, but the service or client machine was not able to handle all calls concurrently, there are chances that something like this was used.

An obvious use case would be getting data from the remote API page by page and it is not uncommon that it could be hundreds if not thousands pages long (which was by the way the initial motivation to create the library at first place).

Deferred pattern

At the core of the implementation is something which could be described as deferred Promise. We can work with an array of Promises hundreds or thousands items long, but we can't trigger them at the same time. So the the async function calls need to be deferred until the system is able to process them. How to do that I explained in this article about Deferred Promise

Publish - subscribe

The second piece in the puzzle is to implement ability to throttle the calls.
Under the hood this is done by implementing publish-subscribe pattern and schedule-task pattern.
The entire flow is triggered by calling start method which in turn emits event that the queue is available to take a call. (Fig.1). Promise Queue is scheduling the individual async function calls periodically (Fig.2) until the buffer is filled. Buffer here is the limit on currently pending promises. Once this limit is reached the scheduler postpones the next task until at least one pending promise is converted to settled (Fig.3).

Flow Chart

Fig.1, Initial call

Fig.2, periodic loop

Fig.3, coming back from blocked queue

Notice that all the parts (apart beginning) starts and end with emitting an event. This allows to add multiple consumers e.g. for logging or debugging.

Wrap up

If anybody find this useful I wrapped all of the above in the platform agnostic npm package, which can be installed by

$ npm install prmsq

The basic usage example:

// fetchPage(pageNumber) is http call returning promise
const workload = range(1, 1000)
    .map((pageNo) => () => fetchPage(pageNo));

const delayBetweenCalls = 100;
const concurentLimit = 10
const pq = new PrmsQ(workload, delayBetweenCalls, concurentLimit )

Promise.allSettled(pq.promises)
.then(doWhenAllSettled)
.catch(handleProblem)

pq.start();

It also can be found on github.com/webduvet/prsmq
with the example in /src/example/