event.test|title=Event utils EventMultiplexer hot start works|occurrence=1

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));