Simple implementation of a promise queue that allows only some number of concurrent tasks to execute at once

README.md

promise-queue.js

Simple implementation of a Promises/A+ queue module.

Exports a single function that creates a new queue, and can be passed a configuration object with the following keys:

• concurrency: The maximum number of concurrently-executing functions. (Default: 1)
• Promise: Promise constructor. (Default: global.Promise)

The return value is the enqueue function, which accepts a single function as its argument. It will return a promise for the result of that function. The function is considered executing until this promise resolves; that is, if the function returns a promise, it will still be considered executing until that promise is settled.

'use strict';

const promiseQueue = require('./promise-queue.js');

let queue = promiseQueue({ concurrency: 2 });

[2000, 1000, 1500, 250].forEach(d =>
  queue(() => {
    console.log('starting:', d);
    return new Promise(r => setTimeout(() => r(d), d));
  })
  .then(r => console.log('finished:', r))
);

Output:

starting: 2000
starting: 1000
starting: 1500
finished: 1000
starting: 250
finished: 2000
finished: 250
finished: 1500

(The output order may depend on the promise implementation; the settling of the promise returned from the enqueue function may happen immediately before or after the next task is started, as both actions are triggered by the settling of the internal promise for the result of the task function. In other words, the output shown above makes it appear that three tasks are being executed at once, but in reality there is a pause after the second line of output. The third and fourth show up together a second later because the 1000 delay task finishes, causing the queue to start the 1500 task. The promise implementation informs the queue that the promise is settled before propagating the result to the promise returned by the enqueue function for the 1000 delay task.)

promise-queue.js

'use strict';

function promiseQueue(options) {
    options = Object.assign({
        concurrency: 1,
        Promise: global.Promise
    }, options);

    if (typeof options.Promise !== 'function') {
        throw new Error('no promise implementation');
    }

    let queue = [];

    let running = 0;

    function completed() {
        running -= 1;
        processQueue();
    }

    function processQueue() {
        while (running < options.concurrency && queue.length) {
            let state = queue.shift();

            running += 1;

            let fn = state.f;
            let fnPromise = new options.Promise(r => r(fn()));

            state.r(fnPromise);

            fnPromise.then(completed, completed);
        }
    }

    return function enqueue(f) {
        return new options.Promise(r => {
            queue.push({ f, r });

            processQueue();
        });
    };
}

module.exports = promiseQueue;