src/vs/base/common/async.ts
2749 LOC · 2326 covered · 423 uncovered · 534 ranges · 11657 concepts · 201 introducers · 5963 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
>
/*---------------------------------------------------------------------------------------------
async.ts ×202
2
>
* Copyright (c) Microsoft Corporation. All rights reserved.
3
>
* Licensed under the MIT License. See License.txt in the project root for license information.
4
>
*--------------------------------------------------------------------------------------------*/
5
>
6
>
import { CancellationToken, CancellationTokenSource } from './cancellation.js';
7
>
import { BugIndicatingError, CancellationError, isCancellationError } from './errors.js';
8
>
import { Emitter, Event } from './event.js';
9
>
import { Disposable, DisposableMap, DisposableStore, IDisposable, isDisposable, MutableDisposable, toDisposable } from './lifecycle.js';
10
>
import { extUri as defaultExtUri, IExtUri } from './resources.js';
11
>
import { URI } from './uri.js';
12
>
import { setTimeout0 } from './platform.js';
13
>
import { MicrotaskDelay } from './symbols.js';
14
>
import { Lazy } from './lazy.js';
15
>
16
>
export function isThenable<T>(obj: unknown): obj is Promise<T> {
18
>
}
20
>
export interface CancelablePromise<T> extends Promise<T> {
21
>
cancel(): void;
22
>
}
23
>
24
>
/**
25
>
* Returns a promise that can be cancelled using the provided cancellation token.
26
>
*
27
>
* @remarks When cancellation is requested, the promise will be rejected with a {@link CancellationError}.
28
>
* If the promise resolves to a disposable object, it will be automatically disposed when cancellation
29
>
* is requested.
30
>
*
31
>
* @param callback A function that accepts a cancellation token and returns a promise
32
>
* @returns A promise that can be cancelled
33
>
*/
34
>
export function createCancelablePromise<T>(callback: (token: CancellationToken) => Promise<T>): CancelablePromise<T> {
36
>
37
>
const thenable = callback(source.token);
38
>
39
>
let isCancelled = false;
40
>
41
>
const promise = new Promise<T>((resolve, reject) => {
42
>
const subscription = source.token.onCancellationRequested(() => {
44
>
subscription.dispose();
45
>
reject(new CancellationError());
47
>
Promise.resolve(thenable).then(value => {
49
>
source.dispose();
50
>
51
>
if (!isCancelled) {
53
>
56
>
// be cleaned up
57
>
value.dispose();
58
>
}
61
>
source.dispose();
62
>
reject(err);
64
>
});
65
>
66
>
return <CancelablePromise<T>>new class {
67
>
cancel() {
69
>
source.dispose();
70
>
}
71
>
then<TResult1 = T, TResult2 = never>(resolve?: ((value: T) => TResult1 | Promise<TResult1>) | undefined | null, reject?: ((reason: unknown) => TResult2 | Promise<TResult2>) | undefined | null): Promise<TResult1 | TResult2> {
async.ts ×7
72
>
return promise.then(resolve, reject);
73
>
}
74
>
catch<TResult = never>(reject?: ((reason: unknown) => TResult | Promise<TResult>) | undefined | null): Promise<T | TResult> {
75
return this.then(undefined, reject);
76
}
79
>
}
81
>
}
83
>
/**
84
>
* Returns a promise that resolves with `undefined` as soon as the passed token is cancelled.
85
>
* @see {@link raceCancellationError}
86
>
*/
87
>
export function raceCancellation<T>(promise: Promise<T>, token: CancellationToken): Promise<T | undefined>;
88
>
89
>
/**
90
>
* Returns a promise that resolves with `defaultValue` as soon as the passed token is cancelled.
91
>
* @see {@link raceCancellationError}
92
>
*/
93
>
export function raceCancellation<T>(promise: Promise<T>, token: CancellationToken, defaultValue: T): Promise<T>;
94
>
95
>
export function raceCancellation<T>(promise: Promise<T>, token: CancellationToken, defaultValue?: T): Promise<T | undefined> {
97
>
const ref = token.onCancellationRequested(() => {
99
>
resolve(defaultValue);
101
>
promise.then(resolve, reject).finally(() => ref.dispose());
102
>
});
103
>
}
105
>
/**
106
>
* Returns a promise that rejects with an {@CancellationError} as soon as the passed token is cancelled.
107
>
* @see {@link raceCancellation}
108
>
*/
109
>
export function raceCancellationError<T>(promise: Promise<T>, token: CancellationToken): Promise<T> {
111
>
const ref = token.onCancellationRequested(() => {
112
ref.dispose();
113
reject(new CancellationError());
115
>
promise.then(resolve, reject).finally(() => ref.dispose());
116
>
});
117
>
}
119
>
export function rejectIfNotCanceled(err: unknown): undefined {
120
if (isCancellationError(err)) {
121
return undefined;
122
}
123
return Promise.reject(err) as never;
124
}
126
>
/**
127
>
* Wraps a cancellable promise such that it is no cancellable. Can be used to
128
>
* avoid issues with shared promises that would normally be returned as
129
>
* cancellable to consumers.
130
>
*/
131
>
export function notCancellablePromise<T>(promise: CancelablePromise<T>): Promise<T> {
133
>
promise.then(resolve, reject);
134
>
});
135
>
}
137
>
/**
138
>
* Returns as soon as one of the promises resolves or rejects and cancels remaining promises
139
>
*/
140
>
export function raceCancellablePromises<T>(cancellablePromises: (CancelablePromise<T> | Promise<T>)[]): CancelablePromise<T> {
142
>
const promises = cancellablePromises.map((promise, index) => promise.then(result => { resolvedPromiseIndex = index; return result; }));
143
>
const promise = Promise.race(promises) as CancelablePromise<T>;
144
>
promise.cancel = () => {
145
>
cancellablePromises.forEach((cancellablePromise, index) => {
146
>
if (index !== resolvedPromiseIndex && (cancellablePromise as CancelablePromise<T>).cancel) {
147
>
(cancellablePromise as CancelablePromise<T>).cancel();
148
>
}
149
>
});
150
>
};
151
>
promise.finally(() => {
152
>
promise.cancel();
153
>
});
154
>
return promise;
155
>
}
157
>
export function raceTimeout<T>(promise: Promise<T>, timeout: number, onTimeout?: () => void): Promise<T | undefined> {
159
>
160
>
const timer = setTimeout(() => {
162
>
onTimeout?.();
164
>
165
>
return Promise.race([
166
>
promise.finally(() => clearTimeout(timer)),
167
>
new Promise<T | undefined>(resolve => promiseResolve = resolve)
168
>
]);
169
>
}
171
>
export function asPromise<T>(callback: () => T | Thenable<T>): Promise<T> {
172
return new Promise<T>((resolve, reject) => {
173
const item = callback();
174
if (isThenable<T>(item)) {
175
item.then(resolve, reject);
176
} else {
177
resolve(item);
178
}
179
});
180
}
182
>
/**
183
>
* Creates and returns a new promise, plus its `resolve` and `reject` callbacks.
184
>
*
185
>
* Replace with standardized [`Promise.withResolvers`](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Promise/withResolvers) once it is supported
186
>
*/
187
>
export function promiseWithResolvers<T>(): { promise: Promise<T>; resolve: (value: T | PromiseLike<T>) => void; reject: (err?: any) => void } {
189
>
let reject: (reason?: any) => void;
190
>
const promise = new Promise<T>((res, rej) => {
191
>
resolve = res;
192
>
reject = rej;
193
>
});
194
>
return { promise, resolve: resolve!, reject: reject! };
195
>
}
197
>
export interface ITask<T> {
198
>
(): T;
199
>
}
200
>
201
>
export interface ICancellableTask<T> {
202
>
(token: CancellationToken): T;
203
>
}
204
>
205
>
/**
206
>
* A helper to prevent accumulation of sequential async tasks.
207
>
*
208
>
* Imagine a mail man with the sole task of delivering letters. As soon as
209
>
* a letter submitted for delivery, he drives to the destination, delivers it
210
>
* and returns to his base. Imagine that during the trip, N more letters were submitted.
211
>
* When the mail man returns, he picks those N letters and delivers them all in a
212
>
* single trip. Even though N+1 submissions occurred, only 2 deliveries were made.
213
>
*
214
>
* The throttler implements this via the queue() method, by providing it a task
215
>
* factory. Following the example:
216
>
*
217
>
* const throttler = new Throttler();
218
>
* const letters = [];
219
>
*
220
>
* function deliver() {
221
>
* const lettersToDeliver = letters;
222
>
* letters = [];
223
>
* return makeTheTrip(lettersToDeliver);
224
>
* }
225
>
*
226
>
* function onLetterReceived(l) {
227
>
* letters.push(l);
228
>
* throttler.queue(deliver);
229
>
* }
230
>
*/
231
>
export class Throttler implements IDisposable {
232
>
233
>
private activePromise: Promise<any> | null;
234
>
private queuedPromise: Promise<any> | null;
235
>
private queuedPromiseFactory: ICancellableTask<Promise<any>> | null;
236
>
private cancellationTokenSource: CancellationTokenSource;
237
>
238
>
constructor() {
240
>
this.queuedPromise = null;
241
>
this.queuedPromiseFactory = null;
242
>
243
>
this.cancellationTokenSource = new CancellationTokenSource();
244
>
}
246
>
queue<T>(promiseFactory: ICancellableTask<Promise<T>>): Promise<T> {
249
>
}
251
>
if (this.activePromise) {
253
>
254
>
if (!this.queuedPromise) {
255
>
const onComplete = () => {
256
>
this.queuedPromise = null;
257
>
258
>
if (this.cancellationTokenSource.token.isCancellationRequested) {
260
>
}
262
>
const result = this.queue(this.queuedPromiseFactory!);
263
>
this.queuedPromiseFactory = null;
264
>
265
>
return result;
267
>
268
>
this.queuedPromise = new Promise(resolve => {
269
>
this.activePromise!.then(onComplete, onComplete).then(resolve);
270
>
});
271
>
}
272
>
273
>
return new Promise((resolve, reject) => {
274
>
this.queuedPromise!.then(resolve, reject);
275
>
});
276
>
}
278
>
this.activePromise = promiseFactory(this.cancellationTokenSource.token);
279
>
280
>
return new Promise((resolve, reject) => {
281
>
this.activePromise!.then((result: T) => {
282
>
this.activePromise = null;
283
>
resolve(result);
284
>
}, (err: unknown) => {
285
this.activePromise = null;
286
reject(err);
288
>
});
291
>
dispose(): void {
293
>
}
295
>
296
>
export class Sequencer {
298
>
private current: Promise<unknown> = Promise.resolve(null);
300
>
queue<T>(promiseTask: ITask<Promise<T>>): Promise<T> {
301
>
return this.current = this.current.then(() => promiseTask(), () => promiseTask());
async.ts ×1
302
>
}
304
>
305
>
/**
306
>
* A {@link Throttler} per key. Calls for the same key coalesce (only the most
307
>
* recently queued task runs after the active one settles); calls for different
308
>
* keys are independent. Idle keys are cleaned up automatically.
309
>
*/
310
>
export class ThrottlerByKey<TKey> implements IDisposable {
312
>
private readonly throttlers = new Map<TKey, { throttler: Throttler; count: number }>();
314
>
queue<T>(key: TKey, task: ITask<Promise<T>>): Promise<T> {
316
>
if (!entry) {
317
>
entry = { throttler: new Throttler(), count: 0 };
318
>
this.throttlers.set(key, entry);
319
>
}
320
>
321
>
entry.count++;
322
>
return entry.throttler.queue(task).finally(() => {
323
>
if (--entry!.count === 0) {
324
>
entry!.throttler.dispose();
325
>
this.throttlers.delete(key);
326
>
}
327
>
});
328
>
}
330
>
dispose(): void {
333
>
}
335
>
}
337
>
338
>
export class SequencerByKey<TKey> {
340
>
private promiseMap = new Map<TKey, Promise<unknown>>();
342
>
queue<T>(key: TKey, promiseTask: ITask<Promise<T>>): Promise<T> {
344
>
const newPromise = runningPromise
345
>
.catch(() => { })
346
>
.then(promiseTask)
347
>
.finally(() => {
348
>
if (this.promiseMap.get(key) === newPromise) {
349
>
this.promiseMap.delete(key);
350
>
}
351
>
});
352
>
this.promiseMap.set(key, newPromise);
353
>
return newPromise;
354
>
}
356
>
peek(key: TKey): Promise<unknown> | undefined {
357
return this.promiseMap.get(key) || undefined;
358
}
360
>
keys(): IterableIterator<TKey> {
361
return this.promiseMap.keys();
362
}
364
>
365
>
interface IScheduledLater extends IDisposable {
366
>
isTriggered(): boolean;
367
>
}
368
>
369
>
const timeoutDeferred = (timeout: number, fn: () => void): IScheduledLater => {
371
>
const handle = setTimeout(() => {
373
>
fn();
375
>
return {
376
>
isTriggered: () => scheduled,
377
>
dispose: () => {
379
>
scheduled = false;
380
>
},
382
>
};
384
>
const microtaskDeferred = (fn: () => void): IScheduledLater => {
386
>
queueMicrotask(() => {
387
>
if (scheduled) {
389
>
fn();
390
>
}
392
>
393
>
return {
394
>
isTriggered: () => scheduled,
395
>
dispose: () => { scheduled = false; },
396
>
};
397
>
};
399
>
/**
400
>
* A helper to delay (debounce) execution of a task that is being requested often.
401
>
*
402
>
* Following the throttler, now imagine the mail man wants to optimize the number of
403
>
* trips proactively. The trip itself can be long, so he decides not to make the trip
404
>
* as soon as a letter is submitted. Instead he waits a while, in case more
405
>
* letters are submitted. After said waiting period, if no letters were submitted, he
406
>
* decides to make the trip. Imagine that N more letters were submitted after the first
407
>
* one, all within a short period of time between each other. Even though N+1
408
>
* submissions occurred, only 1 delivery was made.
409
>
*
410
>
* The delayer offers this behavior via the trigger() method, into which both the task
411
>
* to be executed and the waiting period (delay) must be passed in as arguments. Following
412
>
* the example:
413
>
*
414
>
* const delayer = new Delayer(WAITING_PERIOD);
415
>
* const letters = [];
416
>
*
417
>
* function letterReceived(l) {
418
>
* letters.push(l);
419
>
* delayer.trigger(() => { return makeTheTrip(); });
420
>
* }
421
>
*/
422
>
export class Delayer<T> implements IDisposable {
423
>
424
>
private deferred: IScheduledLater | null;
425
>
private completionPromise: Promise<any> | null;
426
>
private doResolve: ((value?: any | Promise<any>) => void) | null;
427
>
private doReject: ((err: unknown) => void) | null;
428
>
private task: ITask<T | Promise<T>> | null;
429
>
430
>
constructor(public defaultDelay: number | typeof MicrotaskDelay) {
432
>
this.completionPromise = null;
433
>
this.doResolve = null;
434
>
this.doReject = null;
435
>
this.task = null;
436
>
}
438
>
trigger(task: ITask<T | Promise<T>>, delay = this.defaultDelay): Promise<T> {
440
>
this.cancelTimeout();
441
>
442
>
if (!this.completionPromise) {
443
>
this.completionPromise = new Promise((resolve, reject) => {
444
>
this.doResolve = resolve;
445
>
this.doReject = reject;
446
>
}).then(() => {
448
>
this.doResolve = null;
449
>
if (this.task) {
450
>
const task = this.task;
451
>
this.task = null;
452
>
return task();
453
>
}
454
return undefined;
456
>
}
457
>
458
>
const fn = () => {
460
>
this.doResolve?.(null);
461
>
};
463
>
this.deferred = delay === MicrotaskDelay ? microtaskDeferred(fn) : timeoutDeferred(delay, fn);
464
>
465
>
return this.completionPromise;
466
>
}
468
>
isTriggered(): boolean {
470
>
}
472
>
cancel(): void {
474
>
475
>
if (this.completionPromise) {
477
>
this.completionPromise = null;
478
>
}
481
>
private cancelTimeout(): void {
483
>
this.deferred = null;
484
>
}
486
>
dispose(): void {
488
>
}
490
>
491
>
/**
492
>
* A helper to delay execution of a task that is being requested often, while
493
>
* preventing accumulation of consecutive executions, while the task runs.
494
>
*
495
>
* The mail man is clever and waits for a certain amount of time, before going
496
>
* out to deliver letters. While the mail man is going out, more letters arrive
497
>
* and can only be delivered once he is back. Once he is back the mail man will
498
>
* do one more trip to deliver the letters that have accumulated while he was out.
499
>
*/
500
>
export class ThrottledDelayer<T> {
501
>
502
>
private delayer: Delayer<Promise<T>>;
503
>
private throttler: Throttler;
504
>
505
>
constructor(defaultDelay: number) {
507
>
this.throttler = new Throttler();
508
>
}
510
>
trigger(promiseFactory: ICancellableTask<Promise<T>>, delay?: number): Promise<T> {
511
>
return this.delayer.trigger(() => this.throttler.queue(promiseFactory), delay) as unknown as Promise<T>;
async.ts ×1
512
>
}
514
>
isTriggered(): boolean {
515
return this.delayer.isTriggered();
516
}
518
>
cancel(): void {
520
>
}
522
>
dispose(): void {
524
>
this.throttler.dispose();
525
>
}
527
>
528
>
/**
529
>
* A barrier that is initially closed and then becomes opened permanently.
530
>
*/
531
>
export class Barrier {
532
>
private _isOpen: boolean;
533
>
private _promise: Promise<boolean>;
534
>
private _completePromise!: (v: boolean) => void;
535
>
536
>
constructor() {
538
>
this._promise = new Promise<boolean>((c, e) => {
539
>
this._completePromise = c;
540
>
});
541
>
}
543
>
isOpen(): boolean {
544
return this._isOpen;
545
}
547
>
open(): void {
549
>
this._completePromise(true);
550
>
}
552
>
wait(): Promise<boolean> {
554
>
}
556
>
557
>
/**
558
>
* A barrier that is initially closed and then becomes opened permanently after a certain period of
559
>
* time or when open is called explicitly
560
>
*/
561
>
export class AutoOpenBarrier extends Barrier {
562
>
563
>
private readonly _timeout: Timeout;
564
>
565
>
constructor(autoOpenTimeMs: number) {
566
super();
567
this._timeout = setTimeout(() => this.open(), autoOpenTimeMs);
568
}
570
>
override open(): void {
571
clearTimeout(this._timeout);
572
super.open();
573
}
575
>
576
>
export function timeout(millis: number): CancelablePromise<void>;
577
>
export function timeout(millis: number, token: CancellationToken): Promise<void>;
578
>
export function timeout(millis: number, token?: CancellationToken): CancelablePromise<void> | Promise<void> {
581
>
}
583
>
return new Promise((resolve, reject) => {
584
>
const handle = setTimeout(() => {
586
>
resolve();
588
>
const disposable = token.onCancellationRequested(() => {
590
>
disposable.dispose();
591
>
reject(new CancellationError());
593
>
});
594
>
}
596
>
/**
597
>
* Creates a timeout that can be disposed using its returned value.
598
>
* @param handler The timeout handler.
599
>
* @param timeout An optional timeout in milliseconds.
600
>
* @param store An optional {@link DisposableStore} that will have the timeout disposable managed automatically.
601
>
*
602
>
* @example
603
>
* const store = new DisposableStore;
604
>
* // Call the timeout after 1000ms at which point it will be automatically
605
>
* // evicted from the store.
606
>
* const timeoutDisposable = disposableTimeout(() => {}, 1000, store);
607
>
*
608
>
* if (foo) {
609
>
* // Cancel the timeout and evict it from store.
610
>
* timeoutDisposable.dispose();
611
>
* }
612
>
*/
613
>
export function disposableTimeout(handler: () => void, timeout = 0, store?: DisposableStore): IDisposable {
616
>
if (store) {
618
>
}
620
>
const disposable = toDisposable(() => {
621
>
clearTimeout(timer);
622
>
store?.delete(disposable);
623
>
});
624
>
store?.add(disposable);
625
>
return disposable;
626
>
}
628
>
/**
629
>
* The largest delay (in milliseconds) a single `setTimeout` can represent.
630
>
* Larger values overflow its internal 32-bit signed integer and fire (almost)
631
>
* immediately instead of waiting.
632
>
*/
633
>
export const MAX_TIMEOUT_DELAY = 2 ** 31 - 1; // ~24.8 days
634
>
635
>
/**
636
>
* Like {@link disposableTimeout}, but supports delays larger than
637
>
* {@link MAX_TIMEOUT_DELAY} (~24.8 days), which a single `setTimeout` cannot
638
>
* represent. The wait is split into chunks and re-armed until the target time is
639
>
* reached, so the handler fires at approximately `Date.now() + timeout`.
640
>
*
641
>
* Note: like `setTimeout`, firing is best-effort and may drift across system
642
>
* sleep or wall-clock changes; do not rely on it for precise scheduling.
643
>
*
644
>
* @param handler The timeout handler.
645
>
* @param timeout The timeout in milliseconds. May exceed {@link MAX_TIMEOUT_DELAY}.
646
>
* @param store An optional {@link DisposableStore} that will have the timeout disposable managed automatically.
647
>
*/
648
>
export function disposableLongTimeout(handler: () => void, timeout: number, store?: DisposableStore): IDisposable {
650
>
let timer: Timeout;
651
>
652
>
const arm = () => {
654
>
if (remaining <= 0) {
656
>
if (store) {
658
>
}
660
>
}
662
>
};
664
>
const disposable = toDisposable(() => {
665
>
clearTimeout(timer);
666
>
store?.delete(disposable);
667
>
});
668
>
669
>
timer = setTimeout(arm, Math.min(Math.max(0, timeout), MAX_TIMEOUT_DELAY));
670
>
store?.add(disposable);
671
>
return disposable;
672
>
}
674
>
/**
675
>
* Runs the provided list of promise factories in sequential order. The returned
676
>
* promise will complete to an array of results from each promise.
677
>
*/
678
>
679
>
export function sequence<T>(promiseFactories: ITask<Promise<T>>[]): Promise<T[]> {
681
>
let index = 0;
682
>
const len = promiseFactories.length;
683
>
684
>
function next(): Promise<T> | null {
685
>
return index < len ? promiseFactories[index++]() : null;
686
>
}
687
>
688
>
function thenHandler(result: unknown): Promise<any> {
689
>
if (result !== undefined && result !== null) {
690
>
results.push(result as T);
691
>
}
692
>
693
>
const n = next();
694
>
if (n) {
695
>
return n.then(thenHandler);
696
>
}
697
>
698
>
return Promise.resolve(results);
699
>
}
700
>
701
>
return Promise.resolve(null).then(thenHandler);
702
>
}
704
>
export function first<T>(promiseFactories: ITask<Promise<T>>[], shouldStop: (t: T) => boolean = t => !!t, defaultValue: T | null = null): Promise<T | null> {
705
let index = 0;
706
const len = promiseFactories.length;
707
708
const loop: () => Promise<T | null> = () => {
709
if (index >= len) {
710
return Promise.resolve(defaultValue);
711
}
712
713
const factory = promiseFactories[index++];
714
const promise = Promise.resolve(factory());
715
716
return promise.then(result => {
717
if (shouldStop(result)) {
718
return Promise.resolve(result);
719
}
720
721
return loop();
722
});
723
};
724
725
return loop();
726
}
728
>
/**
729
>
* Returns the result of the first promise that matches the "shouldStop",
730
>
* running all promises in parallel. Supports cancelable promises.
731
>
*/
732
>
export function firstParallel<T>(promiseList: Promise<T>[], shouldStop?: (t: T) => boolean, defaultValue?: T | null): Promise<T | null>;
733
>
export function firstParallel<T, R extends T>(promiseList: Promise<T>[], shouldStop: (t: T) => t is R, defaultValue?: R | null): Promise<R | null>;
734
>
export function firstParallel<T>(promiseList: Promise<T>[], shouldStop: (t: T) => boolean = t => !!t, defaultValue: T | null = null) {
737
>
}
739
>
let todo = promiseList.length;
740
>
const finish = () => {
742
>
for (const promise of promiseList) {
743
>
(promise as Partial<CancelablePromise<T>>).cancel?.();
744
>
}
745
>
};
747
>
return new Promise<T | null>((resolve, reject) => {
748
>
for (const promise of promiseList) {
749
>
promise.then(result => {
752
>
resolve(result);
755
>
}
757
>
.catch(err => {
760
>
reject(err);
761
>
}
763
>
}
764
>
});
765
>
}
767
>
interface ILimitedTaskFactory<T> {
768
>
factory: ITask<Promise<T>>;
769
>
c: (value: T | Promise<T>) => void;
770
>
e: (error?: unknown) => void;
771
>
}
772
>
773
>
export interface ILimiter<T> {
774
>
775
>
readonly size: number;
776
>
777
>
queue(factory: ITask<Promise<T>>): Promise<T>;
778
>
779
>
clear(): void;
780
>
}
781
>
782
>
/**
783
>
* A helper to queue N promises and run them all with a max degree of parallelism. The helper
784
>
* ensures that at any time no more than M promises are running at the same time.
785
>
*/
786
>
export class Limiter<T> implements ILimiter<T> {
787
>
788
>
private _size = 0;
789
>
private _isDisposed = false;
790
>
private runningPromises: number;
791
>
private readonly maxDegreeOfParalellism: number;
792
>
private readonly outstandingPromises: ILimitedTaskFactory<T>[];
793
>
private readonly _onDrained: Emitter<void>;
794
>
795
>
constructor(maxDegreeOfParalellism: number) {
797
>
this.outstandingPromises = [];
798
>
this.runningPromises = 0;
799
>
this._onDrained = new Emitter<void>();
800
>
}
802
>
/**
803
>
*
804
>
* @returns A promise that resolved when all work is done (onDrained) or when
805
>
* there is nothing to do
806
>
*/
807
>
whenIdle(): Promise<void> {
809
>
? Event.toPromise(this.onDrained)
810
: Promise.resolve();
813
>
get onDrained(): Event<void> {
815
>
}
817
>
get size(): number {
819
>
}
821
>
queue(factory: ITask<Promise<T>>): Promise<T> {
823
throw new Error('Object has been disposed');
824
}
826
>
827
>
return new Promise<T>((c, e) => {
828
>
this.outstandingPromises.push({ factory, c, e });
829
>
this.consume();
830
>
});
831
>
}
833
>
private consume(): void {
834
>
while (this.outstandingPromises.length && this.runningPromises < this.maxDegreeOfParalellism) {
async.ts ×3
835
>
const iLimitedTask = this.outstandingPromises.shift()!;
836
>
this.runningPromises++;
837
>
838
>
const promise = iLimitedTask.factory();
839
>
promise.then(iLimitedTask.c, iLimitedTask.e);
840
>
promise.then(() => this.consumed(), () => this.consumed());
841
>
}
842
>
}
844
>
private consumed(): void {
847
>
}
849
>
if (--this._size === 0) {
850
>
this._onDrained.fire();
851
>
}
852
>
853
>
if (this.outstandingPromises.length > 0) {
855
>
}
858
>
clear(): void {
860
throw new Error('Object has been disposed');
861
}
863
>
this._size = this.runningPromises;
864
>
}
866
>
dispose(): void {
868
>
this.outstandingPromises.length = 0; // stop further processing
869
>
this._size = 0;
870
>
this._onDrained.dispose();
871
>
}
873
>
874
>
/**
875
>
* A queue is handles one promise at a time and guarantees that at any time only one promise is executing.
876
>
*/
877
>
export class Queue<T> extends Limiter<T> {
878
>
879
>
constructor() {
881
>
}
883
>
884
>
/**
885
>
* Same as `Queue`, ensures that only 1 task is executed at the same time. The difference to `Queue` is that
886
>
* there is only 1 task about to be scheduled next. As such, calling `queue` while a task is executing will
887
>
* replace the currently queued task until it executes.
888
>
*
889
>
* As such, the returned promise may not be from the factory that is passed in but from the next factory that
890
>
* is running after having called `queue`.
891
>
*/
892
>
export class LimitedQueue {
894
>
private readonly sequentializer = new TaskSequentializer();
895
>
896
>
private tasks = 0;
898
>
queue(factory: ITask<Promise<void>>): Promise<void> {
900
>
return this.sequentializer.run(this.tasks++, factory());
901
>
}
902
>
903
>
return this.sequentializer.queue(() => {
904
>
return this.sequentializer.run(this.tasks++, factory());
905
>
});
906
>
}
908
>
909
>
/**
910
>
* A helper to organize queues per resource. The ResourceQueue makes sure to manage queues per resource
911
>
* by disposing them once the queue is empty.
912
>
*/
913
>
export class ResourceQueue implements IDisposable {
915
>
private readonly queues = new Map<string, Queue<void>>();
916
>
917
>
private readonly drainers = new Set<DeferredPromise<void>>();
918
>
919
>
private drainListeners: DisposableMap<number> | undefined = undefined;
920
>
private drainListenerCount = 0;
922
>
async whenDrained(): Promise<void> {
924
>
return;
925
>
}
926
>
927
>
const promise = new DeferredPromise<void>();
928
>
this.drainers.add(promise);
929
>
930
>
return promise.p;
931
>
}
933
>
private isDrained(): boolean {
936
>
return false;
937
>
}
938
>
}
940
>
return true;
941
>
}
943
>
queueSize(resource: URI, extUri: IExtUri = defaultExtUri): number {
944
const key = extUri.getComparisonKey(resource);
945
946
return this.queues.get(key)?.size ?? 0;
947
}
949
>
queueFor(resource: URI, factory: ITask<Promise<void>>, extUri: IExtUri = defaultExtUri): Promise<void> {
951
>
952
>
let queue = this.queues.get(key);
953
>
if (!queue) {
954
>
queue = new Queue<void>();
955
>
const drainListenerId = this.drainListenerCount++;
956
>
const drainListener = Event.once(queue.onDrained)(() => {
957
>
queue?.dispose();
958
>
this.queues.delete(key);
959
>
this.onDidQueueDrain();
960
>
961
>
this.drainListeners?.deleteAndDispose(drainListenerId);
962
>
963
>
if (this.drainListeners?.size === 0) {
964
>
this.drainListeners.dispose();
965
>
this.drainListeners = undefined;
966
>
}
967
>
});
968
>
969
>
if (!this.drainListeners) {
970
>
this.drainListeners = new DisposableMap();
971
>
}
972
>
this.drainListeners.set(drainListenerId, drainListener);
973
>
974
>
this.queues.set(key, queue);
975
>
}
976
>
977
>
return queue.queue(factory);
978
>
}
980
>
private onDidQueueDrain(): void {
983
>
}
985
>
this.releaseDrainers();
986
>
}
988
>
private releaseDrainers(): void {
991
>
}
993
>
this.drainers.clear();
994
>
}
996
>
dispose(): void {
999
>
}
1001
>
this.queues.clear();
1002
>
1003
>
// Even though we might still have pending
1004
>
// tasks queued, after the queues have been
1005
>
// disposed, we can no longer track them, so
1006
>
// we release drainers to prevent hanging
1007
>
// promises when the resource queue is being
1008
>
// disposed.
1009
>
this.releaseDrainers();
1010
>
1011
>
this.drainListeners?.dispose();
1012
>
}
1014
>
1015
>
export type Task<T = void> = () => (Promise<T> | T);
1016
>
1017
>
/**
1018
>
* Wrap a type in an optional promise. This can be useful to avoid the runtime
1019
>
* overhead of creating a promise.
1020
>
*/
1021
>
export type MaybePromise<T> = Promise<T> | T;
1022
>
1023
>
/**
1024
>
* Processes tasks in the order they were scheduled.
1025
>
*/
1026
>
export class TaskQueue {
1027
private _runningTask: Task<any> | undefined = undefined;
1028
private _pendingTasks: { task: Task<any>; deferred: DeferredPromise<any>; setUndefinedWhenCleared: boolean }[] = [];
1030
>
/**
1031
>
* Waits for the current and pending tasks to finish, then runs and awaits the given task.
1032
>
* If the task is skipped because of clearPending, the promise is rejected with a CancellationError.
1033
>
*/
1034
>
public schedule<T>(task: Task<T>): Promise<T> {
1035
const deferred = new DeferredPromise<T>();
1036
this._pendingTasks.push({ task, deferred, setUndefinedWhenCleared: false });
1037
this._runIfNotRunning();
1038
return deferred.p;
1039
}
1041
>
/**
1042
>
* Waits for the current and pending tasks to finish, then runs and awaits the given task.
1043
>
* If the task is skipped because of clearPending, the promise is resolved with undefined.
1044
>
*/
1045
>
public scheduleSkipIfCleared<T>(task: Task<T>): Promise<T | undefined> {
1046
const deferred = new DeferredPromise<T>();
1047
this._pendingTasks.push({ task, deferred, setUndefinedWhenCleared: true });
1048
this._runIfNotRunning();
1049
return deferred.p;
1050
}
1052
>
private _runIfNotRunning(): void {
1053
if (this._runningTask === undefined) {
1054
this._processQueue();
1055
}
1056
}
1058
>
private async _processQueue(): Promise<void> {
1059
if (this._pendingTasks.length === 0) {
1060
return;
1061
}
1062
1063
const next = this._pendingTasks.shift();
1064
if (!next) {
1065
return;
1066
}
1067
1068
if (this._runningTask) {
1069
throw new BugIndicatingError();
1070
}
1071
1072
this._runningTask = next.task;
1073
1074
try {
1075
const result = await next.task();
1076
next.deferred.complete(result);
1077
} catch (e) {
1078
next.deferred.error(e);
1079
} finally {
1080
this._runningTask = undefined;
1081
this._processQueue();
1082
}
1083
}
1085
>
/**
1086
>
* Clears all pending tasks. Does not cancel the currently running task.
1087
>
*/
1088
>
public clearPending(): void {
1089
const tasks = this._pendingTasks;
1090
this._pendingTasks = [];
1091
for (const task of tasks) {
1092
if (task.setUndefinedWhenCleared) {
1093
task.deferred.complete(undefined);
1094
} else {
1095
task.deferred.error(new CancellationError());
1096
}
1097
}
1098
}
1100
>
1101
>
export class TimeoutTimer implements IDisposable {
1102
>
private _token: Timeout | undefined;
1103
>
private _isDisposed = false;
1104
>
1105
>
constructor();
1106
>
constructor(runner: () => void, timeout: number);
1107
>
constructor(runner?: () => void, timeout?: number) {
1109
>
1110
>
if (typeof runner === 'function' && typeof timeout === 'number') {
1112
>
}
1115
>
dispose(): void {
1117
>
this._isDisposed = true;
1118
>
}
1120
>
cancel(): void {
1123
>
this._token = undefined;
1124
>
}
1127
>
cancelAndSet(runner: () => void, timeout: number): void {
1128
if (this._isDisposed) {
1129
throw new BugIndicatingError(`Calling 'cancelAndSet' on a disposed TimeoutTimer`);
1130
}
1131
1132
this.cancel();
1133
this._token = setTimeout(() => {
1134
this._token = undefined;
1135
runner();
1136
}, timeout);
1137
}
1139
>
setIfNotSet(runner: () => void, timeout: number): void {
1141
throw new BugIndicatingError(`Calling 'setIfNotSet' on a disposed TimeoutTimer`);
1142
}
1144
>
if (this._token !== undefined) {
1145
// timer is already set
1146
return;
1147
}
1149
>
this._token = undefined;
1150
>
runner();
1151
>
}, timeout);
1152
>
}
1154
>
1155
>
export class IntervalTimer implements IDisposable {
1157
>
private disposable: IDisposable | undefined = undefined;
1158
>
private isDisposed = false;
1160
>
cancel(): void {
1162
>
this.disposable = undefined;
1163
>
}
1165
>
cancelAndSet(runner: () => void, interval: number, context = globalThis): void {
1167
throw new BugIndicatingError(`Calling 'cancelAndSet' on a disposed IntervalTimer`);
1168
}
1170
>
this.cancel();
1171
>
const handle = context.setInterval(() => {
1174
>
1175
>
this.disposable = toDisposable(() => {
1176
>
context.clearInterval(handle);
1177
>
this.disposable = undefined;
1178
>
});
1179
>
}
1181
>
dispose(): void {
1183
>
this.isDisposed = true;
1184
>
}
1186
>
1187
>
export class RunOnceScheduler<Runner extends (...args: any[]) => any = () => any> implements IDisposable {
1188
>
1189
>
protected runner: Runner | null;
1190
>
1191
>
private timeoutToken: Timeout | undefined;
1192
>
private timeout: number;
1193
>
private timeoutHandler: () => void;
1194
>
1195
>
constructor(runner: Runner, delay: number) {
1197
>
this.runner = runner;
1198
>
this.timeout = delay;
1199
>
this.timeoutHandler = this.onTimeout.bind(this);
1200
>
}
1202
>
/**
1203
>
* Dispose RunOnceScheduler
1204
>
*/
1205
>
dispose(): void {
1207
>
this.runner = null;
1208
>
}
1210
>
/**
1211
>
* Cancel current scheduled runner (if any).
1212
>
*/
1213
>
cancel(): void {
1216
>
this.timeoutToken = undefined;
1217
>
}
1220
>
/**
1221
>
* Cancel previous runner (if any) & schedule a new runner.
1222
>
*/
1223
>
schedule(delay = this.timeout): void {
1225
>
this.timeoutToken = setTimeout(this.timeoutHandler, delay);
1226
>
}
1228
>
get delay(): number {
1229
return this.timeout;
1230
}
1232
>
set delay(value: number) {
1233
this.timeout = value;
1234
}
1236
>
/**
1237
>
* Returns true if scheduled.
1238
>
*/
1239
>
isScheduled(): boolean {
1241
>
}
1243
>
flush(): void {
1244
if (this.isScheduled()) {
1245
this.cancel();
1246
this.doRun();
1247
}
1248
}
1250
>
private onTimeout() {
1252
>
if (this.runner) {
1253
>
this.doRun();
1254
>
}
1255
>
}
1257
>
protected doRun(): void {
1259
>
}
1261
>
1262
>
/**
1263
>
* Same as `RunOnceScheduler`, but doesn't count the time spent in sleep mode.
1264
>
* > **NOTE**: Only offers 1s resolution.
1265
>
*
1266
>
* When calling `setTimeout` with 3hrs, and putting the computer immediately to sleep
1267
>
* for 8hrs, `setTimeout` will fire **as soon as the computer wakes from sleep**. But
1268
>
* this scheduler will execute 3hrs **after waking the computer from sleep**.
1269
>
*/
1270
>
export class ProcessTimeRunOnceScheduler {
1271
>
1272
>
private runner: (() => void) | null;
1273
>
private timeout: number;
1274
>
1275
>
private counter: number;
1276
>
private intervalToken: Timeout | undefined;
1277
>
private intervalHandler: () => void;
1278
>
1279
>
constructor(runner: () => void, delay: number) {
1280
if (delay % 1000 !== 0) {
1281
console.warn(`ProcessTimeRunOnceScheduler resolution is 1s, ${delay}ms is not a multiple of 1000ms.`);
1282
}
1283
this.runner = runner;
1284
this.timeout = delay;
1285
this.counter = 0;
1286
this.intervalToken = undefined;
1287
this.intervalHandler = this.onInterval.bind(this);
1288
}
1290
>
dispose(): void {
1291
this.cancel();
1292
this.runner = null;
1293
}
1295
>
cancel(): void {
1296
if (this.isScheduled()) {
1297
clearInterval(this.intervalToken);
1298
this.intervalToken = undefined;
1299
}
1300
}
1302
>
/**
1303
>
* Cancel previous runner (if any) & schedule a new runner.
1304
>
*/
1305
>
schedule(delay = this.timeout): void {
1306
if (delay % 1000 !== 0) {
1307
console.warn(`ProcessTimeRunOnceScheduler resolution is 1s, ${delay}ms is not a multiple of 1000ms.`);
1308
}
1309
this.cancel();
1310
this.counter = Math.ceil(delay / 1000);
1311
this.intervalToken = setInterval(this.intervalHandler, 1000);
1312
}
1314
>
/**
1315
>
* Returns true if scheduled.
1316
>
*/
1317
>
isScheduled(): boolean {
1318
return this.intervalToken !== undefined;
1319
}
1321
>
private onInterval() {
1322
this.counter--;
1323
if (this.counter > 0) {
1324
// still need to wait
1325
return;
1326
}
1327
1328
// time elapsed
1329
clearInterval(this.intervalToken);
1330
this.intervalToken = undefined;
1331
this.runner?.();
1332
}
1334
>
1335
>
export class RunOnceWorker<T> extends RunOnceScheduler<(units: T[]) => void> {
1336
>
1337
>
private units: T[] = [];
1338
>
1339
>
constructor(runner: (units: T[]) => void, timeout: number) {
1340
super(runner, timeout);
1341
}
1343
>
work(unit: T): void {
1344
this.units.push(unit);
1345
1346
if (!this.isScheduled()) {
1347
this.schedule();
1348
}
1349
}
1351
>
protected override doRun(): void {
1352
const units = this.units;
1353
this.units = [];
1354
1355
this.runner?.(units);
1356
}
1358
>
override dispose(): void {
1359
this.units = [];
1360
1361
super.dispose();
1362
}
1364
>
1365
>
export interface IThrottledWorkerOptions {
1366
>
1367
>
/**
1368
>
* maximum of units the worker will pass onto handler at once
1369
>
*/
1370
>
maxWorkChunkSize: number;
1371
>
1372
>
/**
1373
>
* maximum of units the worker will keep in memory for processing
1374
>
*/
1375
>
maxBufferedWork: number | undefined;
1376
>
1377
>
/**
1378
>
* delay before processing the next round of chunks when chunk size exceeds limits
1379
>
*/
1380
>
throttleDelay: number;
1381
>
1382
>
/**
1383
>
* When enabled will guarantee that two distinct calls to `work()` are not executed
1384
>
* without throttle delay between them.
1385
>
* Otherwise if the worker isn't currently throttling it will execute work immediately.
1386
>
*/
1387
>
waitThrottleDelayBetweenWorkUnits?: boolean;
1388
>
}
1389
>
1390
>
/**
1391
>
* The `ThrottledWorker` will accept units of work `T`
1392
>
* to handle. The contract is:
1393
>
* * there is a maximum of units the worker can handle at once (via `maxWorkChunkSize`)
1394
>
* * there is a maximum of units the worker will keep in memory for processing (via `maxBufferedWork`)
1395
>
* * after having handled `maxWorkChunkSize` units, the worker needs to rest (via `throttleDelay`)
1396
>
*/
1397
>
export class ThrottledWorker<T> extends Disposable {
1398
>
1399
>
private readonly pendingWork: T[] = [];
1400
>
1401
>
private readonly throttler = this._register(new MutableDisposable<RunOnceScheduler>());
1402
>
private disposed = false;
1403
>
private lastExecutionTime = 0;
1404
>
1405
>
constructor(
1407
>
private readonly handler: (units: T[]) => void
1408
>
) {
1409
>
super();
1410
>
}
1412
>
/**
1413
>
* The number of work units that are pending to be processed.
1414
>
*/
1415
>
get pending(): number { return this.pendingWork.length; }
1416
>
1417
>
/**
1418
>
* Add units to be worked on. Use `pending` to figure out
1419
>
* how many units are not yet processed after this method
1420
>
* was called.
1421
>
*
1422
>
* @returns whether the work was accepted or not. If the
1423
>
* worker is disposed, it will not accept any more work.
1424
>
* If the number of pending units would become larger
1425
>
* than `maxPendingWork`, more work will also not be accepted.
1426
>
*/
1427
>
work(units: readonly T[]): boolean {
1430
>
}
1432
>
// Check for reaching maximum of pending work
1433
>
if (typeof this.options.maxBufferedWork === 'number') {
1435
>
// Throttled: simple check if pending + units exceeds max pending
1436
>
if (this.throttler.value) {
1438
>
return false; // work not accepted: too much pending work
1439
>
}
1440
>
}
1442
>
// Unthrottled: same as throttled, but account for max chunk getting
1443
>
// worked on directly without being pending
1444
>
else {
1445
>
if (this.pending + units.length - this.options.maxWorkChunkSize > this.options.maxBufferedWork) {
1447
>
}
1449
>
}
1451
>
// Add to pending units first
1452
>
for (const unit of units) {
1453
>
this.pendingWork.push(unit);
1454
>
}
1455
>
1456
>
const timeSinceLastExecution = Date.now() - this.lastExecutionTime;
1457
>
1458
>
if (!this.throttler.value && (!this.options.waitThrottleDelayBetweenWorkUnits || timeSinceLastExecution >= this.options.throttleDelay)) {
async.ts ×5
1460
>
// enforced to throttle between `work()` calls.
1461
>
this.doWork();
1462
>
} else if (!this.throttler.value && this.options.waitThrottleDelayBetweenWorkUnits) {
1463
// Otherwise, schedule the throttler to work.
1464
this.scheduleThrottler(Math.max(this.options.throttleDelay - timeSinceLastExecution, 0));
1466
>
// Otherwise, our work will be picked up by the running throttler
1467
>
}
1469
>
return true; // work accepted
1472
>
private doWork(): void {
1474
>
1475
>
// Extract chunk to handle and handle it
1476
>
this.handler(this.pendingWork.splice(0, this.options.maxWorkChunkSize));
1477
>
1478
>
// If we have remaining work, schedule it after a delay
1479
>
if (this.pendingWork.length > 0) {
1480
>
this.scheduleThrottler();
1481
>
}
1482
>
}
1484
>
private scheduleThrottler(delay = this.options.throttleDelay): void {
1487
>
1488
>
this.doWork();
1490
>
this.throttler.value.schedule();
1491
>
}
1493
>
override dispose(): void {
1495
>
1496
>
this.pendingWork.length = 0;
1497
>
this.disposed = true;
1498
>
}
1500
>
1501
>
//#region -- run on idle tricks ------------
1502
>
1503
>
export interface IdleDeadline {
1504
>
readonly didTimeout: boolean;
1505
>
timeRemaining(): number;
1506
>
}
1507
>
1508
>
type IdleApi = Pick<typeof globalThis, 'requestIdleCallback' | 'cancelIdleCallback'>;
1509
>
1510
>
1511
>
/**
1512
>
* Execute the callback the next time the browser is idle, returning an
1513
>
* {@link IDisposable} that will cancel the callback when disposed. This wraps
1514
>
* [requestIdleCallback] so it will fallback to [setTimeout] if the environment
1515
>
* doesn't support it.
1516
>
*
1517
>
* @param callback The callback to run when idle, this includes an
1518
>
* [IdleDeadline] that provides the time alloted for the idle callback by the
1519
>
* browser. Not respecting this deadline will result in a degraded user
1520
>
* experience.
1521
>
* @param timeout A timeout at which point to queue no longer wait for an idle
1522
>
* callback but queue it on the regular event loop (like setTimeout). Typically
1523
>
* this should not be used.
1524
>
*
1525
>
* [IdleDeadline]: https://developer.mozilla.org/en-US/docs/Web/API/IdleDeadline
1526
>
* [requestIdleCallback]: https://developer.mozilla.org/en-US/docs/Web/API/Window/requestIdleCallback
1527
>
* [setTimeout]: https://developer.mozilla.org/en-US/docs/Web/API/Window/setTimeout
1528
>
*
1529
>
* **Note** that there is `dom.ts#runWhenWindowIdle` which is better suited when running inside a browser
1530
>
* context
1531
>
*/
1532
>
export let runWhenGlobalIdle: (callback: (idle: IdleDeadline) => void, timeout?: number) => IDisposable;
1533
>
1534
>
export let _runWhenIdle: (targetWindow: IdleApi, callback: (idle: IdleDeadline) => void, timeout?: number) => IDisposable;
1535
>
1536
>
(function () {
1537
>
const safeGlobal: any = globalThis;
1538
>
if (typeof safeGlobal.requestIdleCallback !== 'function' || typeof safeGlobal.cancelIdleCallback !== 'function') {
1539
>
_runWhenIdle = (_targetWindow, runner, timeout?) => {
1543
>
}
1545
>
const deadline: IdleDeadline = {
1546
>
didTimeout: true,
1547
>
timeRemaining() {
1548
return Math.max(0, end - Date.now());
1549
}
1551
>
runner(Object.freeze(deadline));
1553
>
let disposed = false;
1554
>
return {
1555
>
dispose() {
1556
>
if (disposed) {
1557
return;
1558
}
1560
>
}
1561
>
};
1562
>
};
1564
_runWhenIdle = (targetWindow: typeof safeGlobal, runner, timeout?) => {
1565
const handle: number = targetWindow.requestIdleCallback(runner, typeof timeout === 'number' ? { timeout } : undefined);
1566
let disposed = false;
1567
return {
1568
dispose() {
1569
if (disposed) {
1570
return;
1571
}
1572
disposed = true;
1573
targetWindow.cancelIdleCallback(handle);
1574
}
1575
};
1576
};
1577
}
1578
>
runWhenGlobalIdle = (runner, timeout) => _runWhenIdle(globalThis, runner, timeout);
async.ts ×202
1579
>
})();
1580
>
1581
>
export function installFakeRunWhenIdle(fakeImpl: typeof _runWhenIdle): IDisposable {
1582
const origRunWhenIdle = _runWhenIdle;
1583
const origRunWhenGlobalIdle = runWhenGlobalIdle;
1584
_runWhenIdle = fakeImpl;
1585
runWhenGlobalIdle = (runner, timeout) => fakeImpl(globalThis, runner, timeout);
1586
return toDisposable(() => {
1587
_runWhenIdle = origRunWhenIdle;
1588
runWhenGlobalIdle = origRunWhenGlobalIdle;
1589
});
1590
}
1592
>
export abstract class AbstractIdleValue<T> {
1593
>
1594
>
private readonly _executor: () => void;
1595
>
private readonly _handle: IDisposable;
1596
>
1597
>
private _didRun: boolean = false;
1598
>
private _value?: T;
1599
>
private _error: unknown;
1600
>
1601
>
constructor(targetWindow: IdleApi, executor: () => T) {
1603
>
try {
1604
>
this._value = executor();
1605
>
} catch (err) {
1606
this._error = err;
1608
>
this._didRun = true;
1609
>
}
1610
>
};
1611
>
this._handle = _runWhenIdle(targetWindow, () => this._executor());
1612
>
}
1614
>
dispose(): void {
1615
this._handle.dispose();
1616
}
1618
>
get value(): T {
1620
>
this._handle.dispose();
1621
>
this._executor();
1622
>
}
1623
>
if (this._error) {
1624
throw this._error;
1625
}
1627
>
}
1629
>
get isInitialized(): boolean {
1631
>
}
1633
>
1634
>
/**
1635
>
* An `IdleValue` that always uses the current window (which might be throttled or inactive)
1636
>
*
1637
>
* **Note** that there is `dom.ts#WindowIdleValue` which is better suited when running inside a browser
1638
>
* context
1639
>
*/
1640
>
export class GlobalIdleValue<T> extends AbstractIdleValue<T> {
1641
>
1642
>
constructor(executor: () => T) {
1644
>
}
1646
>
1647
>
//#endregion
1648
>
1649
>
export async function retry<T>(task: ITask<Promise<T>>, delay: number, retries: number): Promise<T> {
async.ts ×2
1650
>
let lastError: Error | undefined;
1651
>
1652
>
for (let i = 0; i < retries; i++) {
1653
>
try {
1654
>
return await task();
1655
>
} catch (error) {
1657
>
1658
>
await timeout(delay);
1659
>
}
1662
>
throw lastError;
1663
>
}
1665
>
//#region Task Sequentializer
1666
>
1667
>
interface IRunningTask {
1668
>
readonly taskId: number;
1669
>
readonly cancel: () => void;
1670
>
readonly promise: Promise<void>;
1671
>
}
1672
>
1673
>
interface IQueuedTask {
1674
>
readonly promise: Promise<void>;
1675
>
readonly promiseResolve: () => void;
1676
>
readonly promiseReject: (error: Error) => void;
1677
>
run: ITask<Promise<void>>;
1678
>
}
1679
>
1680
>
export interface ITaskSequentializerWithRunningTask {
1681
>
readonly running: Promise<void>;
1682
>
}
1683
>
1684
>
export interface ITaskSequentializerWithQueuedTask {
1685
>
readonly queued: IQueuedTask;
1686
>
}
1687
>
1688
>
/**
1689
>
* @deprecated use `LimitedQueue` instead for an easier to use API
1690
>
*/
1691
>
export class TaskSequentializer {
1692
>
1693
>
private _running?: IRunningTask;
1694
>
private _queued?: IQueuedTask;
1695
>
1696
>
isRunning(taskId?: number): this is ITaskSequentializerWithRunningTask {
1699
>
}
1701
>
return !!this._running;
1702
>
}
1704
>
get running(): Promise<void> | undefined {
1706
>
}
1708
>
cancelRunning(): void {
1710
>
}
1712
>
run(taskId: number, promise: Promise<void>, onCancel?: () => void,): Promise<void> {
1714
>
1715
>
promise.then(() => this.doneRunning(taskId), () => this.doneRunning(taskId));
1716
>
1717
>
return promise;
1718
>
}
1720
>
private doneRunning(taskId: number): void {
1722
>
1723
>
// only set running to done if the promise finished that is associated with that taskId
1724
>
this._running = undefined;
1725
>
1726
>
// schedule the queued task now that we are free if we have any
1727
>
this.runQueued();
1728
>
}
1729
>
}
1731
>
private runQueued(): void {
1734
>
this._queued = undefined;
1735
>
1736
>
// Run queued task and complete on the associated promise
1737
>
queued.run().then(queued.promiseResolve, queued.promiseReject);
1738
>
}
1741
>
/**
1742
>
* Note: the promise to schedule as next run MUST itself call `run`.
1743
>
* Otherwise, this sequentializer will report `false` for `isRunning`
1744
>
* even when this task is running. Missing this detail means that
1745
>
* suddenly multiple tasks will run in parallel.
1746
>
*/
1747
>
queue(run: ITask<Promise<void>>): Promise<void> {
1749
>
// this is our first queued task, so we create associated promise with it
1750
>
// so that we can return a promise that completes when the task has
1751
>
// completed.
1752
>
if (!this._queued) {
1753
>
const { promise, resolve: promiseResolve, reject: promiseReject } = promiseWithResolvers<void>();
1754
>
this._queued = {
1755
>
run,
1756
>
promise,
1757
>
promiseResolve,
1758
>
promiseReject
1759
>
};
1760
>
}
1762
>
// we have a previous queued task, just overwrite it
1763
>
else {
1764
>
this._queued.run = run;
1765
>
}
1767
>
return this._queued.promise;
1768
>
}
1770
>
hasQueued(): this is ITaskSequentializerWithQueuedTask {
1772
>
}
1774
>
async join(): Promise<void> {
1776
>
}
1778
>
1779
>
//#endregion
1780
>
1781
>
//#region
1782
>
1783
>
/**
1784
>
* The `IntervalCounter` allows to count the number
1785
>
* of calls to `increment()` over a duration of
1786
>
* `interval`. This utility can be used to conditionally
1787
>
* throttle a frequent task when a certain threshold
1788
>
* is reached.
1789
>
*/
1790
>
export class IntervalCounter {
1791
>
1792
>
private lastIncrementTime = 0;
1793
>
1794
>
private value = 0;
1795
>
1796
>
constructor(private readonly interval: number, private readonly nowFn = () => Date.now()) { }
1797
>
1798
>
increment(): number {
1800
>
1801
>
// We are outside of the range of `interval` and as such
1802
>
// start counting from 0 and remember the time
1803
>
if (now - this.lastIncrementTime > this.interval) {
1804
>
this.lastIncrementTime = now;
1805
>
this.value = 0;
1806
>
}
1807
>
1808
>
this.value++;
1809
>
1810
>
return this.value;
1811
>
}
1813
>
1814
>
//#endregion
1815
>
1816
>
//#region
1817
>
1818
>
export type ValueCallback<T = unknown> = (value: T | Promise<T>) => void;
1819
>
1820
>
const enum DeferredOutcome {
1821
>
Resolved,
1822
>
Rejected
1823
>
}
1824
>
1825
>
/**
1826
>
* Creates a promise whose resolution or rejection can be controlled imperatively.
1827
>
*/
1828
>
export class DeferredPromise<T> {
1829
>
1830
>
public static fromPromise<T>(promise: Promise<T>): DeferredPromise<T> {
1831
const deferred = new DeferredPromise<T>();
1832
deferred.settleWith(promise);
1833
return deferred;
1834
}
1836
>
private completeCallback!: ValueCallback<T>;
1837
>
private errorCallback!: (err: unknown) => void;
1838
>
private outcome?: { outcome: DeferredOutcome.Rejected; value: unknown } | { outcome: DeferredOutcome.Resolved; value: T };
1839
>
1840
>
public get isRejected() {
1842
>
}
1844
>
public get isResolved() {
1846
>
}
1848
>
public get isSettled() {
1850
>
}
1852
>
public get value() {
1853
>
return this.outcome?.outcome === DeferredOutcome.Resolved ? this.outcome?.value : undefined;
async.ts ×1
1854
>
}
1856
>
public readonly p: Promise<T>;
1857
>
1858
>
constructor() {
1860
>
this.completeCallback = c;
1861
>
this.errorCallback = e;
1862
>
});
1863
>
}
1865
>
public complete(value: T) {
1868
>
}
1870
>
return new Promise<void>(resolve => {
1871
>
this.completeCallback(value);
1872
>
this.outcome = { outcome: DeferredOutcome.Resolved, value };
1873
>
resolve();
1874
>
});
1875
>
}
1877
>
public error(err: unknown) {
1879
return Promise.resolve();
1880
}
1882
>
return new Promise<void>(resolve => {
1883
>
this.errorCallback(err);
1884
>
this.outcome = { outcome: DeferredOutcome.Rejected, value: err };
1885
>
resolve();
1886
>
});
1887
>
}
1889
>
public settleWith(promise: Promise<T>): Promise<void> {
1890
return promise.then(
1891
value => this.complete(value),
1892
error => this.error(error)
1893
);
1894
}
1896
>
public cancel() {
1898
>
}
1900
>
1901
>
//#endregion
1902
>
1903
>
//#region Promises
1904
>
1905
>
export namespace Promises {
1906
>
1907
>
/**
1908
>
* A drop-in replacement for `Promise.all` with the only difference
1909
>
* that the method awaits every promise to either fulfill or reject.
1910
>
*
1911
>
* Similar to `Promise.all`, only the first error will be returned
1912
>
* if any.
1913
>
*/
1914
>
export async function settled<T>(promises: Promise<T>[]): Promise<T[]> {
1916
>
1917
>
const result = await Promise.all(promises.map(promise => promise.then(value => value, error => {
1919
>
firstError = error;
1920
>
}
1921
>
1922
>
return undefined; // do not rethrow so that other promises can settle
1924
>
1925
>
if (typeof firstError !== 'undefined') {
1927
>
}
1929
>
return result as unknown as T[]; // cast is needed and protected by the `throw` above
1932
>
/**
1933
>
* A helper to create a new `Promise<T>` with a body that is a promise
1934
>
* itself. By default, an error that raises from the async body will
1935
>
* end up as a unhandled rejection, so this utility properly awaits the
1936
>
* body and rejects the promise as a normal promise does without async
1937
>
* body.
1938
>
*
1939
>
* This method should only be used in rare cases where otherwise `async`
1940
>
* cannot be used (e.g. when callbacks are involved that require this).
1941
>
*/
1942
>
export function withAsyncBody<T, E = Error>(bodyFn: (resolve: (value: T) => unknown, reject: (error: E) => unknown) => Promise<unknown>): Promise<T> {
1944
>
return new Promise<T>(async (resolve, reject) => {
1945
>
try {
1946
>
await bodyFn(resolve, reject);
1947
>
} catch (error) {
1949
>
}
1951
>
}
1953
>
1954
>
export class StatefulPromise<T> {
1955
>
private _value: T | undefined = undefined;
1956
>
get value(): T | undefined { return this._value; }
1957
>
1958
>
private _error: unknown = undefined;
1959
>
get error(): unknown { return this._error; }
1960
>
1961
>
private _isResolved = false;
1962
>
get isResolved() { return this._isResolved; }
1963
>
1964
>
public readonly promise: Promise<T>;
1965
>
1966
>
constructor(promise: Promise<T>) {
1967
this.promise = promise.then(
1968
value => {
1969
this._value = value;
1970
this._isResolved = true;
1971
return value;
1972
},
1973
error => {
1974
this._error = error;
1975
this._isResolved = true;
1976
throw error;
1977
}
1978
);
1979
}
1981
>
/**
1982
>
* Returns the resolved value.
1983
>
* Throws if the promise is not resolved yet.
1984
>
*/
1985
>
public requireValue(): T {
1986
if (!this._isResolved) {
1987
throw new BugIndicatingError('Promise is not resolved yet');
1988
}
1989
if (this._error) {
1990
throw this._error;
1991
}
1992
return this._value!;
1993
}
1995
>
1996
>
export class LazyStatefulPromise<T> {
1997
>
private readonly _promise = new Lazy(() => new StatefulPromise(this._compute()));
1998
>
1999
>
constructor(
2000
private readonly _compute: () => Promise<T>,
2001
) { }
2003
>
/**
2004
>
* Returns the resolved value.
2005
>
* Throws if the promise is not resolved yet.
2006
>
*/
2007
>
public requireValue(): T {
2008
return this._promise.value.requireValue();
2009
}
2011
>
/**
2012
>
* Returns the promise (and triggers a computation of the promise if not yet done so).
2013
>
*/
2014
>
public getPromise(): Promise<T> {
2015
return this._promise.value.promise;
2016
}
2018
>
/**
2019
>
* Reads the current value without triggering a computation of the promise.
2020
>
*/
2021
>
public get currentValue(): T | undefined {
2022
return this._promise.rawValue?.value;
2023
}
2025
>
2026
>
//#endregion
2027
>
2028
>
//#region
2029
>
2030
>
const enum AsyncIterableSourceState {
2031
>
Initial,
2032
>
DoneOK,
2033
>
DoneError,
2034
>
}
2035
>
2036
>
/**
2037
>
* An object that allows to emit async values asynchronously or bring the iterable to an error state using `reject()`.
2038
>
* This emitter is valid only for the duration of the executor (until the promise returned by the executor settles).
2039
>
*/
2040
>
export interface AsyncIterableEmitter<T> {
2041
>
/**
2042
>
* The value will be appended at the end.
2043
>
*
2044
>
* **NOTE** If `reject()` has already been called, this method has no effect.
2045
>
*/
2046
>
emitOne(value: T): void;
2047
>
/**
2048
>
* The values will be appended at the end.
2049
>
*
2050
>
* **NOTE** If `reject()` has already been called, this method has no effect.
2051
>
*/
2052
>
emitMany(values: T[]): void;
2053
>
/**
2054
>
* Writing an error will permanently invalidate this iterable.
2055
>
* The current users will receive an error thrown, as will all future users.
2056
>
*
2057
>
* **NOTE** If `reject()` have already been called, this method has no effect.
2058
>
*/
2059
>
reject(error: Error): void;
2060
>
}
2061
>
2062
>
/**
2063
>
* An executor for the `AsyncIterableObject` that has access to an emitter.
2064
>
*/
2065
>
export interface AsyncIterableExecutor<T> {
2066
>
/**
2067
>
* @param emitter An object that allows to emit async values valid only for the duration of the executor.
2068
>
*/
2069
>
(emitter: AsyncIterableEmitter<T>): unknown | Promise<unknown>;
2070
>
}
2071
>
2072
>
/**
2073
>
* A rich implementation for an `AsyncIterable<T>`.
2074
>
*/
2075
>
export class AsyncIterableObject<T> implements AsyncIterable<T> {
2076
>
2077
>
public static fromArray<T>(items: T[]): AsyncIterableObject<T> {
2078
>
return new AsyncIterableObject<T>((writer) => {
2079
>
writer.emitMany(items);
2080
>
});
2081
>
}
2082
>
2083
>
public static fromPromise<T>(promise: Promise<T[]>): AsyncIterableObject<T> {
2084
return new AsyncIterableObject<T>(async (emitter) => {
2085
emitter.emitMany(await promise);
2086
});
2087
}
2089
>
public static fromPromisesResolveOrder<T>(promises: Promise<T>[]): AsyncIterableObject<T> {
2090
return new AsyncIterableObject<T>(async (emitter) => {
2091
await Promise.all(promises.map(async (p) => emitter.emitOne(await p)));
2092
});
2093
}
2095
>
public static merge<T>(iterables: AsyncIterable<T>[]): AsyncIterableObject<T> {
2096
return new AsyncIterableObject(async (emitter) => {
2097
await Promise.all(iterables.map(async (iterable) => {
2098
for await (const item of iterable) {
2099
emitter.emitOne(item);
2100
}
2101
}));
2102
});
2103
}
2105
>
public static EMPTY = AsyncIterableObject.fromArray<any>([]);
2106
>
2107
>
private _state: AsyncIterableSourceState;
2108
>
private _results: T[];
2109
>
private _error: Error | null;
2110
>
private readonly _onReturn?: () => void | Promise<void>;
2111
>
private readonly _onStateChanged: Emitter<void>;
2112
>
2113
>
constructor(executor: AsyncIterableExecutor<T>, onReturn?: () => void | Promise<void>) {
2114
>
this._state = AsyncIterableSourceState.Initial;
2115
>
this._results = [];
2116
>
this._error = null;
2117
>
this._onReturn = onReturn;
2118
>
this._onStateChanged = new Emitter<void>();
2119
>
2120
>
queueMicrotask(async () => {
2121
>
const writer: AsyncIterableEmitter<T> = {
2122
>
emitOne: (item) => this.emitOne(item),
2123
>
emitMany: (items) => this.emitMany(items),
2124
>
reject: (error) => this.reject(error)
2125
>
};
2126
>
try {
2127
>
await Promise.resolve(executor(writer));
2128
>
this.resolve();
2129
>
} catch (err) {
2130
this.reject(err);
2132
>
// The executor has settled; emitting afterwards must be a no-op per the
2133
>
// documented "no effect after resolve()/reject()" contract (see emitOne).
2134
>
writer.emitOne = () => { };
2135
>
writer.emitMany = () => { };
2136
>
writer.reject = () => { };
2137
>
}
2138
>
});
2139
>
}
2140
>
2141
>
[Symbol.asyncIterator](): AsyncIterator<T, undefined, undefined> {
2143
>
return {
2144
>
next: async () => {
2145
>
do {
2146
>
if (this._state === AsyncIterableSourceState.DoneError) {
2147
throw this._error;
2148
}
2151
>
}
2154
>
}
2156
>
} while (true);
2157
>
},
2158
>
return: async () => {
2160
>
return { done: true, value: undefined };
2161
>
}
2163
>
}
2165
>
public static map<T, R>(iterable: AsyncIterable<T>, mapFn: (item: T) => R): AsyncIterableObject<R> {
2166
return new AsyncIterableObject<R>(async (emitter) => {
2167
for await (const item of iterable) {
2168
emitter.emitOne(mapFn(item));
2169
}
2170
});
2171
}
2173
>
public map<R>(mapFn: (item: T) => R): AsyncIterableObject<R> {
2174
return AsyncIterableObject.map(this, mapFn);
2175
}
2177
>
public static filter<T>(iterable: AsyncIterable<T>, filterFn: (item: T) => boolean): AsyncIterableObject<T> {
2178
return new AsyncIterableObject<T>(async (emitter) => {
2179
for await (const item of iterable) {
2180
if (filterFn(item)) {
2181
emitter.emitOne(item);
2182
}
2183
}
2184
});
2185
}
2187
>
public filter<T2 extends T>(filterFn: (item: T) => item is T2): AsyncIterableObject<T2>;
2188
>
public filter(filterFn: (item: T) => boolean): AsyncIterableObject<T>;
2189
>
public filter(filterFn: (item: T) => boolean): AsyncIterableObject<T> {
2190
return AsyncIterableObject.filter(this, filterFn);
2191
}
2193
>
public static coalesce<T>(iterable: AsyncIterable<T | undefined | null>): AsyncIterableObject<T> {
2194
return <AsyncIterableObject<T>>AsyncIterableObject.filter(iterable, item => !!item);
2195
}
2197
>
public coalesce(): AsyncIterableObject<NonNullable<T>> {
2198
return AsyncIterableObject.coalesce(this) as AsyncIterableObject<NonNullable<T>>;
2199
}
2201
>
public static async toPromise<T>(iterable: AsyncIterable<T>): Promise<T[]> {
2203
>
for await (const item of iterable) {
2205
>
}
2207
>
}
2209
>
public toPromise(): Promise<T[]> {
2211
>
}
2213
>
/**
2214
>
* The value will be appended at the end.
2215
>
*
2216
>
* **NOTE** If `resolve()` or `reject()` have already been called, this method has no effect.
2217
>
*/
2218
>
private emitOne(value: T): void {
2219
>
if (this._state !== AsyncIterableSourceState.Initial) {
extHostTerminalShellIntegration.ts ×13
2220
return;
2221
}
2223
>
// as we may have iterators already running on the array
2224
>
this._results.push(value);
2225
>
this._onStateChanged.fire();
2226
>
}
2228
>
/**
2229
>
* The values will be appended at the end.
2230
>
*
2231
>
* **NOTE** If `resolve()` or `reject()` have already been called, this method has no effect.
2232
>
*/
2233
>
private emitMany(values: T[]): void {
2234
>
if (this._state !== AsyncIterableSourceState.Initial) {
2235
return;
2236
}
2238
>
// as we may have iterators already running on the array
2239
>
this._results = this._results.concat(values);
2240
>
this._onStateChanged.fire();
2241
>
}
2242
>
2243
>
/**
2244
>
* Calling `resolve()` will mark the result array as complete.
2245
>
*
2246
>
* **NOTE** `resolve()` must be called, otherwise all consumers of this iterable will hang indefinitely, similar to a non-resolved promise.
2247
>
* **NOTE** If `resolve()` or `reject()` have already been called, this method has no effect.
2248
>
*/
2249
>
private resolve(): void {
2250
>
if (this._state !== AsyncIterableSourceState.Initial) {
2251
return;
2252
}
2254
>
this._onStateChanged.fire();
2255
>
}
2256
>
2257
>
/**
2258
>
* Writing an error will permanently invalidate this iterable.
2259
>
* The current users will receive an error thrown, as will all future users.
2260
>
*
2261
>
* **NOTE** If `resolve()` or `reject()` have already been called, this method has no effect.
2262
>
*/
2263
>
private reject(error: Error) {
2264
if (this._state !== AsyncIterableSourceState.Initial) {
2265
return;
2266
}
2267
this._state = AsyncIterableSourceState.DoneError;
2268
this._error = error;
2269
this._onStateChanged.fire();
2270
}
2272
>
2273
>
2274
>
export function createCancelableAsyncIterableProducer<T>(callback: (token: CancellationToken) => AsyncIterable<T>): CancelableAsyncIterableProducer<T> {
2275
const source = new CancellationTokenSource();
2276
const innerIterable = callback(source.token);
2277
2278
return new CancelableAsyncIterableProducer<T>(source, async (emitter) => {
2279
const subscription = source.token.onCancellationRequested(() => {
2280
subscription.dispose();
2281
source.dispose();
2282
emitter.reject(new CancellationError());
2283
});
2284
try {
2285
for await (const item of innerIterable) {
2286
if (source.token.isCancellationRequested) {
2287
// canceled in the meantime
2288
return;
2289
}
2290
emitter.emitOne(item);
2291
}
2292
subscription.dispose();
2293
source.dispose();
2294
} catch (err) {
2295
subscription.dispose();
2296
source.dispose();
2297
emitter.reject(err);
2298
}
2299
});
2300
}
2302
>
export class AsyncIterableSource<T> {
2303
>
2304
>
private readonly _deferred = new DeferredPromise<void>();
2305
>
private readonly _asyncIterable: AsyncIterableObject<T>;
2306
>
2307
>
private _errorFn: (error: Error) => void;
2308
>
private _emitOneFn: (item: T) => void;
2309
>
private _emitManyFn: (item: T[]) => void;
2310
>
2311
>
/**
2312
>
*
2313
>
* @param onReturn A function that will be called when consuming the async iterable
2314
>
* has finished by the consumer, e.g the for-await-loop has be existed (break, return) early.
2315
>
* This is NOT called when resolving this source by its owner.
2316
>
*/
2317
>
constructor(onReturn?: () => Promise<void> | void) {
2319
>
2320
>
if (earlyError) {
2321
emitter.reject(earlyError);
2322
return;
2323
}
2326
>
}
2328
>
this._emitOneFn = (item: T) => emitter.emitOne(item);
2329
>
this._emitManyFn = (items: T[]) => emitter.emitMany(items);
2330
>
return this._deferred.p;
2331
>
}, onReturn);
2332
>
2333
>
let earlyError: Error | undefined;
2334
>
let earlyItems: T[] | undefined;
2335
>
2336
>
2337
>
this._errorFn = (error: Error) => {
2338
if (!earlyError) {
2339
earlyError = error;
2340
}
2341
};
2344
>
earlyItems = [];
2345
>
}
2346
>
earlyItems.push(item);
2347
>
};
2350
>
earlyItems = items.slice();
2351
>
} else {
2352
items.forEach(item => earlyItems!.push(item));
2353
}
2357
>
get asyncIterable(): AsyncIterableObject<T> {
2359
>
}
2361
>
resolve(): void {
2363
>
}
2365
>
reject(error: Error): void {
2366
this._errorFn(error);
2367
this._deferred.complete();
2368
}
2370
>
emitOne(item: T): void {
2372
>
}
2374
>
emitMany(items: T[]) {
2376
>
}
2378
>
2379
>
export function cancellableIterable<T>(iterableOrIterator: AsyncIterator<T> | AsyncIterable<T>, token: CancellationToken): AsyncIterableIterator<T> {
2380
>
const iterator = Symbol.asyncIterator in iterableOrIterator ? iterableOrIterator[Symbol.asyncIterator]() : iterableOrIterator;
async.ts ×2
2381
>
2382
>
return {
2383
>
async next(): Promise<IteratorResult<T>> {
2384
>
if (token.isCancellationRequested) {
2386
>
}
2388
>
return result || { done: true, value: undefined };
2390
>
throw: iterator.throw?.bind(iterator),
2391
>
return: iterator.return?.bind(iterator),
2392
>
[Symbol.asyncIterator]() {
2393
>
return this;
2394
>
}
2395
>
};
2396
>
}
2398
>
type ProducerConsumerValue<T> = {
2399
>
ok: true;
2400
>
value: T;
2401
>
} | {
2402
>
ok: false;
2403
>
error: Error;
2404
>
};
2405
>
2406
>
class ProducerConsumer<T> {
2407
>
private readonly _unsatisfiedConsumers: DeferredPromise<T>[] = [];
2408
>
private readonly _unconsumedValues: ProducerConsumerValue<T>[] = [];
2409
>
private _finalValue: ProducerConsumerValue<T> | undefined;
2410
>
2411
>
public get hasFinalValue(): boolean {
2412
>
return !!this._finalValue;
2413
>
}
2414
>
2415
>
produce(value: ProducerConsumerValue<T>): void {
2417
>
if (this._unsatisfiedConsumers.length > 0) {
2418
>
const deferred = this._unsatisfiedConsumers.shift()!;
2419
>
this._resolveOrRejectDeferred(deferred, value);
2420
>
} else {
2422
>
}
2425
>
produceFinal(value: ProducerConsumerValue<T>): void {
2426
>
this._ensureNoFinalValue();
2427
>
this._finalValue = value;
2428
>
for (const deferred of this._unsatisfiedConsumers) {
2430
>
}
2432
>
}
2433
>
2434
>
private _ensureNoFinalValue(): void {
2435
>
if (this._finalValue) {
2436
throw new BugIndicatingError('ProducerConsumer: cannot produce after final value has been set');
2437
}
2439
>
2440
>
private _resolveOrRejectDeferred(deferred: DeferredPromise<T>, value: ProducerConsumerValue<T>): void {
2445
>
}
2448
>
consume(): Promise<T> {
2450
>
const value = this._unconsumedValues.length > 0 ? this._unconsumedValues.shift()! : this._finalValue!;
async.ts ×2
2451
>
if (value.ok) {
2455
>
}
2457
>
const deferred = new DeferredPromise<T>();
2458
>
this._unsatisfiedConsumers.push(deferred);
2459
>
return deferred.p;
2460
>
}
2461
>
}
2463
>
2464
>
/**
2465
>
* Important difference to AsyncIterableObject:
2466
>
* If it is iterated two times, the second iterator will not see the values emitted by the first iterator.
2467
>
*/
2468
>
export class AsyncIterableProducer<T> implements AsyncIterable<T> {
2469
>
private readonly _producerConsumer = new ProducerConsumer<IteratorResult<T>>();
2470
>
2471
>
constructor(executor: AsyncIterableExecutor<T>, private readonly _onReturn?: () => void) {
2472
>
queueMicrotask(async () => {
2473
>
const p = executor({
2474
>
emitOne: value => this._producerConsumer.produce({ ok: true, value: { done: false, value: value } }),
2475
>
emitMany: values => {
2476
>
for (const value of values) {
2477
>
this._producerConsumer.produce({ ok: true, value: { done: false, value: value } });
async.ts ×1
2478
>
}
2480
>
reject: error => this._finishError(error),
2481
>
});
2482
>
2483
>
if (!this._producerConsumer.hasFinalValue) {
2484
>
try {
2485
>
await p;
2486
>
this._finishOk();
2487
>
} catch (error) {
2489
>
}
2491
>
});
2492
>
}
2493
>
2494
>
public static fromArray<T>(items: T[]): AsyncIterableProducer<T> {
2495
>
return new AsyncIterableProducer<T>((writer) => {
2496
>
writer.emitMany(items);
2497
>
});
2498
>
}
2499
>
2500
>
public static fromPromise<T>(promise: Promise<T[]>): AsyncIterableProducer<T> {
2501
return new AsyncIterableProducer<T>(async (emitter) => {
2502
emitter.emitMany(await promise);
2503
});
2504
}
2506
>
public static fromPromisesResolveOrder<T>(promises: Promise<T>[]): AsyncIterableProducer<T> {
2507
return new AsyncIterableProducer<T>(async (emitter) => {
2508
await Promise.all(promises.map(async (p) => emitter.emitOne(await p)));
2509
});
2510
}
2512
>
public static merge<T>(iterables: AsyncIterable<T>[]): AsyncIterableProducer<T> {
2513
return new AsyncIterableProducer(async (emitter) => {
2514
await Promise.all(iterables.map(async (iterable) => {
2515
for await (const item of iterable) {
2516
emitter.emitOne(item);
2517
}
2518
}));
2519
});
2520
}
2522
>
public static EMPTY = AsyncIterableProducer.fromArray<any>([]);
2523
>
2524
>
public static map<T, R>(iterable: AsyncIterable<T>, mapFn: (item: T) => R): AsyncIterableProducer<R> {
2525
return new AsyncIterableProducer<R>(async (emitter) => {
2526
for await (const item of iterable) {
2527
emitter.emitOne(mapFn(item));
2528
}
2529
});
2530
}
2532
>
public static tee<T>(iterable: AsyncIterable<T>): [AsyncIterableProducer<T>, AsyncIterableProducer<T>] {
2534
>
let emitter2: AsyncIterableEmitter<T> | undefined;
2535
>
2536
>
const defer = new DeferredPromise<void>();
2537
>
2538
>
const start = async () => {
2539
>
if (!emitter1 || !emitter2) {
2540
>
return; // not yet ready
2541
>
}
2542
>
try {
2543
>
for await (const item of iterable) {
2544
>
emitter1.emitOne(item);
2545
>
emitter2.emitOne(item);
2546
>
}
2547
>
} catch (err) {
2548
emitter1.reject(err);
2549
emitter2.reject(err);
2551
>
defer.complete();
2552
>
}
2553
>
};
2554
>
2555
>
const p1 = new AsyncIterableProducer<T>(async (emitter) => {
2556
>
emitter1 = emitter;
2557
>
start();
2558
>
return defer.p;
2559
>
});
2560
>
const p2 = new AsyncIterableProducer<T>(async (emitter) => {
2561
>
emitter2 = emitter;
2562
>
start();
2563
>
return defer.p;
2564
>
});
2565
>
return [p1, p2];
2566
>
}
2568
>
public map<R>(mapFn: (item: T) => R): AsyncIterableProducer<R> {
2569
return AsyncIterableProducer.map(this, mapFn);
2570
}
2572
>
public static coalesce<T>(iterable: AsyncIterable<T | undefined | null>): AsyncIterableProducer<T> {
2573
return <AsyncIterableProducer<T>>AsyncIterableProducer.filter(iterable, item => !!item);
2574
}
2576
>
public coalesce(): AsyncIterableProducer<NonNullable<T>> {
2577
return AsyncIterableProducer.coalesce(this) as AsyncIterableProducer<NonNullable<T>>;
2578
}
2580
>
public static filter<T>(iterable: AsyncIterable<T>, filterFn: (item: T) => boolean): AsyncIterableProducer<T> {
2581
return new AsyncIterableProducer<T>(async (emitter) => {
2582
for await (const item of iterable) {
2583
if (filterFn(item)) {
2584
emitter.emitOne(item);
2585
}
2586
}
2587
});
2588
}
2590
>
public filter<T2 extends T>(filterFn: (item: T) => item is T2): AsyncIterableProducer<T2>;
2591
>
public filter(filterFn: (item: T) => boolean): AsyncIterableProducer<T>;
2592
>
public filter(filterFn: (item: T) => boolean): AsyncIterableProducer<T> {
2593
return AsyncIterableProducer.filter(this, filterFn);
2594
}
2596
>
private _finishOk(): void {
2597
>
if (!this._producerConsumer.hasFinalValue) {
2598
>
this._producerConsumer.produceFinal({ ok: true, value: { done: true, value: undefined } });
2599
>
}
2600
>
}
2601
>
2602
>
private _finishError(error: Error): void {
2604
>
this._producerConsumer.produceFinal({ ok: false, error: error });
2605
>
}
2606
>
// Warning: this can cause to dropped errors.
2607
>
}
2609
>
private readonly _iterator: AsyncIterator<T, void, void> = {
2610
>
next: () => this._producerConsumer.consume(),
2611
>
return: () => {
2612
this._onReturn?.();
2613
return Promise.resolve({ done: true, value: undefined });
2614
},
2616
this._finishError(e);
2617
return { done: true, value: undefined };
2618
},
2620
>
2621
>
[Symbol.asyncIterator](): AsyncIterator<T, void, void> {
2623
>
}
2625
>
2626
>
export class CancelableAsyncIterableProducer<T> extends AsyncIterableProducer<T> {
2627
>
constructor(
2628
private readonly _source: CancellationTokenSource,
2629
executor: AsyncIterableExecutor<T>
2630
) {
2631
super(executor);
2632
}
2634
>
cancel(): void {
2635
this._source.cancel();
2636
}
2638
>
2639
>
//#endregion
2640
>
2641
>
export const AsyncReaderEndOfStream = Symbol('AsyncReaderEndOfStream');
2642
>
2643
>
export class AsyncReader<T> {
2644
>
private _buffer: T[] = [];
2645
>
private _atEnd = false;
2646
>
2647
>
public get endOfStream(): boolean { return this._buffer.length === 0 && this._atEnd; }
2648
>
private _extendBufferPromise: Promise<void> | undefined;
2649
>
2650
>
constructor(
2652
>
) {
2653
>
}
2655
>
public async read(): Promise<T | typeof AsyncReaderEndOfStream> {
2658
>
}
2661
>
}
2665
>
public async readWhile(predicate: (value: T) => boolean, callback: (element: T) => unknown): Promise<void> {
2667
>
const piece = await this.peek();
2668
>
if (piece === AsyncReaderEndOfStream) {
2670
>
}
2673
>
}
2675
>
await callback(piece);
2677
>
}
2679
>
public readBufferedOrThrow(): T | typeof AsyncReaderEndOfStream {
2681
>
this._buffer.shift();
2682
>
return value;
2683
>
}
2685
>
public async consumeToEnd(): Promise<void> {
2687
>
await this.read();
2688
>
}
2689
>
}
2691
>
public async peek(): Promise<T | typeof AsyncReaderEndOfStream> {
2693
>
await this._extendBuffer();
2694
>
}
2695
>
if (this._buffer.length === 0) {
2697
>
}
2701
>
public peekBufferedOrThrow(): T | typeof AsyncReaderEndOfStream {
2705
>
}
2707
>
}
2709
>
return this._buffer[0];
2712
>
public async peekTimeout(timeoutMs: number): Promise<T | typeof AsyncReaderEndOfStream | undefined> {
2715
>
}
2718
>
}
2721
>
}
2725
>
private _extendBuffer(): Promise<void> {
2727
return Promise.resolve();
2728
}
2730
>
if (!this._extendBufferPromise) {
2731
>
this._extendBufferPromise = (async () => {
2732
>
const { value, done } = await this._source.next();
2733
>
this._extendBufferPromise = undefined;
2734
>
if (done) {
2738
>
}
2740
>
}
2741
>
2742
>
return this._extendBufferPromise;
2743
>
}
2745
>
2746
>
export function createTimeout(ms: number, cb: () => void): IDisposable {
2747
const t = setTimeout(cb, ms);
2748
return toDisposable(() => clearTimeout(t));
2749
}