Node.JS Server Queue Processor

Question

I implemented a simple queuing system for my Node.JS app and wanted a critique on it's structure.

const TPS = 20;
const Queue = {
    counter: 1,
    items: {},
    /**
     * Add an item to the queue, with the given func to call
     * @param {Function} func
     * @param {Boolean} repeating
     * @return {Number}
     */
    add(func, repeating = false) {
        const id = this.counter++;
        this.items[id] = {func, repeating, id};
        return id;
    },
    /**
     * Remove an item from the queue with the given id
     * @param {Number} id
     */
    remove(id) {
        if (this.items.hasOwnProperty(id)) {
            delete this.items[id];
        }
    },
    /**
     * Process items in the queue
     */
    process() {
        for (let id in this.items) {
            // Prevent this item from being processed again
            if (!this.items.hasOwnProperty(id) || this.items[id].processing) {
                continue;
            }

            // Delete this item when it's scheduled for deletion
            if (this.items[id].scheduledForDeletion) {
                delete this.items[id];
                continue;
            }

            // Let the queue know this item is being processed and
            // it's scheduled deletion status
            this.items[id].processing = true;
            this.items[id].scheduledForDeletion = !this.items[id].repeating;

            // Don't wait for item's promise to resolve, since this
            // will create a backlog on the queue
            (async () => {
                try {
                    await this.items[id].func.call(null);
                } catch (err) {
                    // TODO: Handle errors.
                    console.error(err);
                }

                this.items[id].processing = false;
            })();
        }
    }
};

(function tick() {
    setTimeout(tick, 1000 / TPS);
    Queue.process();
})();

This is an example of how it's implemented.

// Add three items to the queue: 1 normal, 1 async and 1 repeating
Queue.add(() => console.info(`[tick] -> ${Date.now()}`));
Queue.add(async () => setTimeout(() => console.info(`[async] -> ${Date.now()}`), 100));
const timeLoop = Queue.add(() => console.info(`[loop] time (loop) -> ${Date.now()}`), true);

// Remove the looping item from the queue
setTimeout(() => Queue.remove(timeLoop), 500);

The idea is to have this run when the server starts and continually process the queue items. Queue is in it's own file and exported. I import this file into my controllers and call (for example) Queue.add('function to add user to DB and send out email').

There's certainly some critiques that could be made about your code, but I'm having difficulty understanding why it is needed. Is there some reason you don't just fire and forget your async functions where you need to instead of introducing more complexity with a "Queue" (which I'd claim really isn't a queue)? — Gerrit0
– Gerrit0, Commented Nov 28, 2018 at 18:05
You are right, for the one-time executions, you could argue that it's unnecessary to add them to this queue. However for the repeating (long running) processes, it is unnecessary to add them to this queue. Mostly this queue is used in a game that I'm building, where a lot of things happen that need to be processed per-tick. — Enijar
– Enijar, Commented Nov 29, 2018 at 18:02
What are you trying to determine? What is the purpose of the immediately invoked async arrow function within process; and async at Queue.add(async () => setTimeout(() => console.info(`[async] -> ${Date.now()}`), 100)) where no Promise is included within or returned from the function passed to Queue.add()? Is Queue.add() and process expected to handle both asynchronous and synchronous functions as parameters? — guest271314
– guest271314, Commented Dec 1, 2018 at 1:33
I invoke the arrow function inside the process method since the await keyword needs to be inside an async body, but I think it might be better to move it to the process function definition like async process(). The Queue.add calls at the end of my code are just examples to show you that you can add async items to the queue. And actually async returns a promise by default so technically they are returning promises. Finally, yes, the process method handles both sync and async items. — Enijar
– Enijar, Commented Dec 1, 2018 at 16:05
@Enijar Where does async () => setTimeout(() => console.info(`[async] -> ${Date.now()}`), 100) return any value? See Why is value undefined at .then() chained to Promise? — guest271314
– guest271314, Commented Dec 1, 2018 at 19:52

Sᴀᴍ Onᴇᴌᴀ · Accepted Answer · 2018-12-03 04:46:03Z

The structure looks fine. It is quite succinct and makes good use of const and let where appropriate.

To adhere to the D.R.Y. principle, process() can and should utilize remove() to remove items from the queue.

I considered suggesting that arguments be accepted with each function but that can be achieved with partially bound functions.

I also considered suggesting you consider using a class, since ES-6 featured can be utilized, but then you would either need to instantiate a queue once or else make all methods static.

I would suggest you consider accepting an error handler callback for each function. That way, instead of writing all errors to the console, the caller could add an appropriate handler.

Stack Exchange Network

Node.JS Server Queue Processor

1 Answer 1

You must log in to answer this question.

Hot Network Questions

Node.JS Server Queue Processor

1 Answer 1

You must log in to answer this question.

Related

Hot Network Questions