diff --git a/.gitignore b/.gitignore index aed69d6..1c19fa0 100644 --- a/.gitignore +++ b/.gitignore @@ -138,6 +138,7 @@ coverage node_modules package.json package-lock.json +pnpm-lock.yaml bun.lockb # Vercel diff --git a/deno.jsonc b/deno.jsonc index da0faee..7faf060 100644 --- a/deno.jsonc +++ b/deno.jsonc @@ -8,7 +8,6 @@ "logtape", "esm" ], - "license": "MIT", "author": "okikio (Okiki Ojo)", "funding": "https://github.com/sponsors/okikio", "homepage": "https://github.com/okikio/logger", diff --git a/mod.ts b/mod.ts index e5a8d28..3694430 100644 --- a/mod.ts +++ b/mod.ts @@ -1,5 +1,11 @@ import { cancelIdle, idle } from "./idle.ts"; +import { isAsyncIterable, isAsyncIterator, isIterable } from "./utils.ts"; +export class CancellationError extends Error { + constructor() { + super("Future was canceled"); + } +} export const Status = { Idle: 0, @@ -66,9 +72,9 @@ export type StatusEnum = typeof Status[keyof typeof Status]; * * @template T - The type of the value that the future will resolve to. */ -export class Future implements PromiseLike, AsyncIterableIterator { - #generator?: AsyncGenerator> | null; - #operation?: ((signal: AbortSignal) => AsyncGenerator>) | null; +export class Future implements PromiseLike { + #generator?: AsyncGenerator | null; + #operation?: ((signal: AbortSignal) => AsyncGenerator) | null; #status: StatusEnum = Status.Idle; readonly #resolvers = { pause: Promise.withResolvers(), @@ -76,9 +82,9 @@ export class Future implements PromiseLike, AsyncIterableIterator { abort: new AbortController(), } - static #EventHandler = class PrivateEventHandler { - #delegate: Future | undefined | null; - constructor(delegate?: Future) { + static #EventHandler = class PrivateEventHandler { + #delegate: Future | undefined | null; + constructor(delegate?: Future) { this.#delegate = delegate; } @@ -99,7 +105,7 @@ export class Future implements PromiseLike, AsyncIterableIterator { * Creates a new `Future` instance. * @param operation - A function that returns an async generator to define the asynchronous task. */ - constructor(operation: (signal: AbortSignal) => AsyncGenerator>) { + constructor(operation: (signal: AbortSignal) => AsyncGenerator) { this.#operation = operation; this.#setup(); @@ -184,7 +190,7 @@ export class Future implements PromiseLike, AsyncIterableIterator { * ``` */ cancel() { - this.#resolvers?.abort?.abort(new CancellationError()); + this.#resolvers?.abort?.abort?.(new CancellationError()); return this; } @@ -257,7 +263,9 @@ export class Future implements PromiseLike, AsyncIterableIterator { this.#resolvers.abort?.signal?.removeEventListener?.("abort", this.#eventhandler); this.#eventhandler?.destroy?.(); - this.#eventhandler = null as unknown as any; + + // @ts-ignore Resetting private properties + this.#eventhandler = null as unknown; Object.assign(this.#resolvers, { pause: null, @@ -353,13 +361,12 @@ export class Future implements PromiseLike, AsyncIterableIterator { * * @yields The values generated by the future. */ - async *[Symbol.asyncIterator](): AsyncGenerator, Awaited, unknown> { + async *[Symbol.asyncIterator](): AsyncGenerator { let err: unknown; try { - if ( - !(this.#generator ?? null) || - typeof this.#generator?.next !== "function" - ) throw new Error("generator not defined"); + if (!this.#generator || typeof this.#generator?.next !== "function") { + throw new Error("generator not defined"); + } // Prime the generator by starting the iteration process let result = await this.#generator?.next?.(); @@ -395,27 +402,36 @@ export class Future implements PromiseLike, AsyncIterableIterator { } } + [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 result of the future. + * @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.fromOperation(async function* () { * yield 42; - * return 100; + * // No explicit return, so the final result will be undefined * }); - * const result = await future.toPromise(); // result is 100 + * const result = await future.toPromise(); // result is undefined * ``` */ - async toPromise() { - let finalValue: T; - for await (const value of this) { - finalValue = value; - } + 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); - return finalValue!; + // If the generator has no explicit return, result.value will be undefined + return result.value; } /** @@ -424,9 +440,9 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @param onrejected Called when the future is rejected. * @returns A promise that resolves with the result of the future. */ - then( - onfulfilled?: ((value: T) => TResult1 | PromiseLike) | undefined | null, - onrejected?: ((reason: any) => TResult2 | PromiseLike) | undefined | null + then( + onfulfilled?: ((value: TReturn) => TResult1 | PromiseLike) | undefined | null, + onrejected?: ((reason: unknown) => TResult2 | PromiseLike) | undefined | null ): Promise { return this.toPromise().then(onfulfilled, onrejected); } @@ -437,8 +453,8 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @returns A promise that resolves or rejects based on the future's outcome. */ catch( - onrejected?: ((reason: any) => TResult | PromiseLike) | undefined | null - ): Promise { + onrejected?: ((reason: unknown) => TResult | PromiseLike) | undefined | null + ): Promise { return this.toPromise().catch(onrejected); } @@ -447,7 +463,7 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @param onfinally Called when the future is complete. * @returns A promise that resolves to the final result. */ - finally(onfinally?: (() => void) | undefined | null): Promise { + finally(onfinally?: (() => void) | undefined | null): Promise { return this.toPromise().finally(onfinally); } @@ -569,13 +585,17 @@ export class Future implements PromiseLike, AsyncIterableIterator { * ``` * */ - static fromOperation( + static fromOperation( operation: - | ((signal: AbortSignal) => AsyncGenerator> | Generator>) - | PromiseLike + | ((signal: AbortSignal) => AsyncGenerator | Generator | PromiseLike) + | PromiseLike + | Future + | ReadableStream | AsyncIterable | Iterable - ): Future { + | Iterator + | AsyncIterator + ) { // Handle Future instances directly if (operation instanceof Future) { return operation; @@ -585,72 +605,45 @@ export class Future implements PromiseLike, AsyncIterableIterator { if (operation instanceof ReadableStream) { return Future.fromStream(operation); } + + if (typeof operation === "function") { + return new Future(async function* (signal: AbortSignal) { + const result = await operation(signal) as + | AsyncGenerator + | Generator + | PromiseLike; + + if (isAsyncIterable(result) || isIterable(result)) { + // Handle async iterable or iterable result + return yield* result as AsyncGenerator; + } - // Handle Promise-like objects - if (typeof (operation as PromiseLike)?.then === 'function') { - return Future.fromPromise(operation as PromiseLike); + // Handle promise-like result or single value + yield result as T; + return result as TReturn; + }); } - return new Future(async function* (signal: AbortSignal) { - // Check what the source of the async generator or generator is - const source = typeof operation === 'function' ? await operation(signal) : operation; - if ((source ?? null) === null) { - yield source; - return source; - } - - // Check if the operation is an async generator or generator - const generator = - (source as AsyncIterable)?.[Symbol.asyncIterator]?.() ?? - (source as Iterable)?.[Symbol.iterator]?.() ?? - source as AsyncIterator; - - // Handle async iterators or generators - if ( - Symbol.asyncIterator in source || - Symbol.iterator in source || - typeof (generator as AsyncIterator)?.next === 'function' - ) { - // Prime the generator once before starting the main loop - let result = await generator.next(); - - // Continue yielding values until the generator completes - while (!result?.done) { - // Wait for the next value to be passed in through next() - result = await generator.next(yield result.value); + return new Future(async function* () { + const result = operation as + | AsyncIterable + | Iterable + | Iterator + | AsyncIterator; + if (isAsyncIterable(result) || isIterable(result)) { + return yield* result; + } else if (isAsyncIterator(result)) { + let iteratorResult = await result.next(); + while (!iteratorResult.done) { + iteratorResult = await result.next(yield iteratorResult.value); } - // Return the final value - return result.value; - } else { - // Handle single value cases (e.g., a plain value or Promise) - yield generator as T; + return iteratorResult.value; } - return generator as T; - }); - } - - /** - * Creates a `Future` from a standard JavaScript `Promise`. - * - * This method allows you to wrap a `Promise` in a `Future` to take advantage of the - * additional control provided by `Future`, such as pausing, resuming, and cancellation. - * - * @param promise - The promise to wrap in a `Future`. - * @returns A new `Future` instance wrapping the given promise. - * - * @example - * ```typescript - * const future = Future.fromPromise(fetch('https://api.example.com/data')); - * - * const data = await future.toPromise(); // data is the fetched result - * ``` - */ - static fromPromise(promise: PromiseLike): Future { - return new Future(async function* () { - yield promise; - return promise; + // Yield a single value from the resolved promise + yield operation as T; + return operation as TReturn; }); } @@ -687,20 +680,17 @@ export class Future implements PromiseLike, AsyncIterableIterator { * } * ``` */ - static fromStream(stream: ReadableStream): Future { - return new Future(async function* (signal: AbortSignal) { + static fromStream(stream: ReadableStream): Future, Awaited | undefined, Awaited>> { + return new Future, Awaited | undefined, Awaited>>(async function* () { const reader = stream.getReader(); try { while (true) { - if (signal.aborted) break; const { done, value } = await reader.read(); - if (done || signal.aborted) { - break; - } - if (value) { - yield value; // Yield each chunk of data - } + if (done) return value; + + // Yield each chunk of data + yield value; } } catch (error) { reader.cancel(error); @@ -711,107 +701,6 @@ export class Future implements PromiseLike, AsyncIterableIterator { }); } - /** - * Utility method to collect and normalize iterable, async iterable, or iterator of Futures or PromiseLike objects. - * Converts each Future or promise into a promise, ensuring consistent handling of both sync and async iterables/iterators. - * - * This method supports the following types: - * - `Iterable | PromiseLike>` - * - `AsyncIterable | PromiseLike>` - * - `Iterator | PromiseLike>` - * - `AsyncIterator | PromiseLike>` - * - * @param futures - An iterable, async iterable, iterator, or async iterator of `Future` or `PromiseLike` objects. - * @returns A promise that resolves with an array of promises. - * - * ### Usage Examples: - * - * ```typescript - * // Example 1: Synchronous iterable - * const syncIterable = [Promise.resolve(1), Promise.resolve(2), Future.fromPromise(Promise.resolve(3))]; - * for await (const result of Future.all(syncIterable)) { - * console.log(result); // Logs: 1, 2, 3 - * } - * - * // Example 2: Asynchronous iterable - * async function* asyncIterable() { - * yield Promise.resolve(4); - * yield Future.fromOperation(async function* () { - * yield 5; - * return 6; - * }); - * } - * - * for await (const result of Future.all(asyncIterable())) { - * console.log(result); // Logs: 4, 6 - * } - * - * // Example 3: Synchronous iterator - * function* syncIterator() { - * yield Promise.resolve(7); - * yield Future.fromPromise(Promise.resolve(8)); - * } - * - * for await (const result of Future.all(syncIterator())) { - * console.log(result); // Logs: 7, 8 - * } - * - * // Example 4: Asynchronous iterator - * async function* asyncIterator() { - * yield Future.fromPromise(Promise.resolve(9)); - * yield Promise.resolve(10); - * } - * - * for await (const result of Future.all(asyncIterator())) { - * console.log(result); // Logs: 9, 10 - * } - * ``` - */ - static async collectIterable( - futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike> - ): Promise | PromiseLike>> { - const collected: Array> = []; - - // 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 as AsyncIterable | PromiseLike>)?.[Symbol.asyncIterator]?.() || - (futures as Iterable | PromiseLike>)?.[Symbol.iterator]?.() || - (futures as AsyncIterator | PromiseLike>); - - // 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 AsyncIterator | PromiseLike>)?.next !== 'function' - ) throw new TypeError("The provided input is not an iterable nor an iterator."); - - // IteratorResult contains `value` and `done` properties. `done` indicates if iteration is complete. - let result: IteratorResult | PromiseLike>; - - // Iterate through the input using `next()` to process each item - while (!(result = await iterator.next()).done) { - const future = result.value; - - // Allow for pausing and resuming of each future - const futureIterator = Symbol.asyncIterator in future ? future[Symbol.asyncIterator]() : (async function* () { - yield (future instanceof Future ? future.toPromise() : future); - })(); - - // Process the future step-by-step, handling the push workflow - let futureResult: IteratorResult = await futureIterator.next(); - - while (!futureResult.done) { - // Yield the value from the current future to the scope consumer - futureResult = await futureIterator.next(); - } - - // Push the normalized promise into the collected array - collected.push(futureResult.value); - } - - // Return the collected array of promises - return collected; - } - /** * Runs multiple Futures concurrently, yielding their results as they all complete. * @@ -821,21 +710,15 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @param futures - An iterable of `Future` or `PromiseLike` objects. * @returns An AsyncIterable yielding each result as they complete. */ - static async *all(futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike>): AsyncIterable { - const collectedFutures = await Future.collectIterable(futures); - - // We trigger all futures at once, using Promise.all to await them concurrently. - const results = await Promise.all(collectedFutures); - for (const result of results) { - yield result; // Yield each result in the original order. - } + 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; + }); } - /** - * Alias for {@link Future.all} - */ - static concurrent = Future.all; - /** * 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. @@ -864,74 +747,14 @@ export class Future implements PromiseLike, AsyncIterableIterator { * } * ``` */ - static allSettled( - futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike> - ): Future> { - return new Future>(async function* () { - const collectedFutures = await Future.collectIterable(futures); - + static allSettled( + futures: Iterable | PromiseLike> + ): Future>, PromiseSettledResult>[], undefined> { + return new Future>, PromiseSettledResult>[], undefined>(async function* () { // We trigger all futures at once, using Promise.allSettled to await them concurrently. - const results = await Promise.allSettled(collectedFutures); - - for (const result of results) { - yield result; // Yield each result in the original order. - } - }); - } - - /** - * Runs multiple Futures in sequence, yielding each result as they complete. - * - * This method is useful when you need to ensure that Futures are run one after the other, - * rather than concurrently. - * - * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. - * @returns An AsyncIterable yielding each result as they complete. - */ - static sequence( - futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike> - ): 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 - // Iterate over the iterable/async iterable futures in a controlled manner - const iterator = - (futures as AsyncIterable | PromiseLike>)?.[Symbol.asyncIterator]?.() || - (futures as Iterable | PromiseLike>)?.[Symbol.iterator]?.() || - (futures as AsyncIterator | PromiseLike>); - - 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 AsyncIterator)?.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 | PromiseLike>; - - // Start the iteration - while (!(result = await iterator.next()).done) { - const future = result.value; - - // Allow for pausing and resuming of each future - const futureIterator = Symbol.asyncIterator in future ? future[Symbol.asyncIterator]() : (async function* (): AsyncGenerator, Awaited, unknown> { - const promise = (future instanceof Future ? future.toPromise() : future) as Promise; - yield promise; - return promise; - })(); - - // Process the future step-by-step, handling the push/pull workflow - let futureResult = await futureIterator.next(); - - while (!futureResult.done) { - // Yield the value from the current future to the scope consumer - futureResult = await futureIterator.next(yield futureResult.value); - } - - // Yield the final result of the future - yield futureResult.value; - if (futureResult.done) return futureResult.value; - } + const results = await Promise.allSettled(futures); + yield* results; + return results; }); } @@ -941,11 +764,13 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @param futures - An iterable or async iterable of `Future` or `PromiseLike` objects. * @returns An AsyncIterable yielding the first result that resolves. */ - static any( - futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike> - ): Future { - return new Future(async function* () { - yield await Promise.race(await Future.collectIterable(futures)); + static any( + futures: Iterable | PromiseLike> + ): Future { + return new Future(async function* () { + const result = Promise.race(futures); + yield result; + return result; }); } @@ -956,16 +781,13 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @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> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike>, + static some( + futures: Iterable | PromiseLike>, count: number - ): Future> { - return new Future>(async function* () { - const collectedFutures = await Future.collectIterable(futures); - yield* Future.allSettled( - collectedFutures.slice(0, count) - ); - }); + ): Future>, PromiseSettledResult>[]> { + return Future.allSettled( + Array.from(futures).slice(0, count) + ); } /** @@ -1022,79 +844,29 @@ export class Future implements PromiseLike, AsyncIterableIterator { * console.log(await iterator.next({ value: 100, done: true })); // { value: 100, done: true } * ``` */ - static scope( - futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike> | Iterator | PromiseLike> | AsyncIterator | PromiseLike> - ): 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 as AsyncIterable)?.[Symbol.asyncIterator]?.() || - (futures as Iterable)?.[Symbol.iterator]?.() || - (futures as AsyncIterator); + 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* () { + 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 AsyncIterator)?.next === 'function' + 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>; + let result: IteratorResult>; // Start the iteration - while (!(result = await iterator.next()).done) { - const future = result.value; - - // Allow for pausing and resuming of each future - const futureIterator = future?.[Symbol.asyncIterator]?.() ?? (async function* () { yield future.toPromise(); })(); - - // Process the future step-by-step, handling the push/pull workflow - let futureResult: IteratorResult = await futureIterator.next(); - - while (!futureResult.done) { - // Yield the value from the current future to the scope consumer - futureResult = await futureIterator.next(yield futureResult.value); - } - - // Yield the final result of the future - yield futureResult.value; + while (!(result = iterator.next()).done) { + yield yield* result.value; } - }); - } - /** - * Chains two futures together, where the result of the first future is passed to the second future. - * This method allows for sequential execution of dependent futures. - * - * @param future - The initial future whose result will be passed to the next future. - * @param continuation - A function that takes the result of the first future and returns a new future. - * @returns A future representing the continuation of both futures. - * @example - * ```typescript - * const future = Future.fromOperation(async function* () { - * yield 42; - * return 100; - * }); - * - * const continuedFuture = Future.continueWith(future, (result) => { - * return Future.fromOperation(async function* () { - * yield result + 10; - * return result + 20; - * }); - * }); - * - * const finalResult = await continuedFuture.toPromise(); // finalResult is 120 - * ``` - */ - static continueWith( - future: Future, - continuation: (result: T) => Future - ): Future { - return new Future(async function* () { - const result = await (yield* future); - yield* continuation(result); + return yield* result.value; }); } @@ -1115,24 +887,43 @@ export class Future implements PromiseLike, AsyncIterableIterator { * const backgroundFuture = Future.inBackground(future); // result is 100, processed in the background * ``` */ - static inBackground(future: Future): Future { - return new Future(async function* () { + 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 { - let finalValue: T; - for await (const value of future) { + 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); - yield (finalValue = value); - + result = await generator.next(yield result.value); + idleResolver = Promise.withResolvers(); idleId = idle(() => idleResolver?.resolve?.()); } - return finalValue!; + return result.value; } finally { idleResolver = null; cancelIdle(idleId); @@ -1153,9 +944,9 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @example * ```typescript * const futures = [ - * Future.fromPromise(Promise.resolve(1)), - * Future.fromPromise(Promise.resolve(2)), - * Future.fromPromise(Promise.resolve(3)), + * Promise.resolve(1), + * Promise.resolve(2), + * Promise.resolve(3), * Future.fromOperation(async function* () { * yield 42; * return 100; @@ -1174,9 +965,9 @@ export class Future implements PromiseLike, AsyncIterableIterator { * @example * ```typescript * const futures = [ - * Future.fromPromise(fetch('https://api.example.com/1')), - * Future.fromPromise(fetch('https://api.example.com/2')), - * Future.fromPromise(fetch('https://api.example.com/3')) + * fetch('https://api.example.com/1'), + * fetch('https://api.example.com/2'), + * fetch('https://api.example.com/3') * ]; * * const limitedFuture = Future.withConcurrencyLimit(futures, 2); @@ -1186,40 +977,54 @@ export class Future implements PromiseLike, AsyncIterableIterator { * } * ``` */ - static withConcurrencyLimit( - futures: Iterable | PromiseLike> | AsyncIterable | PromiseLike>, + static withConcurrencyLimit( + futures: Iterable | PromiseLike>, limit: number - ): Future> { - return new Future>(async function* () { - const activeFutures = new Set>(); - - // 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 as AsyncIterable)?.[Symbol.asyncIterator]?.() || - (futures as Iterable)?.[Symbol.iterator]?.() || - (futures as AsyncIterator); + ) { + // 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: TReturn, promise: PromiseLike }> + >(); + let finalResult: TReturn | undefined; while (activeFutures.size < limit) { const { done, value } = iterator.next(); if (done) break; // Start the future concurrently - const futurePromise = value instanceof Future ? value.toPromise() : value; - activeFutures.add(futurePromise); + 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 finished = await Promise.race(activeFutures); - activeFutures.delete(finished); + 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 (const remainingFuture of activeFutures) { - yield await remainingFuture; + for await (const remainingFuture of activeFutures.values()) { + finalResult = remainingFuture.result; + activeFutures.delete(remainingFuture.promise); + yield finalResult; } + + return finalResult; }); } @@ -1241,8 +1046,8 @@ export class Future implements PromiseLike, AsyncIterableIterator { * const deadlineFuture = Future.withDeadline(future, 1000); // Sets a 1-second deadline * ``` */ - static withDeadline(future: Future, ms: number): Future { - return new Future(async function* () { + static withDeadline(future: Future, ms: number): Future { + return new Future(async function* () { let timeoutId: ReturnType; const timeout = new Promise((_, reject) => { @@ -1252,63 +1057,34 @@ export class Future implements PromiseLike, AsyncIterableIterator { }, ms); }); - try { - const result = await Promise.race([future.toPromise(), timeout]); - clearTimeout(timeoutId); // Clear the timeout if the future resolves in time - yield result; - return result; - } catch (error) { - throw error; - } - }); + const result = await Promise.race([future.toPromise(), timeout]); + if ((timeoutId! ?? null) !== null) clearTimeout(timeoutId!); // Clear the timeout if the future resolves in time + + yield result as T; + return result as TReturn; + }); } /** * Provides resolvers for manually controlling the resolution of a future. * @returns An object containing the Future, the resolve and reject methods. */ - static withResolvers(): { - future: Future; - resolve: (value: T | PromiseLike) => void; - reject: (reason?: any) => void; - } { - let resolveFn: (value: T | PromiseLike) => void; - let rejectFn: (reason?: any) => void; - - const future = new Future(async function* () { - return await new Promise((resolve, reject) => { - resolveFn = resolve; - rejectFn = reject; - }); - }); + static withResolvers(): FutureWithResolvers { + const { promise, resolve, reject } = Promise.withResolvers(); + const future = Future.fromOperation(promise); return { future, - resolve: resolveFn!, - reject: rejectFn! + resolve, + reject }; } } -// Example usage: -const future = Future.scope( - Future.withConcurrencyLimit([ - Promise.resolve(10), - Future.fromOperation(async function* () { - yield 5; - return 5; - }), - Future.fromOperation(async function () { - return await 10; - }).runInBackground(), - Future.all([Promise.resolve(20), Promise.resolve(30)]).runInBackground(), - Future.any([Promise.resolve(5), Promise.resolve(20)]), - ], 5) -); - -setTimeout(() => future.pause(), 1000); -setTimeout(() => future.resume(), 3000); - -for await (const value of future) { - console.log(value); -} + + +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/mod_test.ts b/mod_test.ts index 5a1a25e..cde850f 100644 --- a/mod_test.ts +++ b/mod_test.ts @@ -1,299 +1,7 @@ -import type { Logger } from "@logtape/logtape"; +import { test } from "@libs/testing"; +import { expect } from "@std/expect"; -import { describe, it, afterEach } from "@std/testing/bdd"; -import { spy, assertSpyCalls, } from "@std/testing/mock"; - -import { expect } from "@libs/testing"; -import { createLogger } from './mod.ts'; // Adjust the import based on your file structure - -describe('Logger Library', () => { - const mockLogger = { - error: spy(() => { }), - warn: spy(() => { }), - info: spy(() => { }), - debug: spy(() => { }), - fatal: spy(() => { }), - }; - - const loggers = { - general: mockLogger as unknown as Logger, - init: mockLogger as unknown as Logger, - build: mockLogger as unknown as Logger, - customLogger: mockLogger as unknown as Logger, - }; - - afterEach(() => { - // Reset mock call history - // mockLogger.error.restore(); - // mockLogger.warn.restore(); - // mockLogger.info.restore(); - // mockLogger.debug.restore(); - // mockLogger.fatal.restore(); - }); - - - describe('Basic Logging Operations', () => { - it('should log to the general logger by default', () => { - const logger = createLogger({ loggers }); - logger.info`General Log ${5}`; - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([ - ['General Log ', ''], - 5, - ]); - }); - - it('should log to a specific custom logger', () => { - const logger = createLogger({ loggers }); - logger.init.error('Initialization failed'); - assertSpyCalls(mockLogger.error, 1); - expect(mockLogger.error.calls[0].args).toEqual(['Initialization failed', {}]); - }); - - it('should log to a specific custom logger (using tagged templates)', () => { - const logger = createLogger({ loggers }); - logger.init.error`Initialization failed: ${'test'}`; - assertSpyCalls(mockLogger.error, 1); - expect(mockLogger.error.calls[0].args).toEqual([ - ['Initialization failed: ', ''], - 'test' - ]); - }); - - it('should log with various log levels', () => { - const logger = createLogger({ loggers }); - logger.warn('A warning message'); - logger.debug('Debugging message'); - logger.fatal('A fatal message'); - assertSpyCalls(mockLogger.warn, 1); - expect(mockLogger.warn.calls[0].args).toEqual(['A warning message', {}]); - assertSpyCalls(mockLogger.debug, 1); - expect(mockLogger.debug.calls[0].args).toEqual(['Debugging message', {}]); - assertSpyCalls(mockLogger.fatal, 1); - expect(mockLogger.fatal.calls[0].args).toEqual(['A fatal message', {}]); - }); - - it('should log with various log levels (using tagged templates)', () => { - const logger = createLogger({ loggers }); - logger.warn`A warning message: ${'something'}`; - logger.debug`Debugging message: ${42}`; - logger.fatal`A fatal message: ${new Error('error')}`; - assertSpyCalls(mockLogger.warn, 1); - expect(mockLogger.warn.calls[0].args).toEqual([ - ['A warning message: ', ''], - 'something' - ]); - assertSpyCalls(mockLogger.debug, 1); - expect(mockLogger.debug.calls[0].args).toEqual([ - ['Debugging message: ', ''], - 42 - ]); - assertSpyCalls(mockLogger.fatal, 1); - expect(mockLogger.fatal.calls[0].args).toEqual([ - ['A fatal message: ', ''], - new Error('error') - ]); - }); - }); - - describe('Logger Initialization', () => { - it('should initialize with default options', () => { - const logger = createLogger({}); - expect(logger.general).toBeUndefined(); // General logger is undefined by default - }); - - it('should initialize with a custom template', () => { - const logger = createLogger({ loggers, template: 'Custom Template' }); - logger.info('Logging with custom template'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Logging with custom template', {}]); - }); - - it('should initialize with context data', () => { - const context = { user: 'testUser' }; - const logger = createLogger({ loggers, context }); - logger.info('Logging with context'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Logging with context', context]); - }); - - it('should initialize with multiple custom loggers', () => { - const logger = createLogger({ loggers }); - expect(logger.init).toBe(mockLogger); - expect(logger.build).toBe(mockLogger); - expect(logger.customLogger).toBe(mockLogger); - }); - }); - - describe('Logger Methods', () => { - it('should modify logger options dynamically using with()', () => { - const logger = createLogger({ loggers }); - logger.with({ logger: 'build', template: 'Building...' }).info('Starting build'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Starting build', {}]); - }); - - it('should add context data using value()', () => { - const logger = createLogger({ loggers }); - logger.context({ requestId: '12345' }).info('Request received'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Request received', { requestId: '12345' }]); - }); - - it('should dynamically create getters for custom loggers', () => { - const logger = createLogger({ loggers }); - expect(logger.init).toBe(mockLogger); - expect(logger.build).toBe(mockLogger); - }); - - it('should switch to a custom logger dynamically (using tagged template)', () => { - const logger = createLogger({ loggers }); - logger.init.info`Initialization log: ${'init'}`; - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([ - ['Initialization log: ', ''], - 'init' - ]); - }); - }); - - describe('Error Handling', () => { - it('should throw an error for a non-existent logger', () => { - const logger = createLogger({ loggers }); - // @ts-expect-error nonExistentLogger - expect(() => logger.with({ logger: 'nonExistentLogger' }).info('This should fail')).toThrow('Logger "nonExistentLogger" is not registered.'); - }); - - it('should throw an error if the logger option is provided in with() and it does not exist', () => { - const logger = createLogger({ loggers }); - // @ts-expect-error nonExistentLogger - expect(() => logger.with({ logger: 'nonExistentLogger' })).toThrow('Logger "nonExistentLogger" is not registered.'); - }); - }); - - describe('Log Message Formatting', () => { - it('should log a simple string message', () => { - const logger = createLogger({ loggers }); - logger.info('Simple log message'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Simple log message', {}]); - }); - - it('should log a message with template strings and values', () => { - const logger = createLogger({ loggers }); - logger.info`Template Log ${5}`; - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([ - ['Template Log ', ''], - 5 - ]); - }); - - it('should log a message with a callback function', () => { - const logger = createLogger({ loggers }); - const callback = (prefix: (msg: TemplateStringsArray, ...vals: unknown[]) => unknown[]) => prefix`Callback Log ${5}`; - logger.info(callback); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([callback]); - }); - - it('should log a message with a context object', () => { - const logger = createLogger({ loggers }); - logger.info({ userId: 'abc123' }); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([{ userId: 'abc123' }]); - }); - - it('should merge context data from log call with existing context', () => { - const logger = createLogger({ loggers, context: { user: 'testUser' } }); - logger.info({ requestId: 'abc123' }); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([{ user: 'testUser', requestId: 'abc123' }]); - }); - - it('should use the provided template for logging', () => { - const logger = createLogger({ loggers, template: 'Custom Template' }); - logger.info('Template provided'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Template provided', {}]); - }); - - it('should log with an empty message', () => { - const logger = createLogger({ loggers }); - logger.info(''); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['', {}]); - }); - - it('should handle undefined values in the message template', () => { - const logger = createLogger({ loggers }); - logger.info`Value is ${undefined}`; - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([ - ['Value is ', ''], - undefined - ]); - }); - - it('should log with a default message if only context is provided', () => { - const logger = createLogger({ loggers }); - logger.info({ key: 'value' }); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual([{ key: 'value' }]); - }); - }); - - describe('Integration with LogTapeLogger', () => { - it('should call the correct logger based on the log level', () => { - const logger = createLogger({ loggers }); - logger.error('Error message'); - logger.warn('Warning message'); - logger.debug('Debug message'); - logger.fatal('Fatal message'); - assertSpyCalls(mockLogger.error, 1); - expect(mockLogger.error.calls[0].args).toEqual(['Error message', {}]); - assertSpyCalls(mockLogger.warn, 1); - expect(mockLogger.warn.calls[0].args).toEqual(['Warning message', {}]); - assertSpyCalls(mockLogger.debug, 1); - expect(mockLogger.debug.calls[0].args).toEqual(['Debug message', {}]); - assertSpyCalls(mockLogger.fatal, 1); - expect(mockLogger.fatal.calls[0].args).toEqual(['Fatal message', {}]); - }); - - it('should ensure log level functions are called with correct message and context', () => { - const logger = createLogger({ loggers, context: { sessionId: '123' } }); - logger.warn('This is a warning'); - assertSpyCalls(mockLogger.warn, 1); - expect(mockLogger.warn.calls[0].args).toEqual(['This is a warning', { sessionId: '123' }]); - }); - - it('should allow switching between loggers dynamically', () => { - const logger = createLogger({ loggers }); - logger.init.info('Initializing...'); - logger.build.debug('Building...'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Initializing...', {}]); - assertSpyCalls(mockLogger.debug, 1); - expect(mockLogger.debug.calls[0].args).toEqual(['Building...', {}]); - }); - - it('should call the correct logger getter based on dynamic switching', () => { - const logger = createLogger({ loggers }); - logger.init.info('Switched to init logger'); - assertSpyCalls(mockLogger.info, 1); - expect(mockLogger.info.calls[0].args).toEqual(['Switched to init logger', {}]); - logger.build.warn('Switched to build logger'); - assertSpyCalls(mockLogger.warn, 1); - expect(mockLogger.warn.calls[0].args).toEqual(['Switched to build logger', {}]); - }); - - it('should catch TypeScript errors for incorrect log level usage', () => { - const logger = createLogger({ loggers }); - expect(() => { - // Attempt to use a non-existent log level should trigger a TypeScript error - // @ts-expect-error nonExistentLogger - logger.init.verbose('This should fail'); - }).toThrow('logger.init.verbose is not a function'); - }); - }); -}); +// Test for split function +test("all")("...", async () => { + +}); \ No newline at end of file diff --git a/utils.ts b/utils.ts new file mode 100644 index 0000000..8c48f3b --- /dev/null +++ b/utils.ts @@ -0,0 +1,24 @@ +// Type guards +export function isPromiseLike(obj: any): obj is PromiseLike { + return typeof obj?.then === "function"; +} + +export function isAsyncGenerator(obj: any): obj is AsyncGenerator { + return typeof obj?.[Symbol.asyncIterator] === "function"; +} + +export function isGenerator(obj: any): obj is Generator { + return typeof obj?.[Symbol.iterator] === "function"; +} + +export function isAsyncIterable(obj: any): obj is AsyncIterable { + return typeof obj?.[Symbol.asyncIterator] === "function"; +} + +export function isIterable(obj: any): obj is Iterable { + return typeof obj?.[Symbol.iterator] === "function"; +} + +export function isAsyncIterator(obj: any): obj is AsyncIterator { + return typeof obj?.next === "function"; +} \ No newline at end of file