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