backend/libs/async-queue.ts

import { EventEmitter } from 'events';

interface DSPromiseOption {
	timeout?: number;
}

export function destructPromise<T>(
	options: DSPromiseOption = {}
): [promise: Promise<T>, resolve: (value: T | PromiseLike<T>) => void, reject: (reason?: any) => void] {
	const { timeout } = options;
	let rs: (value: T | PromiseLike<T>) => void;
	let rj: (reason: any) => void;

	return [
		new Promise((resolve, reject) => {
			rs = resolve;
			rj = reject;

			if (timeout >= 0) setTimeout(rj, timeout, 'timeout');
		}),
		rs,
		rj,
	];
}

type AsyncTask = [fn: (data: any) => Promise<any>, payload: any, resolve: (data: any) => void, reject: (reason: any) => void];

export class AsyncQueue extends EventEmitter {
	private working = false;

	tasks: AsyncTask[];

	constructor() {
		super();
		this.working = false;
		this.tasks = [];
		process.nextTick(() => {
			this.emit('idle');
		});
	}

	private async _digest(item: AsyncTask) {
		this.working = true;

		const [taskFn, payload, resolve, reject] = item;
		await taskFn(payload).then(resolve, reject);

		if (this.tasks.length > 0) {
			await this._digest(this.tasks.shift());
		} else {
			this.working = false;
			this.emit('idle');
		}
	}

	/**
	 * 添加队列任务
	 * @param task
	 * @param options
	 */
	addTask(task: [AsyncTask[0], AsyncTask[1]], { timeout = 600000 }: { timeout?: number } = {}): Promise<any> {
		const [promise, resolve, reject] = destructPromise({ timeout });

		if (this.working) {
			this.tasks.push([...task, resolve, reject]);
		} else {
			this._digest([...task, resolve, reject]);
		}

		return promise;
	}
}