Nodejs Create Simple Queue On Async

in this below sample code i can run multi function on the same time Promise.all([sendMoneyToRequestedUser(_data), saveTransferMoneyTransaction(_data)]) .then(function (results)

Solution 1:

I was having issues while trying to use all kinds of modules, and finally I wrote the simplest implementation I could think of for this kind of work.

Check out this simple class I wrote (Plain JS):

classQueue{
    constructor(maxSimultaneously = 1) {
        this.maxSimultaneously = maxSimultaneously;
        this.__active = 0;
        this.__queue = [];
    }

    /** @param { () => Promise<T> } func 
     * @template T
     * @returns {Promise<T>}
    */
    async enqueue(func) {
        if(++this.__active > this.maxSimultaneously) {
            await new Promise(resolve => this.__queue.push(resolve));
        }

        try {
            return await func();
        } catch(err) {
            throw err;
        } finally {
            this.__active--;
            if(this.__queue.length) {
                this.__queue.shift()();
            }
        }
    }
}

Use it like this:

Lets say you have this async function:

const printNumber = async (n) => {
    awaitnewPromise(res =>setTimeout(res, 2000)); // wait 2 secconsole.log(n);
}

so, instead of:

await printNumber(1);
await printNumber(2);
await printNumber(3);
await printNumber(4);

use:

const q = newQueue();

q.enqueue(() =>printNumber(1));
q.enqueue(() =>printNumber(2));
q.enqueue(() =>printNumber(3));
q.enqueue(() =>printNumber(4));

Each function will be executed once the others are done.

The output:

1// after 2 sec2// after 4 sec3// after 6 sec4// after 8 sec

Or you can limit the queue to run up to some number of functions at the same time:

const q = newQueue(3);

q.enqueue(() =>printNumber(1));
q.enqueue(() =>printNumber(2));
q.enqueue(() =>printNumber(3));
q.enqueue(() =>printNumber(4));

The output:

1// after 2 sec2// after 2 sec3// after 2 sec4// after 4 sec

Also, the enqueue method will return/throw the original data from your promise!

Lets suppose you write an API to upload files, and you want to limit the uploads to up to 5 at the same time. You want everything to stay the same as it was, without changing your flow. Here's how you can do that:

asyncfunctionupload(data) {
    // upload...if(something) {
        return200;
    } else {
        throw400;
    }
}

So, instead of doing this:

asyncfunctionwork(data) { 
    // do something...returnawaitupload(data);
}

do this:

const q = newQueue(5); // up to 5 at the same timeasyncfunctionwork(data) { 
    // do something...returnawait q.enqueue(() =>upload(data));
}

classQueue {
    constructor(maxSimultaneously = 1) {
        this.maxSimultaneously = maxSimultaneously;
        this.__active = 0;
        this.__queue = [];
    }

    /** @param { () => Promise<T> } func 
     * @templateT
     * @returns {Promise<T>}
    */asyncenqueue(func) {
        if(++this.__active > this.maxSimultaneously) {
            awaitnewPromise(resolve =>this.__queue.push(resolve));
        }

        try {
            returnawaitfunc();
        } catch(err) {
            throw err;
        } finally {
            this.__active--;
            if(this.__queue.length) {
                this.__queue.shift()();
            }
        }
    }
}

const printNumber = async (n) => {
    awaitnewPromise(res =>setTimeout(res, 2000)); // wait 2 secconsole.log(n);
}

asyncfunctionstart() {
    console.log('starting...');
    const q = newQueue();
    
    q.enqueue(() =>printNumber(1));
    q.enqueue(() =>printNumber(2));
    q.enqueue(() =>printNumber(3));
    q.enqueue(() =>printNumber(4));
}

Click this to run 1 log per 2 sec: <buttononclick="start();">Start</button>

classQueue {
    constructor(maxSimultaneously = 1) {
        this.maxSimultaneously = maxSimultaneously;
        this.__active = 0;
        this.__queue = [];
    }

    /** @param { () => Promise<T> } func 
     * @templateT
     * @returns {Promise<T>}
    */asyncenqueue(func) {
        if(++this.__active > this.maxSimultaneously) {
            awaitnewPromise(resolve =>this.__queue.push(resolve));
        }

        try {
            returnawaitfunc();
        } catch(err) {
            throw err;
        } finally {
            this.__active--;
            if(this.__queue.length) {
                this.__queue.shift()();
            }
        }
    }
}

const printNumber = async (n) => {
    awaitnewPromise(res =>setTimeout(res, 2000)); // wait 2 secconsole.log(n);
}

asyncfunctionstart() {
    console.log('starting...');
    const q = newQueue(3);
    
    q.enqueue(() =>printNumber(1));
    q.enqueue(() =>printNumber(2));
    q.enqueue(() =>printNumber(3));
    q.enqueue(() =>printNumber(4));
}

Click this to run up to 3 logs every 2 sec: <buttononclick="start();">Start</button>

Solution 2:

I think that you're looking for this async.series. Is runs an array of functions one after another in sequential order and passes an array of results to the callbacks from the previous functions.

Example

varasync = require('async');

async.series([
    function(callback) {
        //Do a bunch of relevant stuffcallback(null, 'First function');
    },
    function(callback) {
        //Do some more relevant stuffcallback(null, 'Second function');
    }
],
function(err, results) {
    console.log(results); //Logs ['First function', 'Second function']
});

Solution 3:

UPDATE: The question is not really clear on what is given, and what is desired. From the comments, I've got the following:

three functions of check, send, and post are given and it is desired to be run in series.

Let's say that the mentioned check, send, and post all return Promises, e.g.:

functioncheck() {
  returnnewPromise(function (resolve, reject) {
    // Do some stuff and save them in a var 'results'var results = ...;
    resolve(results);
  });
}

Then you can simply build a queue as follows:

check()
  .then(send)
  .then(post)
  .catch(function(err) { console.error(err); })

At each level the function is called with the value resolved from the previous level.