source: frontend/node_modules/webpack/lib/util/AsyncQueue.js

Last change on this file was 9af201e, checked in by MBK <marija.karapandzova@…>, 12 days ago

Fix frontend appearance
  • Property mode set to 100644
File size: 11.1 KB
Line 
1/*
2 MIT License http://www.opensource.org/licenses/mit-license.php
3 Author Tobias Koppers @sokra
4*/
5
6"use strict";
7
8const { AsyncSeriesHook, SyncHook } = require("tapable");
9const { makeWebpackError } = require("../errors/HookWebpackError");
10const WebpackError = require("../errors/WebpackError");
11const ArrayQueue = require("./ArrayQueue");
12
13const QUEUED_STATE = 0;
14const PROCESSING_STATE = 1;
15const DONE_STATE = 2;
16
17let inHandleResult = 0;
18
19/**
20 * Defines the callback callback.
21 * @template T
22 * @callback Callback
23 * @param {(WebpackError | null)=} err
24 * @param {(T | null)=} result
25 * @returns {void}
26 */
27
28/**
29 * Represents AsyncQueueEntry.
30 * @template T
31 * @template K
32 * @template R
33 */
34class AsyncQueueEntry {
35 /**
36 * Creates an instance of AsyncQueueEntry.
37 * @param {T} item the item
38 * @param {Callback<R>} callback the callback
39 */
40 constructor(item, callback) {
41 this.item = item;
42 /** @type {typeof QUEUED_STATE | typeof PROCESSING_STATE | typeof DONE_STATE} */
43 this.state = QUEUED_STATE;
44 /** @type {Callback<R> | undefined} */
45 this.callback = callback;
46 /** @type {Callback<R>[] | undefined} */
47 this.callbacks = undefined;
48 /** @type {R | null | undefined} */
49 this.result = undefined;
50 /** @type {WebpackError | null | undefined} */
51 this.error = undefined;
52 }
53}
54
55/**
56 * Defines the get key type used by this module.
57 * @template T, K
58 * @typedef {(item: T) => K} getKey
59 */
60
61/**
62 * Defines the processor type used by this module.
63 * @template T, R
64 * @typedef {(item: T, callback: Callback<R>) => void} Processor
65 */
66
67/**
68 * Represents AsyncQueue.
69 * @template T
70 * @template K
71 * @template R
72 */
73class AsyncQueue {
74 /**
75 * Creates an instance of AsyncQueue.
76 * @param {object} options options object
77 * @param {string=} options.name name of the queue
78 * @param {number=} options.parallelism how many items should be processed at once
79 * @param {string=} options.context context of execution
80 * @param {AsyncQueue<EXPECTED_ANY, EXPECTED_ANY, EXPECTED_ANY>=} options.parent parent queue, which will have priority over this queue and with shared parallelism
81 * @param {getKey<T, K>=} options.getKey extract key from item
82 * @param {Processor<T, R>} options.processor async function to process items
83 */
84 constructor({ name, context, parallelism, parent, processor, getKey }) {
85 this._name = name;
86 this._context = context || "normal";
87 this._parallelism = parallelism || 1;
88 this._processor = processor;
89 this._getKey =
90 getKey ||
91 /** @type {getKey<T, K>} */ ((item) => /** @type {T & K} */ (item));
92 /** @type {Map<K, AsyncQueueEntry<T, K, R>>} */
93 this._entries = new Map();
94 /** @type {ArrayQueue<AsyncQueueEntry<T, K, R>>} */
95 this._queued = new ArrayQueue();
96 /** @type {AsyncQueue<T, K, R>[] | undefined} */
97 this._children = undefined;
98 this._activeTasks = 0;
99 this._willEnsureProcessing = false;
100 this._needProcessing = false;
101 this._stopped = false;
102 /** @type {AsyncQueue<T, K, R>} */
103 this._root = parent ? parent._root : this;
104 if (parent) {
105 if (this._root._children === undefined) {
106 this._root._children = [this];
107 } else {
108 this._root._children.push(this);
109 }
110 }
111
112 this.hooks = {
113 /** @type {AsyncSeriesHook<[T]>} */
114 beforeAdd: new AsyncSeriesHook(["item"]),
115 /** @type {SyncHook<[T]>} */
116 added: new SyncHook(["item"]),
117 /** @type {AsyncSeriesHook<[T]>} */
118 beforeStart: new AsyncSeriesHook(["item"]),
119 /** @type {SyncHook<[T]>} */
120 started: new SyncHook(["item"]),
121 /** @type {SyncHook<[T, WebpackError | null | undefined, R | null | undefined]>} */
122 result: new SyncHook(["item", "error", "result"])
123 };
124
125 this._ensureProcessing = this._ensureProcessing.bind(this);
126 }
127
128 /**
129 * Returns context of execution.
130 * @returns {string} context of execution
131 */
132 getContext() {
133 return this._context;
134 }
135
136 /**
137 * Updates context using the provided value.
138 * @param {string} value context of execution
139 */
140 setContext(value) {
141 this._context = value;
142 }
143
144 /**
145 * Processes the provided item.
146 * @param {T} item an item
147 * @param {Callback<R>} callback callback function
148 * @returns {void}
149 */
150 add(item, callback) {
151 if (this._stopped) return callback(new WebpackError("Queue was stopped"));
152 this.hooks.beforeAdd.callAsync(item, (err) => {
153 if (err) {
154 callback(
155 makeWebpackError(err, `AsyncQueue(${this._name}).hooks.beforeAdd`)
156 );
157 return;
158 }
159 const key = this._getKey(item);
160 const entry = this._entries.get(key);
161 if (entry !== undefined) {
162 if (entry.state === DONE_STATE) {
163 if (inHandleResult++ > 3) {
164 process.nextTick(() => callback(entry.error, entry.result));
165 } else {
166 callback(entry.error, entry.result);
167 }
168 inHandleResult--;
169 } else if (entry.callbacks === undefined) {
170 entry.callbacks = [callback];
171 } else {
172 entry.callbacks.push(callback);
173 }
174 return;
175 }
176 const newEntry = new AsyncQueueEntry(item, callback);
177 if (this._stopped) {
178 this.hooks.added.call(item);
179 this._root._activeTasks++;
180 process.nextTick(() =>
181 this._handleResult(newEntry, new WebpackError("Queue was stopped"))
182 );
183 } else {
184 this._entries.set(key, newEntry);
185 this._queued.enqueue(newEntry);
186 const root = this._root;
187 root._needProcessing = true;
188 if (root._willEnsureProcessing === false) {
189 root._willEnsureProcessing = true;
190 setImmediate(root._ensureProcessing);
191 }
192 this.hooks.added.call(item);
193 }
194 });
195 }
196
197 /**
198 * Processes the provided item.
199 * @param {T} item an item
200 * @returns {void}
201 */
202 invalidate(item) {
203 const key = this._getKey(item);
204 const entry =
205 /** @type {AsyncQueueEntry<T, K, R>} */
206 (this._entries.get(key));
207 this._entries.delete(key);
208 if (entry.state === QUEUED_STATE) {
209 this._queued.delete(entry);
210 }
211 }
212
213 /**
214 * Waits for an already started item
215 * @param {T} item an item
216 * @param {Callback<R>} callback callback function
217 * @returns {void}
218 */
219 waitFor(item, callback) {
220 const key = this._getKey(item);
221 const entry = this._entries.get(key);
222 if (entry === undefined) {
223 return callback(
224 new WebpackError(
225 "waitFor can only be called for an already started item"
226 )
227 );
228 }
229 if (entry.state === DONE_STATE) {
230 process.nextTick(() => callback(entry.error, entry.result));
231 } else if (entry.callbacks === undefined) {
232 entry.callbacks = [callback];
233 } else {
234 entry.callbacks.push(callback);
235 }
236 }
237
238 /**
239 * Describes how this stop operation behaves.
240 * @returns {void}
241 */
242 stop() {
243 this._stopped = true;
244 const queue = this._queued;
245 this._queued = new ArrayQueue();
246 const root = this._root;
247 for (const entry of queue) {
248 this._entries.delete(
249 this._getKey(/** @type {AsyncQueueEntry<T, K, R>} */ (entry).item)
250 );
251 root._activeTasks++;
252 this._handleResult(
253 /** @type {AsyncQueueEntry<T, K, R>} */ (entry),
254 new WebpackError("Queue was stopped")
255 );
256 }
257 }
258
259 /**
260 * Increase parallelism.
261 * @returns {void}
262 */
263 increaseParallelism() {
264 const root = this._root;
265 root._parallelism++;
266 /* istanbul ignore next */
267 if (root._willEnsureProcessing === false && root._needProcessing) {
268 root._willEnsureProcessing = true;
269 setImmediate(root._ensureProcessing);
270 }
271 }
272
273 /**
274 * Decrease parallelism.
275 * @returns {void}
276 */
277 decreaseParallelism() {
278 const root = this._root;
279 root._parallelism--;
280 }
281
282 /**
283 * Checks whether this async queue is processing.
284 * @param {T} item an item
285 * @returns {boolean} true, if the item is currently being processed
286 */
287 isProcessing(item) {
288 const key = this._getKey(item);
289 const entry = this._entries.get(key);
290 return entry !== undefined && entry.state === PROCESSING_STATE;
291 }
292
293 /**
294 * Checks whether this async queue is queued.
295 * @param {T} item an item
296 * @returns {boolean} true, if the item is currently queued
297 */
298 isQueued(item) {
299 const key = this._getKey(item);
300 const entry = this._entries.get(key);
301 return entry !== undefined && entry.state === QUEUED_STATE;
302 }
303
304 /**
305 * Checks whether this async queue is done.
306 * @param {T} item an item
307 * @returns {boolean} true, if the item is currently queued
308 */
309 isDone(item) {
310 const key = this._getKey(item);
311 const entry = this._entries.get(key);
312 return entry !== undefined && entry.state === DONE_STATE;
313 }
314
315 /**
316 * Describes how this ensure processing operation behaves.
317 * @returns {void}
318 */
319 _ensureProcessing() {
320 while (this._activeTasks < this._parallelism) {
321 const entry = this._queued.dequeue();
322 if (entry === undefined) break;
323 this._activeTasks++;
324 entry.state = PROCESSING_STATE;
325 this._startProcessing(entry);
326 }
327 this._willEnsureProcessing = false;
328 if (this._queued.length > 0) return;
329 if (this._children !== undefined) {
330 for (const child of this._children) {
331 while (this._activeTasks < this._parallelism) {
332 const entry = child._queued.dequeue();
333 if (entry === undefined) break;
334 this._activeTasks++;
335 entry.state = PROCESSING_STATE;
336 child._startProcessing(entry);
337 }
338 if (child._queued.length > 0) return;
339 }
340 }
341 if (!this._willEnsureProcessing) this._needProcessing = false;
342 }
343
344 /**
345 * Processes the provided entry.
346 * @param {AsyncQueueEntry<T, K, R>} entry the entry
347 * @returns {void}
348 */
349 _startProcessing(entry) {
350 this.hooks.beforeStart.callAsync(entry.item, (err) => {
351 if (err) {
352 this._handleResult(
353 entry,
354 makeWebpackError(err, `AsyncQueue(${this._name}).hooks.beforeStart`)
355 );
356 return;
357 }
358 let inCallback = false;
359 try {
360 this._processor(entry.item, (e, r) => {
361 inCallback = true;
362 this._handleResult(entry, e, r);
363 });
364 } catch (err) {
365 if (inCallback) throw err;
366 this._handleResult(entry, /** @type {WebpackError} */ (err), null);
367 }
368 this.hooks.started.call(entry.item);
369 });
370 }
371
372 /**
373 * Processes the provided entry.
374 * @param {AsyncQueueEntry<T, K, R>} entry the entry
375 * @param {(WebpackError | null)=} err error, if any
376 * @param {(R | null)=} result result, if any
377 * @returns {void}
378 */
379 _handleResult(entry, err, result) {
380 this.hooks.result.callAsync(entry.item, err, result, (hookError) => {
381 const error = hookError
382 ? makeWebpackError(hookError, `AsyncQueue(${this._name}).hooks.result`)
383 : err;
384
385 const callback = /** @type {Callback<R>} */ (entry.callback);
386 const callbacks = entry.callbacks;
387 entry.state = DONE_STATE;
388 entry.callback = undefined;
389 entry.callbacks = undefined;
390 entry.result = result;
391 entry.error = error;
392
393 const root = this._root;
394 root._activeTasks--;
395 if (root._willEnsureProcessing === false && root._needProcessing) {
396 root._willEnsureProcessing = true;
397 setImmediate(root._ensureProcessing);
398 }
399
400 if (inHandleResult++ > 3) {
401 process.nextTick(() => {
402 callback(error, result);
403 if (callbacks !== undefined) {
404 for (const callback of callbacks) {
405 callback(error, result);
406 }
407 }
408 });
409 } else {
410 callback(error, result);
411 if (callbacks !== undefined) {
412 for (const callback of callbacks) {
413 callback(error, result);
414 }
415 }
416 }
417 inHandleResult--;
418 });
419 }
420
421 clear() {
422 this._entries.clear();
423 this._queued.clear();
424 this._activeTasks = 0;
425 this._willEnsureProcessing = false;
426 this._needProcessing = false;
427 this._stopped = false;
428 }
429}
430
431module.exports = AsyncQueue;
Note: See TracBrowser for help on using the repository browser.