joplin/packages/lib/TaskQueue.js

const time = require('./time').default;
const Setting = require('./models/Setting').default;
const Logger = require('./Logger').default;

class TaskQueue {
	constructor(name) {
		this.waitingTasks_ = [];
		this.processingTasks_ = {};
		this.processingQueue_ = false;
		this.stopping_ = false;
		this.results_ = {};
		this.name_ = name;
		this.logger_ = new Logger();
	}

	concurrency() {
		return Setting.value('sync.maxConcurrentConnections');
	}

	push(id, callback) {
		if (this.stopping_) throw new Error('Cannot push task when queue is stopping');

		this.waitingTasks_.push({
			id: id,
			callback: callback,
		});
		this.processQueue_();
	}

	processQueue_() {
		if (this.processingQueue_ || this.stopping_) return;

		this.processingQueue_ = true;

		const completeTask = (task, result, error) => {
			delete this.processingTasks_[task.id];

			const r = {
				id: task.id,
				result: result,
			};

			if (error) r.error = error;

			this.results_[task.id] = r;

			this.processQueue_();
		};

		while (this.waitingTasks_.length > 0 && Object.keys(this.processingTasks_).length < this.concurrency()) {
			if (this.stopping_) break;

			const task = this.waitingTasks_.splice(0, 1)[0];
			this.processingTasks_[task.id] = task;

			task
				.callback()
				.then(result => {
					completeTask(task, result, null);
				})
				.catch(error => {
					if (!error) error = new Error('Unknown error');
					completeTask(task, null, error);
				});
		}

		this.processingQueue_ = false;
	}

	isWaiting(taskId) {
		return this.waitingTasks_.find(task => task.id === taskId);
	}

	isProcessing(taskId) {
		return taskId in this.processingTasks_;
	}

	isDone(taskId) {
		return taskId in this.results_;
	}

	async waitForResult(taskId) {
		if (!this.isWaiting(taskId) && !this.isProcessing(taskId) && !this.isDone(taskId)) throw new Error(`No such task: ${taskId}`);

		while (true) {
			// if (this.stopping_) {
			// 	return {
			// 		id: taskId,
			// 		error: new JoplinError('Queue has been destroyed', 'destroyedQueue'),
			// 	};
			// }

			const task = this.results_[taskId];
			if (task) return task;
			await time.sleep(0.1);
		}
	}

	async stop() {
		this.stopping_ = true;

		this.logger_.info(`TaskQueue.stop: ${this.name_}: waiting for tasks to complete: ${Object.keys(this.processingTasks_).length}`);

		// In general it's not a big issue if some tasks are still running because
		// it won't call anything unexpected in caller code, since the caller has
		// to explicitely retrieve the results
		const startTime = Date.now();
		while (Object.keys(this.processingTasks_).length) {
			await time.sleep(0.1);
			if (Date.now() - startTime >= 30000) {
				this.logger_.warn(`TaskQueue.stop: ${this.name_}: timed out waiting for task to complete`);
				break;
			}
		}

		this.logger_.info(`TaskQueue.stop: ${this.name_}: Done, waited for ${Date.now() - startTime}`);
	}

	isStopping() {
		return this.stopping_;
	}
}

TaskQueue.CONCURRENCY = 5;

module.exports = TaskQueue;