Leaky bucket

leakyBucket.js

/**
 * Leaky bucket queue callback.
 * @callback LeakyBucketCallback
 * @param {any|null} error Call error.
 * @param {any=} data Call result.
 */

/**
 * Leaky bucket queue.
 * @typedef LeakyBucketQueue
 * @property {Function} call Method to execute.
 * @property {LeakyBucketCallback} callback Callback to return the call response.
 */

/**
 * @typedef {object} LeakyBucket
 * @property {Array<LeakyBucketQueue>} queue Queue.
 * @property {ReturnType<typeof setTimeout>=} interval Interval.
 */

/** @type {LeakyBucket} */
const bucket = { queue: [] };
let dropInterval = 0;

/**
 * Drop the next call.
 */
const NEXT_DROP = async () => {
	if (bucket.queue.length > 0) {
		const { call, callback } = /** @type {LeakyBucketQueue} */ (bucket.queue.shift());

		try {
			const data = await call();
			callback(null, data);
		} catch (e) {
			callback(e);
		}
	} else {
		clearInterval(bucket.interval);
		delete bucket.interval;
	}
};

/**
 * Enqueue the given call to be executed.
 * The callback contains the caller's Promise resolution methods.
 * @template {Function} CallFunction
 * @param {CallFunction} call Function to call.
 * @returns {Promise<ReturnType<CallFunction>>} The response of call(params).
 */
const enqueueCall = async (call) =>
	new Promise((resolve, reject) => {
		/** @type {LeakyBucketCallback} */
		const callback = (err, data) => (err ? reject(err) : resolve(data));
		bucket.queue.push({ call, callback });
		if (!bucket.interval) {
			bucket.interval = setInterval(NEXT_DROP, dropInterval);
		}
	});

/**
 * Set drop interval for Leaky bucket.
 * @param {number} interval Drop interval in milliseconds.
 */
const setDropInterval = (interval) => {
	dropInterval = interval;
};

export const leakyBucket = {
	enqueueCall,
	setDropInterval,
};