Repository navigation
Expand file tree
/
Copy pathtask-queue.js
More file actions
111 lines (101 loc) · 3.46 KB
/
Copy pathtask-queue.js
File metadata and controls
111 lines (101 loc) · 3.46 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
const _ = require('lodash');
let taskTagCounter = 0;
class Task {
constructor(job, tag) {
this.tag = tag || 'task-' + taskTagCounter++;
this.promise = new Promise((resolve, reject) => {
this.run = function () {
return job().then(resolve, reject);
};
});
}
}
module.exports = class TaskQueue {
constructor(options) {
this.limit = _.get(options, 'limit', null);
this.interval = _.get(options, 'interval', null);
this.debug = _.get(options, 'debug', false);
this.runCount = 0;
this.lastRunTime = null;
this.timeout = null;
this.taskQueue = [];
if (this.interval) {
this._runTask = this._runTaskInterval.bind(this, this._runTask.bind(this));
}
if (this.limit) {
this._runTask = this._runTaskLimit.bind(this, this._runTask.bind(this));
}
}
addTask(job, tag) {
let task = new Task(job, tag);
this.taskQueue.push(task);
if (this.debug) {
console.log(`[TaskQueue] submitted task to queue, task: ${task.tag}, queue size: ${this.taskQueue.length}, running tasks: ${this.runCount}`);
}
this._executeNextTask();
return task.promise;
}
clearQueue() {
this.taskQueue = [];
if (this.timeout) {
clearTimeout(this.timeout);
this.timeout = null;
}
}
_executeNextTask() {
if (_.isEmpty(this.taskQueue)) {
return;
}
this._runTask();
}
_runTaskLimit(runTask) {
if (this.runCount < this.limit) {
runTask();
}
if (this.debug) {
console.log(
`[TaskQueue] task run count limit (${this.limit}) reached, queue size: ${this.taskQueue.length}, ` +
`running tasks: ${this.runCount}, waiting for previous tasks to finish`
);
}
}
_runTaskInterval(runTask) {
if (!this.lastRunTime) {
runTask();
} else if (!this.timeout) {
let diff = process.hrtime(this.lastRunTime);
let diffMs = diff[0] * 1000 + diff[1] / 1000000;
if (diffMs >= this.interval) {
runTask();
} else {
this.timeout = setTimeout(runTask, this.interval - diffMs);
if (this.debug) {
console.log(
`[TaskQueue] task interval is less than allowed (${this.interval}) reached, queue size: ` +
`${this.taskQueue.length}, running tasks: ${this.runCount}, waiting for ${this.interval - diffMs}ms`
);
}
}
}
}
_runTask() {
const task = this.taskQueue.shift();
if (this.timeout) {
clearTimeout(this.timeout);
this.timeout = null;
}
this.runCount += 1;
this.lastRunTime = process.hrtime();
if (this.debug) {
console.log(`[TaskQueue] running task: ${task.tag}, queue size: ${this.taskQueue.length}, running tasks: ${this.runCount}`);
}
task.run().then(() => {
this.runCount -= 1;
if (this.debug) {
console.log(`[TaskQueue] finished running task: ${task.tag}, queue size: ${this.taskQueue.length}, running tasks: ${this.runCount}`);
}
this._executeNextTask();
});
this._executeNextTask();
}
};