From 4b8628e60231ea15a343fabb6c4d461889ad6766 Mon Sep 17 00:00:00 2001 From: Okiki Date: Mon, 26 Aug 2024 07:45:46 +0000 Subject: [PATCH] feat: breakup the static methods of the Future class into treeshakeable functions Signed-off-by: Okiki --- idle.ts => _idle.ts | 0 _repl.ts | 2 +- utils.ts => _utils.ts | 0 background.ts | 63 +++ concurrent.ts | 187 +++++++ deadline.ts | 35 ++ errors.ts | 5 + from.ts | 294 ++++++++++ future.ts | 508 ++++++++++++++++++ mod.ts | 1195 +---------------------------------------- resolvers.ts | 22 + scope.ts | 81 +++ 12 files changed, 1204 insertions(+), 1188 deletions(-) rename idle.ts => _idle.ts (100%) rename utils.ts => _utils.ts (100%) create mode 100644 background.ts create mode 100644 concurrent.ts create mode 100644 deadline.ts create mode 100644 errors.ts create mode 100644 from.ts create mode 100644 future.ts create mode 100644 resolvers.ts create mode 100644 scope.ts diff --git a/idle.ts b/_idle.ts similarity index 100% rename from idle.ts rename to _idle.ts diff --git a/_repl.ts b/_repl.ts index 89e102b..e5ac7d3 100644 --- a/_repl.ts +++ b/_repl.ts @@ -1,4 +1,4 @@ -import Future from "./mod.ts"; +import * as Future from "./mod.ts"; const future = Future.from(async function* () { let count = 0; diff --git a/utils.ts b/_utils.ts similarity index 100% rename from utils.ts rename to _utils.ts diff --git a/background.ts b/background.ts new file mode 100644 index 0000000..09168b1 --- /dev/null +++ b/background.ts @@ -0,0 +1,63 @@ +import { Future } from "./future.ts"; +import { cancelIdle, idle } from "./_idle.ts"; + + /** + * Sets up a `Future` to execute in the background during idle time. + * + * This method does not execute the future itself, but prepares it to be run using + * `requestIdleCallback`. Execution is still controlled by methods like `toPromise()` or `async` iterators. + * + * @param future The future to be executed in the background. + * @returns A new `Future` instance set up for background execution. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * return 100; + * }); + * const backgroundFuture = Future.inBackground(future); // result is 100, processed in the background + * ``` + */ + export function inBackground(future: Future): Future { + // Check if the input is iterable + const generator = future?.[Symbol.asyncIterator]?.(); + + // Iterate over the iterable/async iterable futures in a controlled manner + return new Future(async function* () { + // If no valid iterator was found, throw an error indicating that the input is not iterable or an iterator + if ( + (generator ?? null) === null || + typeof (generator as AsyncGenerator)?.next === 'function' + ) throw new TypeError("The provided input is not a future."); + + let idleResolver: PromiseWithResolvers | null = Promise.withResolvers(); + let idleId = idle(() => idleResolver?.resolve?.()); + + try { + await idleResolver.promise; + cancelIdle(idleId); + + // Handle the async generator or generator in a pull-based workflow + let result = await generator.next(); + + idleResolver = Promise.withResolvers(); + idleId = idle(() => idleResolver?.resolve?.()); + + // Start the iteration + while (!result.done) { + await idleResolver.promise; + cancelIdle(idleId); + + result = await generator.next(yield result.value); + + idleResolver = Promise.withResolvers(); + idleId = idle(() => idleResolver?.resolve?.()); + } + + return result.value; + } finally { + idleResolver = null; + cancelIdle(idleId); + } + }); + } \ No newline at end of file diff --git a/concurrent.ts b/concurrent.ts new file mode 100644 index 0000000..df399e1 --- /dev/null +++ b/concurrent.ts @@ -0,0 +1,187 @@ +import { Future } from "./future.ts"; + +/** + * Runs multiple Futures concurrently, yielding their results as they all complete. + * + * This is similar to `Promise.all` but with support for yielding results in sequence + * once all the futures have been resolved. + * + * @param futures - An iterable of `Future` or `PromiseLike` objects. + * @returns An AsyncIterable yielding each result as they complete. + */ +export function all(futures: Iterable | PromiseLike>): Future, Awaited[]> { + return new Future, Awaited[]>(async function* () { + // We trigger all futures at once, using Promise.all to await them concurrently. + const results = await Promise.all(futures); + yield* results; + return results; + }); +} + +/** + * Executes multiple futures concurrently and yields their results or errors as soon as they settle. + * This method works similarly to `Promise.allSettled` but yields results incrementally. + * + * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. + * @returns An AsyncIterable yielding each settled result. + * @example + * ```typescript + * const futureSettled = Future.allSettled([ + * Future.from(async function* () { + * yield 42; + * return 100; + * }), + * Future.from(async function* () { + * yield 10; + * return 20; + * }) + * ]); + * + * for await (const result of futureSettled) { + * if (result.status === "fulfilled") { + * console.log(result.value); // Outputs 100 and 20 + * } else { + * console.error(result.reason); + * } + * } + * ``` + */ +export function allSettled( + futures: Iterable | PromiseLike> +): Future, PromiseSettledResult[]> { + return new Future, PromiseSettledResult[]>(async function* () { + // We trigger all futures at once, using Promise.allSettled to await them concurrently. + const results = await Promise.allSettled(futures); + yield* results; + return results; + }); +} + +/** + * Returns the first settled Future, similar to `Promise.race`. + * + * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. + * @returns An AsyncIterable yielding the first result that resolves. + */ +export function race( + futures: Iterable | PromiseLike> +): Future { + return new Future(async function* () { + const result = Promise.race(futures); + yield result; + return result; + }); +} + +/** + * Executes multiple Futures concurrently, but only returns results for the first `count` Futures. + * + * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. + * @param count - The number of Futures to resolve before yielding results. + * @returns An AsyncIterable yielding the first `count` resolved results. + */ +export function some( + futures: Iterable | PromiseLike>, + count: number +): Future, PromiseSettledResult[]> { + return Future.allSettled( + Array.from(futures).slice(0, count) + ); +} + +/** + * Limits the number of concurrent `Future` operations. + * + * This method is useful for controlling how many Futures run at the same time, + * which can prevent overwhelming resources or ensure a more manageable load. + * + * @param futures - An iterable of `Future` or `PromiseLike` objects. + * @param limit - The maximum number of Futures to run concurrently. + * @returns A `Future` that yields results as each operation completes. + * + * @example + * ```typescript + * const futures = [ + * Promise.resolve(1), + * Promise.resolve(2), + * Promise.resolve(3), + * Future.from(async function* () { + * yield 42; + * return 100; + * }), + * Future.from(async function* () { + * yield 10; + * return 20; + * }) + * ]; + * + * for await (const result of future) { + * console.log(result); // Logs 1, 2, 3, 100, 20 in order with a concurrency limit of 2 + * } + * ``` + * + * @example + * ```typescript + * const futures = [ + * fetch('https://api.example.com/1'), + * fetch('https://api.example.com/2'), + * fetch('https://api.example.com/3') + * ]; + * + * const limitedFuture = Future.withConcurrencyLimit(futures, 2); + * + * for await (const result of limitedFuture) { + * console.log(result); // Process each result as it becomes available + * } + * ``` + */ +export function withConcurrencyLimit( + futures: Iterable | PromiseLike>, + limit: number +): Future { + // Check if the input is an async iterator, sync iterator, or iterable + // We use Symbol.asyncIterator and Symbol.iterator to distinguish between different types + const iterator = futures?.[Symbol.iterator]?.(); + + return new Future(async function* () { + // If no valid iterator was found, throw an error indicating that the input is not iterable or an iterator + if ( + (iterator ?? null) === null || + typeof iterator?.next === 'function' + ) throw new TypeError("The provided input is not an iterable nor an iterator."); + + const activeFutures = new Map< + PromiseLike, + Promise<{ result: T | TReturn, promise: PromiseLike }> + >(); + let finalResult: T | TReturn | undefined; + + while (activeFutures.size < limit) { + const { done, value } = iterator.next(); + if (done) break; + + // Start the future concurrently + const promise = value instanceof Future ? value.toPromise() : value; + + // Wrap the promise and store it in the map with the original promise as the key + activeFutures.set(promise, Promise.resolve(promise).then(result => ({ result, promise }))); + + // If we hit the concurrency limit, wait for one to resolve + if (activeFutures.size >= limit) { + const { result: finished, promise } = await Promise.race(activeFutures.values()) + activeFutures.delete(promise); + finalResult = finished; + yield finished; + } + } + + // After the main loop, yield any remaining futures + for await (const remainingFuture of activeFutures.values()) { + finalResult = remainingFuture.result; + activeFutures.delete(remainingFuture.promise); + yield finalResult; + } + + return finalResult; + }); +} \ No newline at end of file diff --git a/deadline.ts b/deadline.ts new file mode 100644 index 0000000..72d20f5 --- /dev/null +++ b/deadline.ts @@ -0,0 +1,35 @@ +import { Future } from "./future.ts"; + +/** + * Sets a deadline for a future, canceling it if it takes longer than the specified time to complete. + * If the future resolves before the deadline, the timeout is cleared. + * + * @param future - The future to set a deadline for. + * @param ms - The time in milliseconds before the future is canceled. + * @returns A future that will be canceled if it exceeds the specified time. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * return 100; + * }); + * + * const deadlineFuture = Future.withDeadline(future, 1000); // Sets a 1-second deadline + * ``` + */ +export function withDeadline(future: Future, ms: number): Future { + return new Future(async function* () { + const { promise: timeout, reject } = Promise.withResolvers(); + + const timeoutId = setTimeout(() => { + future.cancel(new Error("Future timed out")); + reject(new Error("Future timed out")); + }, ms); + + const result = await Promise.race([future.toPromise(), timeout]); + clearTimeout(timeoutId); // Clear the timeout if the future resolves in time + + yield result as T | TReturn; + return result as T | TReturn; + }); +} \ No newline at end of file diff --git a/errors.ts b/errors.ts new file mode 100644 index 0000000..eed5218 --- /dev/null +++ b/errors.ts @@ -0,0 +1,5 @@ +export class CancellationError extends Error { + constructor() { + super("Future was canceled"); + } +} \ No newline at end of file diff --git a/from.ts b/from.ts new file mode 100644 index 0000000..f2fbdbf --- /dev/null +++ b/from.ts @@ -0,0 +1,294 @@ +import type { FutureOperation } from "./future.ts"; + +import { isPromiseLike, isAsyncIterator, isIterator, isAsyncIterable, isIterable, isBuiltinIterable, isAsyncGenerator, isGenerator } from "./_utils.ts"; +import { Future } from "./future.ts"; + +export interface FutureFromOperation { + (abort: AbortController): ReturnType> | PromiseLike | T +} + +/** + * Creates a `Future` from an operation, such as an async generator, a promise-like object, + * or any kind of iterable. This method converts different types of async tasks into a Future. + * + * ### What Does `from` Do? + * + * It takes an operation (which could be a promise, an iterator, an iterable, etc.) and wraps it + * inside a `Future`, making it possible to control its execution with pause/resume/cancel functionalities. + * + * ### Supported Types: + * + * - **PromiseLike**: Handles promise-based operations that resolve asynchronously. + * - **AsyncIterable/Iterable**: Supports both async and sync iterables (like arrays or streams). + * - **AsyncGenerator/Generator**: Handles both async and sync generators. + * + * @param operation The operation to convert into a Future. It could be a promise-like object, + * a generator, an async generator, an iterator, or an iterable. + * + * @returns A future representing the given operation. + * + * ### How It Works: + * + * **Push-Based**: + * - This is the traditional workflow where the generator (or operation) autonomously pushes values to the consumer. + * - In this scenario, the generator continues yielding values until it is complete. + * + * @example Push-Based Workflow: + * ```typescript + * const future = Future.from(async function* () { + * yield 1; + * yield 2; + * yield 3; + * return 4; + * }); + * + * for await (const value of future) { + * console.log(value); // Logs 1, 2, 3 + * } + * ``` + * + * **Pull-Based**: + * - In this more advanced workflow, the generator waits for the consumer to "pull" the next value. + * - Each time the consumer calls `next()`, a value is passed into the generator, resuming its execution. + * + * @example Pull-Based Workflow: + * ```typescript + * const future = Future.from(async function* () { + * let result = { value: 1, done: false }; + * + * while (!result.done) { + * result = await (yield result.value); // Wait for input from the consumer + * } + * + * return result.value; + * }); + * + * const iterator = future[Symbol.asyncIterator](); + * console.log(await iterator.next()); // { value: 1, done: false } + * console.log(await iterator.next({ value: 2, done: false })); // { value: 2, done: false } + * console.log(await iterator.next({ value: 3, done: true })); // { value: 3, done: true } + * ``` + * + * ### Push vs. Pull: + * - **Push-Based**: The generator automatically pushes values without waiting for any input. + * - **Pull-Based**: The generator waits for input before it can produce the next value. + * + * ### Explanation of Iterators and Iterables: + * - **Iterator**: An object that represents a sequence of values. It has a `.next()` method that returns the next value in the sequence. + * - **Iterable**: An object that implements the `Symbol.iterator` method, returning an iterator. + * - **Async Iterators**: Similar to iterators but work with `Promise` objects and use `for await...of` loops for asynchronous iteration. + * + * ### Handling Different Types: + * + * @example Handling a simple promise-like operation: + * ```typescript + * const future = Future.from(Promise.resolve(42)); + * const result = await future.toPromise(); // result is 42 + * ``` + * + * @example Handling a synchronous iterable (like an array): + * ```typescript + * const future = Future.from([1, 2, 3]); + * for await (const value of future) { + * console.log(value); // Logs 1, 2, and 3 + * } + * ``` + * + * @example Handling an async generator: + * ```typescript + * const future = Future.from(async function* () { + * yield 1; + * yield 2; + * yield 3; + * return 4; + * }); + * + * const result = await future.toPromise(); // result is 4 + * ``` + * + * @example Handling an async generator with a pull-based workflow: + * ```typescript + * const future = Future.from(async function* (abort: AbortController) { + * const initialResult = { value: 1, done: false }; + * let result = initialResult; + * + * while (!result.done) { + * result = await (yield result.value); // Pull-based: wait for external input + * } + * return result.value; + * }); + * + * const iterator = future[Symbol.asyncIterator](); + * console.log(await iterator.next()); // { value: 1, done: false } + * console.log(await iterator.next({ value: 2, done: false })); // { value: 2, done: false } + * console.log(await iterator.next({ value: 3, done: true })); // { value: 3, done: true } + * ``` + * + */ +export function from(operation: PromiseLike): ReturnType>; +export function from(operation: ReadableStream): ReturnType>; +export function from(operation: AsyncIterable | Iterable>): ReturnType>; +export function from(operation: Iterable>): ReturnType>; +export function from(operation: AsyncIterator | Iterator, TReturn | PromiseLike, TNext>): ReturnType>; +export function from(operation: FutureFromOperation): ReturnType>; +export function from(operation: Future): Future; +export function from(operation: T): ReturnType>; +export function from( + operation: + | FutureFromOperation + | Future + | ReadableStream + | AsyncIterable + | Iterable> + | Iterator, TReturn | PromiseLike, TNext> + | PromiseLike + | T +) { + // Handle Future instances directly + if (operation instanceof Future) { + return operation; + } + + // Handle ReadableStreams (common in web APIs) + if (operation instanceof ReadableStream) { + return fromStream(operation); + } + + if (isPromiseLike(operation)) { + return fromPromise(operation); + } + + if (isAsyncIterator(operation) || isIterator(operation)) { + return fromIterator(operation); + } + + // We skip arrays and strings as they are built-in iterables, but never return a value directly, so we await them. + if ((isAsyncIterable(operation) || isIterable(operation)) && !isBuiltinIterable(operation)) { + return fromIterable(operation); + } + + if (isBuiltinIterable(operation)) { + return fromBuiltinIterable(operation); + } + + if (typeof operation === "function") { + return fromOperation(operation as FutureFromOperation); + } + + return of(operation); +} + +export function of(value: T): Future { + return new Future(async function* () { + yield value; + return value; + }); +} + +export function fromPromise(promise: PromiseLike): Future { + return new Future(async function* () { + yield promise; + return promise; + }); +} + +export function fromOperation(operation: FutureFromOperation) { + return new Future(async function* (abort: AbortController) { + const result = operation(abort); + + if (isAsyncGenerator(result) || isGenerator(result)) { + // Handle async iterable or iterable result + return yield* result; + } + + if (isBuiltinIterable(result)) { + // Handle built-in iterable result + yield* result as Iterable>; + return result as TReturn; + } + + // Handle promise-like result or single value + yield result; + return result as TReturn; + }); +} + +export function fromIterable(iterable: AsyncIterable | Iterable>): Future { + return new Future(async function* () { + return yield* iterable; + }); +} + +export function fromBuiltinIterable(iterable: Iterable>): Future>> { + return new Future>>(async function* () { + yield* iterable; + return iterable; + }); +} + +export function fromIterator(iterator: AsyncIterator | Iterator, TReturn | PromiseLike, TNext>): Future { + return new Future(async function* () { + let iteratorResult = await iterator.next(); + while (!iteratorResult.done) { + iteratorResult = await iterator.next(yield iteratorResult.value); + } + + return iteratorResult.value; + }); +} + +/** + * Creates a `Future` from a readable stream. + * + * This method allows you to process data from a `ReadableStream` as it becomes available, + * yielding each chunk of data and providing full control over the stream's lifecycle. + * + * @param stream - The `ReadableStream` to convert into a `Future`. + * @returns A `Future` that yields chunks of data from the stream. + * + * @example + * ```typescript + * const response = await fetch('https://api.example.com/large-file'); + * const future = Future.fromStream(response.body!); + * + * for await (const chunk of future) { + * console.log(chunk); // Process each chunk of data + * } + * ``` + * + * @example + * ```typescript + * const stream = new ReadableStream({ + * pull(controller) { + * controller.enqueue(new Uint8Array([1, 2, 3])); + * controller.close(); + * } + * }); + * const future = Future.fromStream(stream); + * for await (const chunk of future) { + * console.log(chunk); // Logs Uint8Array([1, 2, 3]) + * } + * ``` + */ +export function fromStream(stream: ReadableStream): Future { + return new Future(async function* () { + const reader = stream.getReader(); + + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + // Yield each chunk of data + yield value; + } + } catch (error) { + reader.cancel(error); + throw error; + } finally { + reader.releaseLock(); + } + + return undefined; + }); +} \ No newline at end of file diff --git a/future.ts b/future.ts new file mode 100644 index 0000000..28e151a --- /dev/null +++ b/future.ts @@ -0,0 +1,508 @@ +import { CancellationError } from "./errors.ts"; + +export const Status = { + Idle: 0, + Running: 1, + Paused: 2, + Completed: 3, + Cancelled: 4, + Destroyed: -1, +} + +export type StatusEnum = typeof Status[keyof typeof Status]; + +export interface FutureOperation { + (abort: AbortController): AsyncGenerator | Generator +} + +/** + * The `Future` class is a more powerful and flexible alternative to JavaScript's native `Promise`, + * designed to address several weaknesses of Promises such as lack of cancellation, limited concurrency control, and + * inability to pause/resume tasks. `Future` instances are cancellable, pausable, compositional, and structured for + * concurrency management. + * + * ## Key Features + * + * - **Pausable**: Execution can be paused and resumed at will, giving precise control over flow. + * - **Cancellable**: Futures can be cancelled at any point. + * - **Background Execution**: You can prepare Futures for background execution with the `inBackground` method. + * - **Compositional**: Chain, sequence, and compose multiple futures with ease. + * - **Advanced Concurrency Control**: Supports throttling, limiting concurrency, and structured concurrency. + * + * ## Usage Examples + * + * ### Simple Future Creation + * ```typescript + * const future = Future.from(async function* () { + * yield 42; // Pauses and returns 42 + * return 100; // Completes and returns 100 + * }); + * + * console.log(await future.toPromise()); // Logs 100 + * ``` + * + * ### Pausing and Resuming + * ```typescript + * const future = Future.from(async function* () { + * yield 1; + * yield 2; + * return 3; + * }); + * + * future.pause(); + * setTimeout(() => future.resume(), 1000); + * + * for await (const value of future) { + * console.log(value); // Logs 1, 2, then 3 + * } + * ``` + * + * ### Running in the Background + * ```typescript + * const backgroundFuture = Future.inBackground(Future.from(async function* () { + * yield 1; + * return 2; + * })); + * + * console.log(await backgroundFuture.toPromise()); // Executes in idle time, returns 2 + * ``` + * + * @template T - The type of the value that the future will resolve to. + */ +export class Future + // @ts-ignore Iterator is defined but typescript doesn't recognize it yet + extends globalThis.Iterator + implements PromiseLike { + #generator?: ReturnType> | null; + #operation?: FutureOperation | null; + #status: StatusEnum = Status.Idle; + #abort: AbortController | null = new AbortController(); + readonly #resolvers = { + pause: Promise.withResolvers(), + complete: Promise.withResolvers(), + abort: Promise.withResolvers(), + } + + static #EventHandler = class PrivateEventHandler { + #delegate: Future | undefined | null; + constructor(delegate?: Future) { + this.#delegate = delegate; + } + + handleEvent(event: Event) { + if (this.#delegate) { + this.#delegate.#handleEvent(event); + } + } + + destroy() { + this.#delegate = null; + } + } + + #eventhandler = new Future.#EventHandler(this); + + /** + * Creates a new `Future` instance. + * @param operation - A function that returns an async generator to define the asynchronous task. + */ + constructor(operation: FutureOperation) { + super(); + this.#operation = operation; + this.#setup(); + + this.#generator = this.#operation?.(this.#abort!); + } + + #setup() { + this.#abort?.signal?.removeEventListener?.("abort", this.#eventhandler); + + this.#abort = new AbortController(); + Object.assign(this.#resolvers, { + pause: Promise.withResolvers(), + complete: Promise.withResolvers(), + abort: Promise.withResolvers(), + }); + + this.#abort?.signal?.addEventListener?.("abort", this.#eventhandler); + this.#resolvers.complete?.promise?.finally?.(() => (this.#status = Status.Completed)); + + const rotatingResolver = () => { + this.#status = Status.Running; + this.#resolvers.pause = Promise.withResolvers(); + this.#resolvers.pause?.promise?.then?.(rotatingResolver); + } + + this.#resolvers.pause?.promise?.then?.(rotatingResolver); + this.#status = Status.Idle; + } + + #handleEvent(event: Event) { + if (event.type === "abort") { + this.#status = Status.Cancelled; + this.#resolvers.abort?.resolve?.(this.#reason); + this.#resolvers?.pause?.reject?.(this.#reason); + this.#resolvers?.complete?.reject?.(this.#reason); + } + } + + get #reason() { + return this.#abort?.signal?.reason; + } + + isCancelled() { + return this.#status === Status.Cancelled; + } + + isPaused() { + return this.#status === Status.Paused; + } + + isComplete() { + return this.#status === Status.Completed; + } + + isRunning() { + return this.#status === Status.Running; + } + + isDestroyed() { + return this.#status === Status.Destroyed; + } + + isIdle() { + return this.#status === Status.Idle; + } + + paused() { + return this.#resolvers.pause?.promise; + } + + completed() { + return this.#resolvers.complete?.promise; + } + + cancelled() { + return this.#resolvers.abort?.promise; + } + + complete(value: TReturn) { + // If the generator does not exist, resolve immediately with the provided value. + this.#resolvers.complete.resolve(value); + return this; + } + + /** + * Cancels the future, preventing further execution. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * return 100; + * }); + * future.cancel(); // Aborts future execution + * ``` + */ + cancel(reason: unknown = new CancellationError()) { + this.#abort?.abort?.(reason); + return this; + } + + /** + * Pauses the execution of the future. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * return 100; + * }); + * future.pause(); // Pauses future execution + * ``` + */ + pause() { + if (!this.isPaused()) { + this.#status = Status.Paused; + } + return this; + } + + /** + * Resumes a paused future. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * return 100; + * }); + * future.resume(); // Resumes future execution + * ``` + */ + resume() { + if (this.isPaused()) { + this.#resolvers?.pause?.resolve?.(); + } + return this; + } + + /** + * Resets the future for re-execution, allowing it to run from the beginning. + * This can only be done if the future is complete. + * @throws Error if the future is not complete. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * return 100; + * }); + * await future.toPromise(); // Completes the future + * future.reset(); // Resets the future for reuse + * ``` + */ + reset() { + if (this.isDestroyed()) { + throw new Error("Cannot reset a destroyed future"); + } + + if (!this.isComplete()) { + throw new Error("Cannot reset an incomplete future"); + } + + this.#setup(); + this.#generator = this.#operation?.(this.#abort!); + return this; + } + + destroy() { + this.cancel(); + + this.#abort?.signal?.removeEventListener?.("abort", this.#eventhandler); + this.#eventhandler?.destroy?.(); + + // @ts-ignore Resetting private properties + this.#eventhandler = null as unknown; + + Object.assign(this.#resolvers, { + pause: null, + complete: null, + abort: null, + }); + + this.#abort = null; + this.#generator = null; + this.#operation = null; + + this.#status = Status.Destroyed; + } + + /** + * Implements the async iterator protocol, allowing futures to be used in `for await...of` loops. + * + * This method is designed to support both **push-based** and **pull-based** workflows, + * where values can either be automatically yielded by the generator or "pulled" from it via external input. + * + * ### Key Concepts to Understand: + * + * **Iterators and Iterables**: + * - An **iterator** is an object that defines a sequence of values, typically with a `.next()` method. + * - An **iterable** is an object that implements the `Symbol.iterator` method, returning an iterator. + * - **Async iterators** are similar to iterators but involve asynchronous operations (using `Promise`). + * - **for await...of loops** allow you to consume async iterators just like regular iterators but in an asynchronous context. + * + * **next() and yield**: + * - The `yield` keyword pauses a generator and returns a value to the outside world. + * - The `next()` method resumes the generator and can **both send a value in** and **receive the next yielded value**. + * - The confusing part: `yield` can be the value that `next()` returns, and the input passed into `next()` can be used as the value of the last `yield`. + * + * In essence, each call to `next()` resumes the generator from where it last left off, and the value passed to `next()` can be accessed within the generator. + * + * ### Push-Based Workflow: + * The generator automatically yields values in a push-based workflow, so the consumer doesn't need to send any input. + * + * @example Push-Based Workflow + * ```typescript + * // Push-based async generator example + * const future = Future.from(async function* () { + * yield 1; + * yield 2; + * yield 3; + * return 4; + * }); + * + * for await (const value of future) { + * console.log(value); // Logs 1, 2, 3 + * } + * ``` + * + * ### Pull-Based Workflow: + * In a pull-based workflow, the generator waits for input via `next()`. The generator will only proceed when `next()` is called. + * + * @example Pull-Based Workflow + * ```typescript + * // Pull-based async generator example + * const future = Future.from(async function* () { + * let result = { value: 1, done: false }; + * + * // Wait for external input before proceeding + * while (!result.done) { + * result = await (yield result.value); + * } + * + * return result.value; + * }); + * + * const iterator = future[Symbol.asyncIterator](); + * console.log(await iterator.next()); // { value: 1, done: false } + * console.log(await iterator.next({ value: 2, done: false })); // { value: 2, done: false } + * console.log(await iterator.next({ value: 3, done: true })); // { value: 3, done: true } + * ``` + * + * ### Breakdown of the Code: + * + * **Push-Based**: + * - The generator automatically yields values, and the consumer just needs to await those values. + * - Example: A generator yielding values without expecting input. + * + * **Pull-Based**: + * - The generator waits for input from the consumer using the `yield` keyword. + * - The consumer controls when the next value is processed by passing in data via `next()`. + * + * ### Advanced Concepts: + * + * - **Priming the Generator**: + * - We prime the generator by calling `next()` once before the loop starts. This ensures the generator is ready to receive input. + * - **Handling Pauses and Cancellation**: + * - If the future is paused, the generator will wait until it's resumed. + * - If the future is canceled, an error will be thrown. + * + * @yields The values generated by the future. + */ + async *[Symbol.asyncIterator](): AsyncGenerator { + let err: unknown; + let result: IteratorResult | undefined; + try { + if (!this.#generator || typeof this.#generator?.next !== "function") { + throw new Error("generator not defined"); + } + + if (this.isComplete()) { + const value = (await this.#resolvers.complete?.promise) ?? result?.value; + const finished = await this.#generator?.return?.(value as unknown as TReturn); + if (finished) return finished?.value; + } + + // Prime the generator by starting the iteration process + result = await this.#generator?.next?.(); + + // Continue yielding values until the generator completes + while (!result?.done) { + if (this.isCancelled()) { + await this.#generator?.throw?.(this.#reason); + throw this.#reason; + } + + // Handle pausing by awaiting the pause resolver + if (this.isPaused()) { + await this.#resolvers.pause?.promise; + continue; + } + + if (this.isComplete()) { + const value = (await this.#resolvers.complete?.promise) ?? result?.value; + const finished = await this.#generator?.return?.(value as unknown as TReturn); + if (finished) return finished?.value; + break; + } + + // Set status to running + if (this.#status !== Status.Running) { + this.#status = Status.Running; + } + + // Yield the value to the consumer and wait for the next input + result = await this.#generator?.next?.(yield result?.value); + } + + // Return the final value once iteration completes + return result?.value; + } catch (error) { + // Handle errors during iteration + if (!this.isCancelled()) this.#abort?.abort?.(error); + throw (err = error); + } finally { + // Resolve or reject based on the completion state + if (err) this.#resolvers.complete?.reject?.(err); + else this.#resolvers.complete?.resolve?.(result?.value); + } + } + + next(...args: [] | [TNext]): PromiseLike> { + return this[Symbol.asyncIterator]()?.next?.(...args); + } + + [Symbol.dispose]() { + this.destroy(); + } + + /** + * Converts the future into a promise and waits for its resolution. + * Includes support for control flow operations including pausing, cancelling, and more... + * + * @returns A promise that resolves to the final result of the future. + * If the generator does not have an explicit return, it will return undefined. + * @example + * ```typescript + * const future = Future.from(async function* () { + * yield 42; + * // No explicit return, so the final result will be undefined + * }); + * const result = await future.toPromise(); // result is undefined + * ``` + */ + async toPromise(): Promise { + let result: IteratorResult; + const generator = this[Symbol.asyncIterator](); + + // Iterate through the generator until completion + do { + result = await generator.next(); + } while (!result.done); + + // If the generator has no explicit return, result.value will be undefined + return result.value; + } + + /** + * `then` method to allow Future to be used with `await` and promise chaining. + * @param onfulfilled Called when the future resolves successfully. + * @param onrejected Called when the future is rejected. + * @returns A promise that resolves with the result of the future. + */ + then( + onfulfilled?: ((value: T | TReturn) => TResult1 | PromiseLike) | undefined | null, + onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | undefined | null + ): Promise { + return this.toPromise().then(onfulfilled, onrejected); + } + + /** + * `catch` method to handle rejections. + * @param onrejected A callback that handles the rejection reason. + * @returns A promise that resolves or rejects based on the future's outcome. + */ + catch( + onrejected?: ((reason: unknown) => TResult | PromiseLike) | undefined | null + ): Promise { + return this.toPromise().catch(onrejected); + } + + /** + * `finally` method to allow adding a cleanup step after the future resolves or rejects. + * @param onfinally Called when the future is complete. + * @returns A promise that resolves to the final result. + */ + finally(onfinally?: (() => void) | undefined | null): Promise { + return this.toPromise().finally(onfinally); + } +} + +export default Future; \ No newline at end of file diff --git a/mod.ts b/mod.ts index 4b214b5..b157a30 100644 --- a/mod.ts +++ b/mod.ts @@ -1,1187 +1,8 @@ -import { cancelIdle, idle } from "./idle.ts"; -import { isAsyncIterator, isIterator, isAsyncGenerator, isGenerator, isAsyncIterable, isIterable, isPromiseLike, isBuiltinIterable } from "./utils.ts"; - -export class CancellationError extends Error { - constructor() { - super("Future was canceled"); - } -} - -export const Status = { - Idle: 0, - Running: 1, - Paused: 2, - Completed: 3, - Cancelled: 4, - Destroyed: -1, -} - -export type StatusEnum = typeof Status[keyof typeof Status]; - -/** - * The `Future` class is a more powerful and flexible alternative to JavaScript's native `Promise`, - * designed to address several weaknesses of Promises such as lack of cancellation, limited concurrency control, and - * inability to pause/resume tasks. `Future` instances are cancellable, pausable, compositional, and structured for - * concurrency management. - * - * ## Key Features - * - * - **Pausable**: Execution can be paused and resumed at will, giving precise control over flow. - * - **Cancellable**: Futures can be cancelled at any point. - * - **Background Execution**: You can prepare Futures for background execution with the `inBackground` method. - * - **Compositional**: Chain, sequence, and compose multiple futures with ease. - * - **Advanced Concurrency Control**: Supports throttling, limiting concurrency, and structured concurrency. - * - * ## Usage Examples - * - * ### Simple Future Creation - * ```typescript - * const future = Future.from(async function* () { - * yield 42; // Pauses and returns 42 - * return 100; // Completes and returns 100 - * }); - * - * console.log(await future.toPromise()); // Logs 100 - * ``` - * - * ### Pausing and Resuming - * ```typescript - * const future = Future.from(async function* () { - * yield 1; - * yield 2; - * return 3; - * }); - * - * future.pause(); - * setTimeout(() => future.resume(), 1000); - * - * for await (const value of future) { - * console.log(value); // Logs 1, 2, then 3 - * } - * ``` - * - * ### Running in the Background - * ```typescript - * const backgroundFuture = Future.inBackground(Future.from(async function* () { - * yield 1; - * return 2; - * })); - * - * console.log(await backgroundFuture.toPromise()); // Executes in idle time, returns 2 - * ``` - * - * @template T - The type of the value that the future will resolve to. - */ -export class Future - // @ts-ignore Iterator is defined but ts doesn't recognize it yet - extends globalThis.Iterator - implements PromiseLike { - #generator?: ReturnType> | null; - #operation?: FutureOperation | null; - #status: StatusEnum = Status.Idle; - #abort: AbortController | null = new AbortController(); - readonly #resolvers = { - pause: Promise.withResolvers(), - complete: Promise.withResolvers(), - abort: Promise.withResolvers(), - } - - static #EventHandler = class PrivateEventHandler { - #delegate: Future | undefined | null; - constructor(delegate?: Future) { - this.#delegate = delegate; - } - - handleEvent(event: Event) { - if (this.#delegate) { - this.#delegate.#handleEvent(event); - } - } - - destroy() { - this.#delegate = null; - } - } - - #eventhandler = new Future.#EventHandler(this); - - /** - * Creates a new `Future` instance. - * @param operation - A function that returns an async generator to define the asynchronous task. - */ - constructor(operation: FutureOperation) { - super(); - this.#operation = operation; - this.#setup(); - - this.#generator = this.#operation?.(this.#abort!); - } - - #setup() { - this.#abort?.signal?.removeEventListener?.("abort", this.#eventhandler); - - this.#abort = new AbortController(); - Object.assign(this.#resolvers, { - pause: Promise.withResolvers(), - complete: Promise.withResolvers(), - abort: Promise.withResolvers(), - }); - - this.#abort?.signal?.addEventListener?.("abort", this.#eventhandler); - this.#resolvers.complete?.promise?.finally?.(() => (this.#status = Status.Completed)); - - const rotatingResolver = () => { - this.#status = Status.Running; - this.#resolvers.pause = Promise.withResolvers(); - this.#resolvers.pause?.promise?.then?.(rotatingResolver); - } - - this.#resolvers.pause?.promise?.then?.(rotatingResolver); - this.#status = Status.Idle; - } - - #handleEvent(event: Event) { - if (event.type === "abort") { - this.#status = Status.Cancelled; - this.#resolvers.abort?.resolve?.(this.#reason); - this.#resolvers?.pause?.reject?.(this.#reason); - this.#resolvers?.complete?.reject?.(this.#reason); - } - } - - get #reason() { - return this.#abort?.signal?.reason; - } - - isCancelled() { - return this.#status === Status.Cancelled; - } - - isPaused() { - return this.#status === Status.Paused; - } - - isComplete() { - return this.#status === Status.Completed; - } - - isRunning() { - return this.#status === Status.Running; - } - - isDestroyed() { - return this.#status === Status.Destroyed; - } - - isIdle() { - return this.#status === Status.Idle; - } - - paused() { - return this.#resolvers.pause?.promise; - } - - completed() { - return this.#resolvers.complete?.promise; - } - - cancelled() { - return this.#resolvers.abort?.promise; - } - - complete(value: TReturn) { - // If the generator does not exist, resolve immediately with the provided value. - this.#resolvers.complete.resolve(value); - return this; - } - - /** - * Cancels the future, preventing further execution. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * return 100; - * }); - * future.cancel(); // Aborts future execution - * ``` - */ - cancel(reason: unknown = new CancellationError()) { - this.#abort?.abort?.(reason); - return this; - } - - /** - * Pauses the execution of the future. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * return 100; - * }); - * future.pause(); // Pauses future execution - * ``` - */ - pause() { - if (!this.isPaused()) { - this.#status = Status.Paused; - } - return this; - } - - /** - * Resumes a paused future. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * return 100; - * }); - * future.resume(); // Resumes future execution - * ``` - */ - resume() { - if (this.isPaused()) { - this.#resolvers?.pause?.resolve?.(); - } - return this; - } - - /** - * Resets the future for re-execution, allowing it to run from the beginning. - * This can only be done if the future is complete. - * @throws Error if the future is not complete. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * return 100; - * }); - * await future.toPromise(); // Completes the future - * future.reset(); // Resets the future for reuse - * ``` - */ - reset() { - if (this.isDestroyed()) { - throw new Error("Cannot reset a destroyed future"); - } - - if (!this.isComplete()) { - throw new Error("Cannot reset an incomplete future"); - } - - this.#setup(); - this.#generator = this.#operation?.(this.#abort!); - return this; - } - - destroy() { - this.cancel(); - - this.#abort?.signal?.removeEventListener?.("abort", this.#eventhandler); - this.#eventhandler?.destroy?.(); - - // @ts-ignore Resetting private properties - this.#eventhandler = null as unknown; - - Object.assign(this.#resolvers, { - pause: null, - complete: null, - abort: null, - }); - - this.#abort = null; - this.#generator = null; - this.#operation = null; - - this.#status = Status.Destroyed; - } - - /** - * Implements the async iterator protocol, allowing futures to be used in `for await...of` loops. - * - * This method is designed to support both **push-based** and **pull-based** workflows, - * where values can either be automatically yielded by the generator or "pulled" from it via external input. - * - * ### Key Concepts to Understand: - * - * **Iterators and Iterables**: - * - An **iterator** is an object that defines a sequence of values, typically with a `.next()` method. - * - An **iterable** is an object that implements the `Symbol.iterator` method, returning an iterator. - * - **Async iterators** are similar to iterators but involve asynchronous operations (using `Promise`). - * - **for await...of loops** allow you to consume async iterators just like regular iterators but in an asynchronous context. - * - * **next() and yield**: - * - The `yield` keyword pauses a generator and returns a value to the outside world. - * - The `next()` method resumes the generator and can **both send a value in** and **receive the next yielded value**. - * - The confusing part: `yield` can be the value that `next()` returns, and the input passed into `next()` can be used as the value of the last `yield`. - * - * In essence, each call to `next()` resumes the generator from where it last left off, and the value passed to `next()` can be accessed within the generator. - * - * ### Push-Based Workflow: - * The generator automatically yields values in a push-based workflow, so the consumer doesn't need to send any input. - * - * @example Push-Based Workflow - * ```typescript - * // Push-based async generator example - * const future = Future.from(async function* () { - * yield 1; - * yield 2; - * yield 3; - * return 4; - * }); - * - * for await (const value of future) { - * console.log(value); // Logs 1, 2, 3 - * } - * ``` - * - * ### Pull-Based Workflow: - * In a pull-based workflow, the generator waits for input via `next()`. The generator will only proceed when `next()` is called. - * - * @example Pull-Based Workflow - * ```typescript - * // Pull-based async generator example - * const future = Future.from(async function* () { - * let result = { value: 1, done: false }; - * - * // Wait for external input before proceeding - * while (!result.done) { - * result = await (yield result.value); - * } - * - * return result.value; - * }); - * - * const iterator = future[Symbol.asyncIterator](); - * console.log(await iterator.next()); // { value: 1, done: false } - * console.log(await iterator.next({ value: 2, done: false })); // { value: 2, done: false } - * console.log(await iterator.next({ value: 3, done: true })); // { value: 3, done: true } - * ``` - * - * ### Breakdown of the Code: - * - * **Push-Based**: - * - The generator automatically yields values, and the consumer just needs to await those values. - * - Example: A generator yielding values without expecting input. - * - * **Pull-Based**: - * - The generator waits for input from the consumer using the `yield` keyword. - * - The consumer controls when the next value is processed by passing in data via `next()`. - * - * ### Advanced Concepts: - * - * - **Priming the Generator**: - * - We prime the generator by calling `next()` once before the loop starts. This ensures the generator is ready to receive input. - * - **Handling Pauses and Cancellation**: - * - If the future is paused, the generator will wait until it's resumed. - * - If the future is canceled, an error will be thrown. - * - * @yields The values generated by the future. - */ - async *[Symbol.asyncIterator](): AsyncGenerator { - let err: unknown; - let result: IteratorResult | undefined; - try { - if (!this.#generator || typeof this.#generator?.next !== "function") { - throw new Error("generator not defined"); - } - - if (this.isComplete()) { - const value = (await this.#resolvers.complete?.promise) ?? result?.value; - const finished = await this.#generator?.return?.(value as unknown as TReturn); - if (finished) return finished?.value; - } - - // Prime the generator by starting the iteration process - result = await this.#generator?.next?.(); - - // Continue yielding values until the generator completes - while (!result?.done) { - if (this.isCancelled()) { - await this.#generator?.throw?.(this.#reason); - throw this.#reason; - } - - // Handle pausing by awaiting the pause resolver - if (this.isPaused()) { - await this.#resolvers.pause?.promise; - continue; - } - - if (this.isComplete()) { - const value = (await this.#resolvers.complete?.promise) ?? result?.value; - const finished = await this.#generator?.return?.(value as unknown as TReturn); - if (finished) return finished?.value; - break; - } - - // Set status to running - if (this.#status !== Status.Running) { - this.#status = Status.Running; - } - - // Yield the value to the consumer and wait for the next input - result = await this.#generator?.next?.(yield result?.value); - } - - // Return the final value once iteration completes - return result?.value; - } catch (error) { - // Handle errors during iteration - if (!this.isCancelled()) this.#abort?.abort?.(error); - throw (err = error); - } finally { - // Resolve or reject based on the completion state - if (err) this.#resolvers.complete?.reject?.(err); - else this.#resolvers.complete?.resolve?.(result?.value); - } - } - - next(...args: [] | [TNext]): PromiseLike> { - return this[Symbol.asyncIterator]()?.next?.(...args); - } - - [Symbol.dispose]() { - this.destroy(); - } - - /** - * Converts the future into a promise and waits for its resolution. - * Includes support for control flow operations including pausing, cancelling, and more... - * - * @returns A promise that resolves to the final result of the future. - * If the generator does not have an explicit return, it will return undefined. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * // No explicit return, so the final result will be undefined - * }); - * const result = await future.toPromise(); // result is undefined - * ``` - */ - async toPromise(): Promise { - let result: IteratorResult; - const generator = this[Symbol.asyncIterator](); - - // Iterate through the generator until completion - do { - result = await generator.next(); - } while (!result.done); - - // If the generator has no explicit return, result.value will be undefined - return result.value; - } - - /** - * `then` method to allow Future to be used with `await` and promise chaining. - * @param onfulfilled Called when the future resolves successfully. - * @param onrejected Called when the future is rejected. - * @returns A promise that resolves with the result of the future. - */ - then( - onfulfilled?: ((value: T | TReturn) => TResult1 | PromiseLike) | undefined | null, - onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | undefined | null - ): Promise { - return this.toPromise().then(onfulfilled, onrejected); - } - - /** - * `catch` method to handle rejections. - * @param onrejected A callback that handles the rejection reason. - * @returns A promise that resolves or rejects based on the future's outcome. - */ - catch( - onrejected?: ((reason: unknown) => TResult | PromiseLike) | undefined | null - ): Promise { - return this.toPromise().catch(onrejected); - } - - /** - * `finally` method to allow adding a cleanup step after the future resolves or rejects. - * @param onfinally Called when the future is complete. - * @returns A promise that resolves to the final result. - */ - finally(onfinally?: (() => void) | undefined | null): Promise { - return this.toPromise().finally(onfinally); - } - - /** - * Creates a `Future` from an operation, such as an async generator, a promise-like object, - * or any kind of iterable. This method converts different types of async tasks into a Future. - * - * ### What Does `from` Do? - * - * It takes an operation (which could be a promise, an iterator, an iterable, etc.) and wraps it - * inside a `Future`, making it possible to control its execution with pause/resume/cancel functionalities. - * - * ### Supported Types: - * - * - **PromiseLike**: Handles promise-based operations that resolve asynchronously. - * - **AsyncIterable/Iterable**: Supports both async and sync iterables (like arrays or streams). - * - **AsyncGenerator/Generator**: Handles both async and sync generators. - * - * @param operation The operation to convert into a Future. It could be a promise-like object, - * a generator, an async generator, an iterator, or an iterable. - * - * @returns A future representing the given operation. - * - * ### How It Works: - * - * **Push-Based**: - * - This is the traditional workflow where the generator (or operation) autonomously pushes values to the consumer. - * - In this scenario, the generator continues yielding values until it is complete. - * - * @example Push-Based Workflow: - * ```typescript - * const future = Future.from(async function* () { - * yield 1; - * yield 2; - * yield 3; - * return 4; - * }); - * - * for await (const value of future) { - * console.log(value); // Logs 1, 2, 3 - * } - * ``` - * - * **Pull-Based**: - * - In this more advanced workflow, the generator waits for the consumer to "pull" the next value. - * - Each time the consumer calls `next()`, a value is passed into the generator, resuming its execution. - * - * @example Pull-Based Workflow: - * ```typescript - * const future = Future.from(async function* () { - * let result = { value: 1, done: false }; - * - * while (!result.done) { - * result = await (yield result.value); // Wait for input from the consumer - * } - * - * return result.value; - * }); - * - * const iterator = future[Symbol.asyncIterator](); - * console.log(await iterator.next()); // { value: 1, done: false } - * console.log(await iterator.next({ value: 2, done: false })); // { value: 2, done: false } - * console.log(await iterator.next({ value: 3, done: true })); // { value: 3, done: true } - * ``` - * - * ### Push vs. Pull: - * - **Push-Based**: The generator automatically pushes values without waiting for any input. - * - **Pull-Based**: The generator waits for input before it can produce the next value. - * - * ### Explanation of Iterators and Iterables: - * - **Iterator**: An object that represents a sequence of values. It has a `.next()` method that returns the next value in the sequence. - * - **Iterable**: An object that implements the `Symbol.iterator` method, returning an iterator. - * - **Async Iterators**: Similar to iterators but work with `Promise` objects and use `for await...of` loops for asynchronous iteration. - * - * ### Handling Different Types: - * - * @example Handling a simple promise-like operation: - * ```typescript - * const future = Future.from(Promise.resolve(42)); - * const result = await future.toPromise(); // result is 42 - * ``` - * - * @example Handling a synchronous iterable (like an array): - * ```typescript - * const future = Future.from([1, 2, 3]); - * for await (const value of future) { - * console.log(value); // Logs 1, 2, and 3 - * } - * ``` - * - * @example Handling an async generator: - * ```typescript - * const future = Future.from(async function* () { - * yield 1; - * yield 2; - * yield 3; - * return 4; - * }); - * - * const result = await future.toPromise(); // result is 4 - * ``` - * - * @example Handling an async generator with a pull-based workflow: - * ```typescript - * const future = Future.from(async function* (abort: AbortController) { - * const initialResult = { value: 1, done: false }; - * let result = initialResult; - * - * while (!result.done) { - * result = await (yield result.value); // Pull-based: wait for external input - * } - * return result.value; - * }); - * - * const iterator = future[Symbol.asyncIterator](); - * console.log(await iterator.next()); // { value: 1, done: false } - * console.log(await iterator.next({ value: 2, done: false })); // { value: 2, done: false } - * console.log(await iterator.next({ value: 3, done: true })); // { value: 3, done: true } - * ``` - * - */ - static from(operation: PromiseLike): ReturnType>; - static from(operation: ReadableStream): ReturnType>; - static from(operation: AsyncIterable | Iterable>): ReturnType>; - static from(operation: Iterable>): ReturnType>; - static from(operation: AsyncIterator | Iterator, TReturn | PromiseLike, TNext>): ReturnType>; - static from(operation: FutureFromOperation): ReturnType>; - static from(operation: Future): Future; - static from(operation: T): ReturnType>; - static from( - operation: - | FutureFromOperation - | Future - | ReadableStream - | AsyncIterable - | Iterable> - | Iterator, TReturn | PromiseLike, TNext> - | PromiseLike - | T - ) { - // Handle Future instances directly - if (operation instanceof Future) { - return operation; - } - - // Handle ReadableStreams (common in web APIs) - if (operation instanceof ReadableStream) { - return Future.fromStream(operation); - } - - if (isPromiseLike(operation)) { - return Future.fromPromise(operation); - } - - if (isAsyncIterator(operation) || isIterator(operation)) { - return Future.fromIterator(operation); - } - - // We skip arrays and strings as they are built-in iterables, but never return a value directly, so we await them. - if ((isAsyncIterable(operation) || isIterable(operation)) && !isBuiltinIterable(operation)) { - return Future.fromIterable(operation); - } - - if (isBuiltinIterable(operation)) { - return Future.fromBuiltinIterable(operation); - } - - if (typeof operation === "function") { - return Future.fromOperation(operation as FutureFromOperation); - } - - return Future.of(operation); - } - - static of(value: T): Future { - return new Future(async function* () { - yield value; - return value; - }); - } - - static fromPromise(promise: PromiseLike): Future { - return new Future(async function* () { - yield promise; - return promise; - }); - } - - static fromOperation(operation: FutureFromOperation) { - return new Future(async function* (abort: AbortController) { - const result = operation(abort); - - if (isAsyncGenerator(result) || isGenerator(result)) { - // Handle async iterable or iterable result - return yield* result; - } - - if (isBuiltinIterable(result)) { - // Handle built-in iterable result - yield* result as Iterable>; - return result as TReturn; - } - - // Handle promise-like result or single value - yield result; - return result as TReturn; - }); - } - - static fromIterable(iterable: AsyncIterable | Iterable>): Future { - return new Future(async function* () { - return yield* iterable; - }); - } - - static fromBuiltinIterable(iterable: Iterable>): Future>> { - return new Future>>(async function* () { - yield* iterable; - return iterable; - }); - } - - static fromIterator(iterator: AsyncIterator | Iterator, TReturn | PromiseLike, TNext>): Future { - return new Future(async function* () { - let iteratorResult = await iterator.next(); - while (!iteratorResult.done) { - iteratorResult = await iterator.next(yield iteratorResult.value); - } - - return iteratorResult.value; - }); - } - - /** - * Creates a `Future` from a readable stream. - * - * This method allows you to process data from a `ReadableStream` as it becomes available, - * yielding each chunk of data and providing full control over the stream's lifecycle. - * - * @param stream - The `ReadableStream` to convert into a `Future`. - * @returns A `Future` that yields chunks of data from the stream. - * - * @example - * ```typescript - * const response = await fetch('https://api.example.com/large-file'); - * const future = Future.fromStream(response.body!); - * - * for await (const chunk of future) { - * console.log(chunk); // Process each chunk of data - * } - * ``` - * - * @example - * ```typescript - * const stream = new ReadableStream({ - * pull(controller) { - * controller.enqueue(new Uint8Array([1, 2, 3])); - * controller.close(); - * } - * }); - * const future = Future.fromStream(stream); - * for await (const chunk of future) { - * console.log(chunk); // Logs Uint8Array([1, 2, 3]) - * } - * ``` - */ - static fromStream(stream: ReadableStream): Future { - return new Future(async function* () { - const reader = stream.getReader(); - - try { - while (true) { - const { done, value } = await reader.read(); - if (done) break; - - // Yield each chunk of data - yield value; - } - } catch (error) { - reader.cancel(error); - throw error; - } finally { - reader.releaseLock(); - } - - return undefined; - }); - } - - /** - * Runs multiple Futures concurrently, yielding their results as they all complete. - * - * This is similar to `Promise.all` but with support for yielding results in sequence - * once all the futures have been resolved. - * - * @param futures - An iterable of `Future` or `PromiseLike` objects. - * @returns An AsyncIterable yielding each result as they complete. - */ - static all(futures: Iterable | PromiseLike>): Future, Awaited[]> { - return new Future, Awaited[]>(async function* () { - // We trigger all futures at once, using Promise.all to await them concurrently. - const results = await Promise.all(futures); - yield* results; - return results; - }); - } - - /** - * Executes multiple futures concurrently and yields their results or errors as soon as they settle. - * This method works similarly to `Promise.allSettled` but yields results incrementally. - * - * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. - * @returns An AsyncIterable yielding each settled result. - * @example - * ```typescript - * const futureSettled = Future.allSettled([ - * Future.from(async function* () { - * yield 42; - * return 100; - * }), - * Future.from(async function* () { - * yield 10; - * return 20; - * }) - * ]); - * - * for await (const result of futureSettled) { - * if (result.status === "fulfilled") { - * console.log(result.value); // Outputs 100 and 20 - * } else { - * console.error(result.reason); - * } - * } - * ``` - */ - static allSettled( - futures: Iterable | PromiseLike> - ): Future, PromiseSettledResult[]> { - return new Future, PromiseSettledResult[]>(async function* () { - // We trigger all futures at once, using Promise.allSettled to await them concurrently. - const results = await Promise.allSettled(futures); - yield* results; - return results; - }); - } - - /** - * Returns the first settled Future, similar to `Promise.race`. - * - * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. - * @returns An AsyncIterable yielding the first result that resolves. - */ - static race( - futures: Iterable | PromiseLike> - ): Future { - return new Future(async function* () { - const result = Promise.race(futures); - yield result; - return result; - }); - } - - /** - * Executes multiple Futures concurrently, but only returns results for the first `count` Futures. - * - * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. - * @param count - The number of Futures to resolve before yielding results. - * @returns An AsyncIterable yielding the first `count` resolved results. - */ - static some( - futures: Iterable | PromiseLike>, - count: number - ): Future, PromiseSettledResult[]> { - return Future.allSettled( - Array.from(futures).slice(0, count) - ); - } - - /** - * Runs multiple Futures within a defined scope, ensuring that all Futures are executed - * with full control over their execution (e.g., pausing, resuming, and pulling/pushing values). - * - * This method supports `Iterable` and `AsyncIterable` containers of Futures, allowing for - * both push-based and pull-based workflows. - * - * `Future.scope` is designed to handle non-concurrent execution by default, allowing you to - * control the flow of each future one at a time. If concurrency is needed, you can use - * `Future.all`, `Future.some`, or similar methods inside the scope. - * - * ### Push vs. Pull Workflows: - * - * - **Push-Based Workflow**: The generator yields values autonomously, and the consumer simply awaits those values. - * - **Pull-Based Workflow**: The generator waits for external input before proceeding to the next value. - * - * @param futures - An iterable or async iterable of Futures to be run within the scope. - * @returns A Future that yields the results of the contained Futures, controlled by the scope. - * - * ### Example Usage: - * - * ```typescript - * const scopeFuture = Future.scope([ - * Future.from(async function* () { - * yield 42; - * return 100; - * }), - * Future.from(async function* () { - * yield 10; - * return 20; - * }) - * ]); - * - * for await (const result of scopeFuture) { - * console.log(result); // Logs 100, 20 - * } - * - * const pullFuture = Future.scope([ - * Future.from(async function* () { - * let result = { value: 42, done: false }; - * - * while (!result.done) { - * result = await (yield result.value); // Wait for input from the consumer - * } - * - * return result.value; - * }) - * ]); - * - * const iterator = pullFuture[Symbol.asyncIterator](); - * console.log(await iterator.next()); // { value: 42, done: false } - * console.log(await iterator.next({ value: 100, done: true })); // { value: 100, done: true } - * ``` - */ - static scope( - futures: Iterable> - ): Future { - // Check if the input is iterable - const iterator = futures?.[Symbol.iterator]?.(); - - // Iterate over the iterable/async iterable futures in a controlled manner - return new Future(async function* () { - // If no valid iterator was found, throw an error indicating that the input is not iterable or an iterator - if ( - (iterator ?? null) === null || - typeof (iterator as Iterator>)?.next === 'function' - ) throw new TypeError("The provided input is not an iterable nor an iterator."); - - // Handle the async generator or generator in a pull-based workflow - let result: IteratorResult>; - - // Start the iteration - while (!(result = iterator.next()).done) { - yield yield* result.value; - } - - return yield* result.value; - }); - } - - /** - * Sets up a `Future` to execute in the background during idle time. - * - * This method does not execute the future itself, but prepares it to be run using - * `requestIdleCallback`. Execution is still controlled by methods like `toPromise()` or `async` iterators. - * - * @param future The future to be executed in the background. - * @returns A new `Future` instance set up for background execution. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * return 100; - * }); - * const backgroundFuture = Future.inBackground(future); // result is 100, processed in the background - * ``` - */ - static inBackground(future: Future): Future { - // Check if the input is iterable - const generator = future?.[Symbol.asyncIterator]?.(); - - // Iterate over the iterable/async iterable futures in a controlled manner - return new Future(async function* () { - // If no valid iterator was found, throw an error indicating that the input is not iterable or an iterator - if ( - (generator ?? null) === null || - typeof (generator as AsyncGenerator)?.next === 'function' - ) throw new TypeError("The provided input is not a future."); - - let idleResolver: PromiseWithResolvers | null = Promise.withResolvers(); - let idleId = idle(() => idleResolver?.resolve?.()); - - try { - await idleResolver.promise; - cancelIdle(idleId); - - // Handle the async generator or generator in a pull-based workflow - let result = await generator.next(); - - idleResolver = Promise.withResolvers(); - idleId = idle(() => idleResolver?.resolve?.()); - - // Start the iteration - while (!result.done) { - await idleResolver.promise; - cancelIdle(idleId); - - result = await generator.next(yield result.value); - - idleResolver = Promise.withResolvers(); - idleId = idle(() => idleResolver?.resolve?.()); - } - - return result.value; - } finally { - idleResolver = null; - cancelIdle(idleId); - } - }); - } - - /** - * Limits the number of concurrent `Future` operations. - * - * This method is useful for controlling how many Futures run at the same time, - * which can prevent overwhelming resources or ensure a more manageable load. - * - * @param futures - An iterable of `Future` or `PromiseLike` objects. - * @param limit - The maximum number of Futures to run concurrently. - * @returns A `Future` that yields results as each operation completes. - * - * @example - * ```typescript - * const futures = [ - * Promise.resolve(1), - * Promise.resolve(2), - * Promise.resolve(3), - * Future.from(async function* () { - * yield 42; - * return 100; - * }), - * Future.from(async function* () { - * yield 10; - * return 20; - * }) - * ]; - * - * for await (const result of future) { - * console.log(result); // Logs 1, 2, 3, 100, 20 in order with a concurrency limit of 2 - * } - * ``` - * - * @example - * ```typescript - * const futures = [ - * fetch('https://api.example.com/1'), - * fetch('https://api.example.com/2'), - * fetch('https://api.example.com/3') - * ]; - * - * const limitedFuture = Future.withConcurrencyLimit(futures, 2); - * - * for await (const result of limitedFuture) { - * console.log(result); // Process each result as it becomes available - * } - * ``` - */ - static withConcurrencyLimit( - futures: Iterable | PromiseLike>, - limit: number - ): Future { - // Check if the input is an async iterator, sync iterator, or iterable - // We use Symbol.asyncIterator and Symbol.iterator to distinguish between different types - const iterator = futures?.[Symbol.iterator]?.(); - - return new Future(async function* () { - // If no valid iterator was found, throw an error indicating that the input is not iterable or an iterator - if ( - (iterator ?? null) === null || - typeof iterator?.next === 'function' - ) throw new TypeError("The provided input is not an iterable nor an iterator."); - - const activeFutures = new Map< - PromiseLike, - Promise<{ result: T | TReturn, promise: PromiseLike }> - >(); - let finalResult: T | TReturn | undefined; - - while (activeFutures.size < limit) { - const { done, value } = iterator.next(); - if (done) break; - - // Start the future concurrently - const promise = value instanceof Future ? value.toPromise() : value; - - // Wrap the promise and store it in the map with the original promise as the key - activeFutures.set(promise, Promise.resolve(promise).then(result => ({ result, promise }))); - - // If we hit the concurrency limit, wait for one to resolve - if (activeFutures.size >= limit) { - const { result: finished, promise } = await Promise.race(activeFutures.values()) - activeFutures.delete(promise); - finalResult = finished; - yield finished; - } - } - - // After the main loop, yield any remaining futures - for await (const remainingFuture of activeFutures.values()) { - finalResult = remainingFuture.result; - activeFutures.delete(remainingFuture.promise); - yield finalResult; - } - - return finalResult; - }); - } - - /** - * Sets a deadline for a future, canceling it if it takes longer than the specified time to complete. - * If the future resolves before the deadline, the timeout is cleared. - * - * @param future - The future to set a deadline for. - * @param ms - The time in milliseconds before the future is canceled. - * @returns A future that will be canceled if it exceeds the specified time. - * @example - * ```typescript - * const future = Future.from(async function* () { - * yield 42; - * return 100; - * }); - * - * const deadlineFuture = Future.withDeadline(future, 1000); // Sets a 1-second deadline - * ``` - */ - static withDeadline(future: Future, ms: number) { - return new Future(async function* () { - const { promise: timeout, reject } = Promise.withResolvers(); - - const timeoutId = setTimeout(() => { - future.cancel(new Error("Future timed out")); - reject(new Error("Future timed out")); - }, ms); - - const result = await Promise.race([future.toPromise(), timeout]); - clearTimeout(timeoutId); // Clear the timeout if the future resolves in time - - yield result as T | TReturn; - return result as T | TReturn; - }); - } - - /** - * Provides resolvers for manually controlling the resolution of a future. - * @returns An object containing the Future, the resolve and reject methods. - */ - static withResolvers(): FutureWithResolvers { - const { promise, resolve, reject } = Promise.withResolvers(); - const future = Future.fromPromise(promise); - - return { - future, - resolve, - reject - }; - } -} - -export interface FutureWithResolvers extends Omit, "promise"> { - future: Future; - resolve: (value: TReturn | PromiseLike) => void; - reject: (reason?: unknown) => void; -} - -export interface FutureOperation { - (abort: AbortController): AsyncGenerator | Generator -} - -export interface FutureFromOperation { - (abort: AbortController): ReturnType> | PromiseLike | T -} - -export default Future; \ No newline at end of file +export * from "./future.ts"; +export * from "./from.ts"; +export * from "./deadline.ts"; +export * from "./deadline.ts"; +export * from "./concurrent.ts"; +export * from "./resolvers.ts"; + +export { default } from "./future.ts"; \ No newline at end of file diff --git a/resolvers.ts b/resolvers.ts new file mode 100644 index 0000000..14091e0 --- /dev/null +++ b/resolvers.ts @@ -0,0 +1,22 @@ +import { Future } from "./future.ts"; + +/** + * Provides resolvers for manually controlling the resolution of a future. + * @returns An object containing the Future, the resolve and reject methods. + */ +export function withResolvers(): FutureWithResolvers { + const { promise, resolve, reject } = Promise.withResolvers(); + const future = Future.fromPromise(promise); + + return { + future, + resolve, + reject + }; +} + +export interface FutureWithResolvers extends Omit, "promise"> { + future: Future; + resolve: (value: TReturn | PromiseLike) => void; + reject: (reason?: unknown) => void; +} \ No newline at end of file diff --git a/scope.ts b/scope.ts new file mode 100644 index 0000000..b9e5f8d --- /dev/null +++ b/scope.ts @@ -0,0 +1,81 @@ +import { Future } from "./future.ts"; + +/** + * Runs multiple Futures within a defined scope, ensuring that all Futures are executed + * with full control over their execution (e.g., pausing, resuming, and pulling/pushing values). + * + * This method supports `Iterable` and `AsyncIterable` containers of Futures, allowing for + * both push-based and pull-based workflows. + * + * `Future.scope` is designed to handle non-concurrent execution by default, allowing you to + * control the flow of each future one at a time. If concurrency is needed, you can use + * `Future.all`, `Future.some`, or similar methods inside the scope. + * + * ### Push vs. Pull Workflows: + * + * - **Push-Based Workflow**: The generator yields values autonomously, and the consumer simply awaits those values. + * - **Pull-Based Workflow**: The generator waits for external input before proceeding to the next value. + * + * @param futures - An iterable or async iterable of Futures to be run within the scope. + * @returns A Future that yields the results of the contained Futures, controlled by the scope. + * + * ### Example Usage: + * + * ```typescript + * const scopeFuture = Future.scope([ + * Future.from(async function* () { + * yield 42; + * return 100; + * }), + * Future.from(async function* () { + * yield 10; + * return 20; + * }) + * ]); + * + * for await (const result of scopeFuture) { + * console.log(result); // Logs 100, 20 + * } + * + * const pullFuture = Future.scope([ + * Future.from(async function* () { + * let result = { value: 42, done: false }; + * + * while (!result.done) { + * result = await (yield result.value); // Wait for input from the consumer + * } + * + * return result.value; + * }) + * ]); + * + * const iterator = pullFuture[Symbol.asyncIterator](); + * console.log(await iterator.next()); // { value: 42, done: false } + * console.log(await iterator.next({ value: 100, done: true })); // { value: 100, done: true } + * ``` + */ +export function scope( + futures: Iterable> +): Future { + // Check if the input is iterable + const iterator = futures?.[Symbol.iterator]?.(); + + // Iterate over the iterable/async iterable futures in a controlled manner + return new Future(async function* () { + // If no valid iterator was found, throw an error indicating that the input is not iterable or an iterator + if ( + (iterator ?? null) === null || + typeof (iterator as Iterator>)?.next === 'function' + ) throw new TypeError("The provided input is not an iterable nor an iterator."); + + // Handle the async generator or generator in a pull-based workflow + let result: IteratorResult>; + + // Start the iteration + while (!(result = iterator.next()).done) { + yield yield* result.value; + } + + return yield* result.value; + }); +} \ No newline at end of file -- 2.51.2