-
Notifications
You must be signed in to change notification settings - Fork 0
/
index.js
119 lines (105 loc) · 3.05 KB
/
index.js
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
112
113
114
115
116
117
118
119
(function (root, factory) {
// eslint-disable-next-line no-undef
if (typeof define === "function" && define.amd) {
// eslint-disable-next-line no-undef
define("fifo-process-queue", [], factory);
// eslint-disable-next-line no-undef
} else if (typeof module === "object" && module.exports) {
// eslint-disable-next-line no-undef
module.exports = factory();
} else {
root.FIFOProcessQueue = factory();
}
// eslint-disable-next-line no-undef
})(typeof self !== "undefined" ? self : this, function () {
function simpleFIFOProcessQueue(processor) {
var queue = [];
function done() {
queue.shift();
if (queue.length > 0) {
processor(queue[0], done);
}
}
function push(data) {
queue.push(data);
if (queue.length === 1) {
processor(queue[0], done);
}
}
function pushAll(data) {
data.forEach(push);
}
return {
push: push,
pushAll: pushAll,
};
}
// 2^53 is the largest power of 2 for which 2^x !== 2^x - 1
var maxId = Math.pow(2, 53);
function postProcessingFIFOProcessQueue(processor, postProcessor, maxConcurrency) {
var currentId = 0;
var pending = [];
var processing = [];
function nextId() {
currentId = (currentId + 1) % maxId;
return currentId;
}
function startPending() {
while (processing.length < maxConcurrency && pending.length > 0) {
var item = {
id: nextId(),
data: pending.shift(),
done: false,
};
var callback = (function (id) {
return function () {
done(id);
};
})(item.id);
processing.push(item);
processor(item.data, callback);
}
}
function done(id) {
var item;
for (var i = 0; i < processing.length; i++) {
item = processing[i];
if (item.id === id && !item.done) {
item.done = true;
}
}
while (processing.length > 0 && processing[0].done) {
item = processing.shift();
postProcessor(item.data);
}
startPending();
}
function push(data) {
pending.push(data);
startPending();
}
function pushAll(data) {
data.forEach(push);
}
return {
push: push,
pushAll: pushAll,
};
}
return function (processor, postProcessor, maxConcurrency) {
if (typeof processor !== "function") {
throw new Error("processor must be a function");
}
if (typeof postProcessor !== "function" && typeof postProcessor !== "undefined") {
throw new Error("postProcessor must be a function");
}
if (typeof maxConcurrency === "undefined") {
maxConcurrency = Number.MAX_VALUE;
} else if (typeof maxConcurrency !== "number") {
throw new Error("maxConcurrency must be a number");
} else if (maxConcurrency < 1) {
throw new Error("maxConcurrency must be at least 1");
}
return postProcessor ? postProcessingFIFOProcessQueue(processor, postProcessor, maxConcurrency) : simpleFIFOProcessQueue(processor);
};
});