Vihren Atlas

async.ts ×202

Code frontier

unlabeled · c_78b40b2db89f

5963 tests · 6855 LOC · 31 files · introduces 0 tests · 1247 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
203 ranges1247 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
978 ranges6855 lines · 31 files · Browse complete extent
All tests (intent)
5963 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

2 files ranked by introduced lines: 1247 introduced LOC across 203 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

src/vs/base/common/async.ts 1238 introduced LOC · 202 ranges

Open complete file

1 > /*--------------------------------------------------------------------------------------------- async.ts
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> {
17 return !!obj && typeof (obj as unknown as Promise<T>).then === 'function';
18 }
19 > async.ts
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> {
35 const source = new CancellationTokenSource();
36
80 };
81 }
82 > async.ts
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> {
96 return new Promise((resolve, reject) => {
97 const ref = token.onCancellationRequested(() => {
102 });
103 }
104 > async.ts
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> {
110 return new Promise((resolve, reject) => {
111 const ref = token.onCancellationRequested(() => {
116 });
117 }
118 > async.ts
119 > export function rejectIfNotCanceled(err: unknown): undefined {
120 if (isCancellationError(err)) {
121 return undefined;
123 return Promise.reject(err) as never;
124 }
125 > async.ts
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> {
132 return new Promise<T>((resolve, reject) => {
133 promise.then(resolve, reject);
134 });
135 }
136 > async.ts
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> {
141 let resolvedPromiseIndex = -1;
142 const promises = cancellablePromises.map((promise, index) => promise.then(result => { resolvedPromiseIndex = index; return result; }));
154 return promise;
155 }
156 > async.ts
157 > export function raceTimeout<T>(promise: Promise<T>, timeout: number, onTimeout?: () => void): Promise<T | undefined> {
158 let promiseResolve: ((value: T | undefined) => void) | undefined = undefined;
159
168 ]);
169 }
170 > async.ts
171 > export function asPromise<T>(callback: () => T | Thenable<T>): Promise<T> {
172 return new Promise<T>((resolve, reject) => {
173 const item = callback();
179 });
180 }
181 > async.ts
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 } {
188 let resolve: (value: T | PromiseLike<T>) => void;
189 let reject: (reason?: any) => void;
194 return { promise, resolve: resolve!, reject: reject! };
195 }
196 > async.ts
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() {
239 this.activePromise = null;
240 this.queuedPromise = null;
243 this.cancellationTokenSource = new CancellationTokenSource();
244 }
245 > async.ts
246 > queue<T>(promiseFactory: ICancellableTask<Promise<T>>): Promise<T> {
247 if (this.cancellationTokenSource.token.isCancellationRequested) {
248 return Promise.reject(new Error('Throttler is disposed'));
288 });
289 }
290 > async.ts
291 > dispose(): void {
292 this.cancellationTokenSource.cancel();
293 }
294 > } async.ts
295 >
296 > export class Sequencer {
297
298 private current: Promise<unknown> = Promise.resolve(null);
299 > async.ts
300 > queue<T>(promiseTask: ITask<Promise<T>>): Promise<T> {
301 return this.current = this.current.then(() => promiseTask(), () => promiseTask());
302 }
303 > } async.ts
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 {
311
312 private readonly throttlers = new Map<TKey, { throttler: Throttler; count: number }>();
313 > async.ts
314 > queue<T>(key: TKey, task: ITask<Promise<T>>): Promise<T> {
315 let entry = this.throttlers.get(key);
316 if (!entry) {
327 });
328 }
329 > async.ts
330 > dispose(): void {
331 for (const { throttler } of this.throttlers.values()) {
332 throttler.dispose();
334 this.throttlers.clear();
335 }
336 > } async.ts
337 >
338 > export class SequencerByKey<TKey> {
339
340 private promiseMap = new Map<TKey, Promise<unknown>>();
341 > async.ts
342 > queue<T>(key: TKey, promiseTask: ITask<Promise<T>>): Promise<T> {
343 const runningPromise = this.promiseMap.get(key) ?? Promise.resolve();
344 const newPromise = runningPromise
353 return newPromise;
354 }
355 > async.ts
356 > peek(key: TKey): Promise<unknown> | undefined {
357 return this.promiseMap.get(key) || undefined;
358 }
359 > async.ts
360 > keys(): IterableIterator<TKey> {
361 return this.promiseMap.keys();
362 }
363 > } async.ts
364 >
365 > interface IScheduledLater extends IDisposable {
366 > isTriggered(): boolean;
367 > }
368 >
369 > const timeoutDeferred = (timeout: number, fn: () => void): IScheduledLater => {
370 let scheduled = true;
371 const handle = setTimeout(() => {
381 };
382 };
383 > async.ts
384 > const microtaskDeferred = (fn: () => void): IScheduledLater => {
385 let scheduled = true;
386 queueMicrotask(() => {
396 };
397 };
398 > async.ts
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) {
431 this.deferred = null;
432 this.completionPromise = null;
435 this.task = null;
436 }
437 > async.ts
438 > trigger(task: ITask<T | Promise<T>>, delay = this.defaultDelay): Promise<T> {
439 this.task = task;
440 this.cancelTimeout();
465 return this.completionPromise;
466 }
467 > async.ts
468 > isTriggered(): boolean {
469 return !!this.deferred?.isTriggered();
470 }
471 > async.ts
472 > cancel(): void {
473 this.cancelTimeout();
474
478 }
479 }
480 > async.ts
481 > private cancelTimeout(): void {
482 this.deferred?.dispose();
483 this.deferred = null;
484 }
485 > async.ts
486 > dispose(): void {
487 this.cancel();
488 }
489 > } async.ts
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) {
506 this.delayer = new Delayer(defaultDelay);
507 this.throttler = new Throttler();
508 }
509 > async.ts
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>;
512 }
513 > async.ts
514 > isTriggered(): boolean {
515 return this.delayer.isTriggered();
516 }
517 > async.ts
518 > cancel(): void {
519 this.delayer.cancel();
520 }
521 > async.ts
522 > dispose(): void {
523 this.delayer.dispose();
524 this.throttler.dispose();
525 }
526 > } async.ts
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() {
537 this._isOpen = false;
538 this._promise = new Promise<boolean>((c, e) => {
540 });
541 }
542 > async.ts
543 > isOpen(): boolean {
544 return this._isOpen;
545 }
546 > async.ts
547 > open(): void {
548 this._isOpen = true;
549 this._completePromise(true);
550 }
551 > async.ts
552 > wait(): Promise<boolean> {
553 return this._promise;
554 }
555 > } async.ts
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 }
569 > async.ts
570 > override open(): void {
571 clearTimeout(this._timeout);
572 super.open();
573 }
574 > } async.ts
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> {
579 if (!token) {
580 return createCancelablePromise(token => timeout(millis, token));
593 });
594 }
595 > async.ts
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 {
614 const timer = setTimeout(() => {
615 handler();
625 return disposable;
626 }
627 > async.ts
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 {
649 const target = Date.now() + timeout;
650 let timer: Timeout;
671 return disposable;
672 }
673 > async.ts
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[]> {
680 const results: T[] = [];
681 let index = 0;
701 return Promise.resolve(null).then(thenHandler);
702 }
703 > async.ts
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;
725 return loop();
726 }
727 > async.ts
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) {
735 if (promiseList.length === 0) {
736 return Promise.resolve(defaultValue);
764 });
765 }
766 > async.ts
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) {
796 this.maxDegreeOfParalellism = maxDegreeOfParalellism;
797 this.outstandingPromises = [];
799 this._onDrained = new Emitter<void>();
800 }
801 > async.ts
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> {
808 return this.size > 0
809 ? Event.toPromise(this.onDrained)
810 : Promise.resolve();
811 }
812 > async.ts
813 > get onDrained(): Event<void> {
814 return this._onDrained.event;
815 }
816 > async.ts
817 > get size(): number {
818 return this._size;
819 }
820 > async.ts
821 > queue(factory: ITask<Promise<T>>): Promise<T> {
822 if (this._isDisposed) {
823 throw new Error('Object has been disposed');
830 });
831 }
832 > async.ts
833 > private consume(): void {
834 while (this.outstandingPromises.length && this.runningPromises < this.maxDegreeOfParalellism) {
835 const iLimitedTask = this.outstandingPromises.shift()!;
841 }
842 }
843 > async.ts
844 > private consumed(): void {
845 if (this._isDisposed) {
846 return;
855 }
856 }
857 > async.ts
858 > clear(): void {
859 if (this._isDisposed) {
860 throw new Error('Object has been disposed');
863 this._size = this.runningPromises;
864 }
865 > async.ts
866 > dispose(): void {
867 this._isDisposed = true;
868 this.outstandingPromises.length = 0; // stop further processing
870 this._onDrained.dispose();
871 }
872 > } async.ts
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() {
880 super(1);
881 }
882 > } async.ts
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 {
893
894 private readonly sequentializer = new TaskSequentializer();
895
896 private tasks = 0;
897 > async.ts
898 > queue(factory: ITask<Promise<void>>): Promise<void> {
899 if (!this.sequentializer.isRunning()) {
900 return this.sequentializer.run(this.tasks++, factory());
905 });
906 }
907 > } async.ts
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 {
914
915 private readonly queues = new Map<string, Queue<void>>();
919 private drainListeners: DisposableMap<number> | undefined = undefined;
920 private drainListenerCount = 0;
921 > async.ts
922 > async whenDrained(): Promise<void> {
923 if (this.isDrained()) {
924 return;
930 return promise.p;
931 }
932 > async.ts
933 > private isDrained(): boolean {
934 for (const [, queue] of this.queues) {
935 if (queue.size > 0) {
940 return true;
941 }
942 > async.ts
943 > queueSize(resource: URI, extUri: IExtUri = defaultExtUri): number {
944 const key = extUri.getComparisonKey(resource);
945
946 return this.queues.get(key)?.size ?? 0;
947 }
948 > async.ts
949 > queueFor(resource: URI, factory: ITask<Promise<void>>, extUri: IExtUri = defaultExtUri): Promise<void> {
950 const key = extUri.getComparisonKey(resource);
951
977 return queue.queue(factory);
978 }
979 > async.ts
980 > private onDidQueueDrain(): void {
981 if (!this.isDrained()) {
982 return; // not done yet
985 this.releaseDrainers();
986 }
987 > async.ts
988 > private releaseDrainers(): void {
989 for (const drainer of this.drainers) {
990 drainer.complete();
993 this.drainers.clear();
994 }
995 > async.ts
996 > dispose(): void {
997 for (const [, queue] of this.queues) {
998 queue.dispose();
1011 this.drainListeners?.dispose();
1012 }
1013 > } async.ts
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 }[] = [];
1029 > async.ts
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 });
1038 return deferred.p;
1039 }
1040 > async.ts
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 });
1049 return deferred.p;
1050 }
1051 > async.ts
1052 > private _runIfNotRunning(): void {
1053 if (this._runningTask === undefined) {
1054 this._processQueue();
1055 }
1056 }
1057 > async.ts
1058 > private async _processQueue(): Promise<void> {
1059 if (this._pendingTasks.length === 0) {
1060 return;
1082 }
1083 }
1084 > async.ts
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 = [];
1097 }
1098 }
1099 > } async.ts
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) {
1108 this._token = undefined;
1109
1112 }
1113 }
1114 > async.ts
1115 > dispose(): void {
1116 this.cancel();
1117 this._isDisposed = true;
1118 }
1119 > async.ts
1120 > cancel(): void {
1121 if (this._token !== undefined) {
1122 clearTimeout(this._token);
1124 }
1125 }
1126 > async.ts
1127 > cancelAndSet(runner: () => void, timeout: number): void {
1128 if (this._isDisposed) {
1129 throw new BugIndicatingError(`Calling 'cancelAndSet' on a disposed TimeoutTimer`);
1136 }, timeout);
1137 }
1138 > async.ts
1139 > setIfNotSet(runner: () => void, timeout: number): void {
1140 if (this._isDisposed) {
1141 throw new BugIndicatingError(`Calling 'setIfNotSet' on a disposed TimeoutTimer`);
1151 }, timeout);
1152 }
1153 > } async.ts
1154 >
1155 > export class IntervalTimer implements IDisposable {
1156
1157 private disposable: IDisposable | undefined = undefined;
1158 private isDisposed = false;
1159 > async.ts
1160 > cancel(): void {
1161 this.disposable?.dispose();
1162 this.disposable = undefined;
1163 }
1164 > async.ts
1165 > cancelAndSet(runner: () => void, interval: number, context = globalThis): void {
1166 if (this.isDisposed) {
1167 throw new BugIndicatingError(`Calling 'cancelAndSet' on a disposed IntervalTimer`);
1178 });
1179 }
1180 > async.ts
1181 > dispose(): void {
1182 this.cancel();
1183 this.isDisposed = true;
1184 }
1185 > } async.ts
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) {
1196 this.timeoutToken = undefined;
1197 this.runner = runner;
1199 this.timeoutHandler = this.onTimeout.bind(this);
1200 }
1201 > async.ts
1202 > /**
1203 > * Dispose RunOnceScheduler
1204 > */
1205 > dispose(): void {
1206 this.cancel();
1207 this.runner = null;
1208 }
1209 > async.ts
1210 > /**
1211 > * Cancel current scheduled runner (if any).
1212 > */
1213 > cancel(): void {
1214 if (this.isScheduled()) {
1215 clearTimeout(this.timeoutToken);
1217 }
1218 }
1219 > async.ts
1220 > /**
1221 > * Cancel previous runner (if any) & schedule a new runner.
1222 > */
1223 > schedule(delay = this.timeout): void {
1224 this.cancel();
1225 this.timeoutToken = setTimeout(this.timeoutHandler, delay);
1226 }
1227 > async.ts
1228 > get delay(): number {
1229 return this.timeout;
1230 }
1231 > async.ts
1232 > set delay(value: number) {
1233 this.timeout = value;
1234 }
1235 > async.ts
1236 > /**
1237 > * Returns true if scheduled.
1238 > */
1239 > isScheduled(): boolean {
1240 return this.timeoutToken !== undefined;
1241 }
1242 > async.ts
1243 > flush(): void {
1244 if (this.isScheduled()) {
1245 this.cancel();
1247 }
1248 }
1249 > async.ts
1250 > private onTimeout() {
1251 this.timeoutToken = undefined;
1252 if (this.runner) {
1254 }
1255 }
1256 > async.ts
1257 > protected doRun(): void {
1258 this.runner?.();
1259 }
1260 > } async.ts
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.`);
1287 this.intervalHandler = this.onInterval.bind(this);
1288 }
1289 > async.ts
1290 > dispose(): void {
1291 this.cancel();
1292 this.runner = null;
1293 }
1294 > async.ts
1295 > cancel(): void {
1296 if (this.isScheduled()) {
1297 clearInterval(this.intervalToken);
1299 }
1300 }
1301 > async.ts
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.`);
1311 this.intervalToken = setInterval(this.intervalHandler, 1000);
1312 }
1313 > async.ts
1314 > /**
1315 > * Returns true if scheduled.
1316 > */
1317 > isScheduled(): boolean {
1318 return this.intervalToken !== undefined;
1319 }
1320 > async.ts
1321 > private onInterval() {
1322 this.counter--;
1323 if (this.counter > 0) {
1331 this.runner?.();
1332 }
1333 > } async.ts
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 }
1342 > async.ts
1343 > work(unit: T): void {
1344 this.units.push(unit);
1345
1348 }
1349 }
1350 > async.ts
1351 > protected override doRun(): void {
1352 const units = this.units;
1353 this.units = [];
1355 this.runner?.(units);
1356 }
1357 > async.ts
1358 > override dispose(): void {
1359 this.units = [];
1360
1361 super.dispose();
1362 }
1363 > } async.ts
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(
1406 private options: IThrottledWorkerOptions,
1407 private readonly handler: (units: T[]) => void
1409 super();
1410 }
1411 > async.ts
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 {
1428 if (this.disposed) {
1429 return false; // work not accepted: disposed
1469 return true; // work accepted
1470 }
1471 > async.ts
1472 > private doWork(): void {
1473 this.lastExecutionTime = Date.now();
1474
1481 }
1482 }
1483 > async.ts
1484 > private scheduleThrottler(delay = this.options.throttleDelay): void {
1485 this.throttler.value = new RunOnceScheduler(() => {
1486 this.throttler.clear();
1490 this.throttler.value.schedule();
1491 }
1492 > async.ts
1493 > override dispose(): void {
1494 super.dispose();
1495
1497 this.disposed = true;
1498 }
1499 > } async.ts
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?) => {
1540 setTimeout0(() => {
1541 if (disposed) {
1561 };
1562 };
1563 > } else { async.ts
1564 _runWhenIdle = (targetWindow: typeof safeGlobal, runner, timeout?) => {
1565 const handle: number = targetWindow.requestIdleCallback(runner, typeof timeout === 'number' ? { timeout } : undefined);
1576 };
1577 }
1578 > runWhenGlobalIdle = (runner, timeout) => _runWhenIdle(globalThis, runner, timeout); async.ts
1579 > })();
1580 >
1581 > export function installFakeRunWhenIdle(fakeImpl: typeof _runWhenIdle): IDisposable {
1582 const origRunWhenIdle = _runWhenIdle;
1583 const origRunWhenGlobalIdle = runWhenGlobalIdle;
1589 });
1590 }
1591 > async.ts
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) {
1602 this._executor = () => {
1603 try {
1611 this._handle = _runWhenIdle(targetWindow, () => this._executor());
1612 }
1613 > async.ts
1614 > dispose(): void {
1615 this._handle.dispose();
1616 }
1617 > async.ts
1618 > get value(): T {
1619 if (!this._didRun) {
1620 this._handle.dispose();
1626 return this._value!;
1627 }
1628 > async.ts
1629 > get isInitialized(): boolean {
1630 return this._didRun;
1631 }
1632 > } async.ts
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) {
1643 super(globalThis, executor);
1644 }
1645 > } async.ts
1646 >
1647 > //#endregion
1648 >
1649 export async function retry<T>(task: ITask<Promise<T>>, delay: number, retries: number): Promise<T> {
1650 let lastError: Error | undefined;
1662 throw lastError;
1663 }
1664 > async.ts
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 {
1697 if (typeof taskId === 'number') {
1698 return this._running?.taskId === taskId;
1701 return !!this._running;
1702 }
1703 > async.ts
1704 > get running(): Promise<void> | undefined {
1705 return this._running?.promise;
1706 }
1707 > async.ts
1708 > cancelRunning(): void {
1709 this._running?.cancel();
1710 }
1711 > async.ts
1712 > run(taskId: number, promise: Promise<void>, onCancel?: () => void,): Promise<void> {
1713 this._running = { taskId, cancel: () => onCancel?.(), promise };
1714
1717 return promise;
1718 }
1719 > async.ts
1720 > private doneRunning(taskId: number): void {
1721 if (this._running && taskId === this._running.taskId) {
1722
1728 }
1729 }
1730 > async.ts
1731 > private runQueued(): void {
1732 if (this._queued) {
1733 const queued = this._queued;
1738 }
1739 }
1740 > async.ts
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> {
1748
1749 // this is our first queued task, so we create associated promise with it
1767 return this._queued.promise;
1768 }
1769 > async.ts
1770 > hasQueued(): this is ITaskSequentializerWithQueuedTask {
1771 return !!this._queued;
1772 }
1773 > async.ts
1774 > async join(): Promise<void> {
1775 return this._queued?.promise ?? this._running?.promise;
1776 }
1777 > } async.ts
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 {
1799 const now = this.nowFn();
1800
1810 return this.value;
1811 }
1812 > } async.ts
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 }
1835 > async.ts
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() {
1841 return this.outcome?.outcome === DeferredOutcome.Rejected;
1842 }
1843 > async.ts
1844 > public get isResolved() {
1845 return this.outcome?.outcome === DeferredOutcome.Resolved;
1846 }
1847 > async.ts
1848 > public get isSettled() {
1849 return !!this.outcome;
1850 }
1851 > async.ts
1852 > public get value() {
1853 return this.outcome?.outcome === DeferredOutcome.Resolved ? this.outcome?.value : undefined;
1854 }
1855 > async.ts
1856 > public readonly p: Promise<T>;
1857 >
1858 > constructor() {
1859 this.p = new Promise<T>((c, e) => {
1860 this.completeCallback = c;
1862 });
1863 }
1864 > async.ts
1865 > public complete(value: T) {
1866 if (this.isSettled) {
1867 return Promise.resolve();
1874 });
1875 }
1876 > async.ts
1877 > public error(err: unknown) {
1878 if (this.isSettled) {
1879 return Promise.resolve();
1886 });
1887 }
1888 > async.ts
1889 > public settleWith(promise: Promise<T>): Promise<void> {
1890 return promise.then(
1891 value => this.complete(value),
1893 );
1894 }
1895 > async.ts
1896 > public cancel() {
1897 return this.error(new CancellationError());
1898 }
1899 > } async.ts
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[]> {
1915 let firstError: Error | undefined = undefined;
1916
1929 return result as unknown as T[]; // cast is needed and protected by the `throw` above
1930 }
1931 > async.ts
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> {
1943 // eslint-disable-next-line no-async-promise-executor
1944 return new Promise<T>(async (resolve, reject) => {
1950 });
1951 }
1952 > } async.ts
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 => {
1978 );
1979 }
1980 > async.ts
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');
1992 return this._value!;
1993 }
1994 > } async.ts
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 ) { }
2002 > async.ts
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 }
2010 > async.ts
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 }
2017 > async.ts
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 }
2024 > } async.ts
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 }
2088 > async.ts
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 }
2094 > async.ts
2095 > public static merge<T>(iterables: AsyncIterable<T>[]): AsyncIterableObject<T> {
2096 return new AsyncIterableObject(async (emitter) => {
2097 await Promise.all(iterables.map(async (iterable) => {
2102 });
2103 }
2104 > async.ts
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);
2131 > } finally { async.ts
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> {
2142 let i = 0;
2143 return {
2162 };
2163 }
2164 > async.ts
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) {
2170 });
2171 }
2172 > async.ts
2173 > public map<R>(mapFn: (item: T) => R): AsyncIterableObject<R> {
2174 return AsyncIterableObject.map(this, mapFn);
2175 }
2176 > async.ts
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) {
2184 });
2185 }
2186 > async.ts
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 }
2192 > async.ts
2193 > public static coalesce<T>(iterable: AsyncIterable<T | undefined | null>): AsyncIterableObject<T> {
2194 return <AsyncIterableObject<T>>AsyncIterableObject.filter(iterable, item => !!item);
2195 }
2196 > async.ts
2197 > public coalesce(): AsyncIterableObject<NonNullable<T>> {
2198 return AsyncIterableObject.coalesce(this) as AsyncIterableObject<NonNullable<T>>;
2199 }
2200 > async.ts
2201 > public static async toPromise<T>(iterable: AsyncIterable<T>): Promise<T[]> {
2202 const result: T[] = [];
2203 for await (const item of iterable) {
2206 return result;
2207 }
2208 > async.ts
2209 > public toPromise(): Promise<T[]> {
2210 return AsyncIterableObject.toPromise(this);
2211 }
2212 > async.ts
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) {
2220 return;
2225 this._onStateChanged.fire();
2226 }
2227 > async.ts
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 }
2237 > // it is important to add new values at the end, async.ts
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 }
2253 > this._state = AsyncIterableSourceState.DoneOK; async.ts
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;
2269 this._onStateChanged.fire();
2270 }
2271 > } async.ts
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);
2299 });
2300 }
2301 > async.ts
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) {
2318 this._asyncIterable = new AsyncIterableObject(emitter => {
2319
2354 };
2355 }
2356 > async.ts
2357 > get asyncIterable(): AsyncIterableObject<T> {
2358 return this._asyncIterable;
2359 }
2360 > async.ts
2361 > resolve(): void {
2362 this._deferred.complete();
2363 }
2364 > async.ts
2365 > reject(error: Error): void {
2366 this._errorFn(error);
2367 this._deferred.complete();
2368 }
2369 > async.ts
2370 > emitOne(item: T): void {
2371 this._emitOneFn(item);
2372 }
2373 > async.ts
2374 > emitMany(items: T[]) {
2375 this._emitManyFn(items);
2376 }
2377 > } async.ts
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;
2381
2395 };
2396 }
2397 > async.ts
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 {
2416 this._ensureNoFinalValue();
2417 if (this._unsatisfiedConsumers.length > 0) {
2422 }
2423 }
2424 > async.ts
2425 > produceFinal(value: ProducerConsumerValue<T>): void {
2426 > this._ensureNoFinalValue();
2427 > this._finalValue = value;
2428 > for (const deferred of this._unsatisfiedConsumers) {
2429 this._resolveOrRejectDeferred(deferred, value);
2430 }
2431 > this._unsatisfiedConsumers.length = 0; async.ts
2432 > }
2433 >
2434 > private _ensureNoFinalValue(): void {
2435 > if (this._finalValue) {
2436 throw new BugIndicatingError('ProducerConsumer: cannot produce after final value has been set');
2437 }
2438 > } async.ts
2439 >
2440 > private _resolveOrRejectDeferred(deferred: DeferredPromise<T>, value: ProducerConsumerValue<T>): void {
2441 if (value.ok) {
2442 deferred.complete(value.value);
2445 }
2446 }
2447 > async.ts
2448 > consume(): Promise<T> {
2449 if (this._unconsumedValues.length > 0 || this._finalValue) {
2450 const value = this._unconsumedValues.length > 0 ? this._unconsumedValues.shift()! : this._finalValue!;
2460 }
2461 }
2462 > } async.ts
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 } });
2478 }
2479 > }, async.ts
2480 > reject: error => this._finishError(error),
2481 > });
2482 >
2483 > if (!this._producerConsumer.hasFinalValue) {
2484 > try {
2485 > await p;
2486 > this._finishOk();
2487 > } catch (error) {
2488 this._finishError(error);
2489 }
2490 > } async.ts
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 }
2505 > async.ts
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 }
2511 > async.ts
2512 > public static merge<T>(iterables: AsyncIterable<T>[]): AsyncIterableProducer<T> {
2513 return new AsyncIterableProducer(async (emitter) => {
2514 await Promise.all(iterables.map(async (iterable) => {
2519 });
2520 }
2521 > async.ts
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) {
2529 });
2530 }
2531 > async.ts
2532 > public static tee<T>(iterable: AsyncIterable<T>): [AsyncIterableProducer<T>, AsyncIterableProducer<T>] {
2533 let emitter1: AsyncIterableEmitter<T> | undefined;
2534 let emitter2: AsyncIterableEmitter<T> | undefined;
2565 return [p1, p2];
2566 }
2567 > async.ts
2568 > public map<R>(mapFn: (item: T) => R): AsyncIterableProducer<R> {
2569 return AsyncIterableProducer.map(this, mapFn);
2570 }
2571 > async.ts
2572 > public static coalesce<T>(iterable: AsyncIterable<T | undefined | null>): AsyncIterableProducer<T> {
2573 return <AsyncIterableProducer<T>>AsyncIterableProducer.filter(iterable, item => !!item);
2574 }
2575 > async.ts
2576 > public coalesce(): AsyncIterableProducer<NonNullable<T>> {
2577 return AsyncIterableProducer.coalesce(this) as AsyncIterableProducer<NonNullable<T>>;
2578 }
2579 > async.ts
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) {
2587 });
2588 }
2589 > async.ts
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 }
2595 > async.ts
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 {
2603 if (!this._producerConsumer.hasFinalValue) {
2604 this._producerConsumer.produceFinal({ ok: false, error: error });
2606 // Warning: this can cause to dropped errors.
2607 }
2608 > async.ts
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 },
2615 > throw: async (e) => { async.ts
2616 this._finishError(e);
2617 return { done: true, value: undefined };
2618 },
2619 > }; async.ts
2620 >
2621 > [Symbol.asyncIterator](): AsyncIterator<T, void, void> {
2622 return this._iterator;
2623 }
2624 > } async.ts
2625 >
2626 > export class CancelableAsyncIterableProducer<T> extends AsyncIterableProducer<T> {
2627 > constructor(
2628 private readonly _source: CancellationTokenSource,
2629 executor: AsyncIterableExecutor<T>
2631 super(executor);
2632 }
2633 > async.ts
2634 > cancel(): void {
2635 this._source.cancel();
2636 }
2637 > } async.ts
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(
2651 private readonly _source: AsyncIterator<T>
2652 ) {
2653 }
2654 > async.ts
2655 > public async read(): Promise<T | typeof AsyncReaderEndOfStream> {
2656 if (this._buffer.length === 0 && !this._atEnd) {
2657 await this._extendBuffer();
2662 return this._buffer.shift()!;
2663 }
2664 > async.ts
2665 > public async readWhile(predicate: (value: T) => boolean, callback: (element: T) => unknown): Promise<void> {
2666 do {
2667 const piece = await this.peek();
2676 } while (true);
2677 }
2678 > async.ts
2679 > public readBufferedOrThrow(): T | typeof AsyncReaderEndOfStream {
2680 const value = this.peekBufferedOrThrow();
2681 this._buffer.shift();
2682 return value;
2683 }
2684 > async.ts
2685 > public async consumeToEnd(): Promise<void> {
2686 while (!this.endOfStream) {
2687 await this.read();
2688 }
2689 }
2690 > async.ts
2691 > public async peek(): Promise<T | typeof AsyncReaderEndOfStream> {
2692 if (this._buffer.length === 0 && !this._atEnd) {
2693 await this._extendBuffer();
2698 return this._buffer[0];
2699 }
2700 > async.ts
2701 > public peekBufferedOrThrow(): T | typeof AsyncReaderEndOfStream {
2702 if (this._buffer.length === 0) {
2703 if (this._atEnd) {
2709 return this._buffer[0];
2710 }
2711 > async.ts
2712 > public async peekTimeout(timeoutMs: number): Promise<T | typeof AsyncReaderEndOfStream | undefined> {
2713 if (this._buffer.length === 0 && !this._atEnd) {
2714 await raceTimeout(this._extendBuffer(), timeoutMs);
2722 return this._buffer[0];
2723 }
2724 > async.ts
2725 > private _extendBuffer(): Promise<void> {
2726 if (this._atEnd) {
2727 return Promise.resolve();
2742 return this._extendBufferPromise;
2743 }
2744 > } async.ts
2745 >
2746 > export function createTimeout(ms: number, cb: () => void): IDisposable {
2747 const t = setTimeout(cb, ms);
2748 return toDisposable(() => clearTimeout(t));
src/vs/base/common/symbols.ts 9 introduced LOC · 1 range

Open complete file

1 > /*--------------------------------------------------------------------------------------------- symbols.ts
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 > /**
7 > * Can be passed into the Delayed to defer using a microtask
8 > * */
9 > export const MicrotaskDelay = Symbol('MicrotaskDelay');

Direct hierarchy

Parents

Children