sequentialize/sequential-queue.js

"use strict";

// FIXME: Make this a stand-alone package; it should be useful as a general-purpose mechanism for sequentializing multiple asynchronous operations that may originate from multiple different callsites (like with read/peek/abort)

const Promise = require("bluebird");
const pDefer = require("p-defer");
const unreachable = require("@joepie91/unreachable")("@promistream/sequentialize"); // FIXME: Change name when moved out into a stand-alone package

module.exports = function createSequentialQueue() {
	/* TODO: Does this need a more efficient FIFO queue implementation? */
	let queue = [];
	let processing = false;
	
	function nextItem() {
		if (queue.length > 0) {
			let item = queue.shift();
			item();
		} else {
			unreachable("Tried to process an item from an empty queue");
		}
	}

	function tryStart() {
		if (processing === false) {
			processing = true;
			nextItem();
		}
	}

	function markCompleted() {
		if (queue.length > 0) {
			nextItem();
		} else {
			processing = false;
		}
	}

	return function withQueue(callback) {
		let { resolve, promise } = pDefer();
		queue.push(resolve);
		tryStart();

		return Promise.try(() => {
			return promise;
		}).then(() => {
			return callback();
		}).tap(() => {
			markCompleted();
		});
	};
}