From 8c7eff737a1bc6af9ba52e9046912cefd640e97c Mon Sep 17 00:00:00 2001 From: Nate Buckareff Date: Tue, 11 Nov 2025 11:12:37 -0800 Subject: [PATCH 1/3] Async iterator utils --- src/next/deferred.ts | 49 +++++++++++ src/next/split-async-iterable.ts | 32 +++++++ src/next/split-async-iterator.ts | 147 +++++++++++++++++++++++++++++++ 3 files changed, 228 insertions(+) create mode 100644 src/next/deferred.ts create mode 100644 src/next/split-async-iterable.ts create mode 100644 src/next/split-async-iterator.ts diff --git a/src/next/deferred.ts b/src/next/deferred.ts new file mode 100644 index 0000000..07966dd --- /dev/null +++ b/src/next/deferred.ts @@ -0,0 +1,49 @@ +export enum DeferredState { + PENDING = 0, + RESOLVED = 1, + REJECTED = 2, +} + +export class Deferred { + private _state = DeferredState.PENDING; + private _promise: Promise; + private _resolve!: (value: T) => void; + private _reject!: (reason?: unknown) => void; + + constructor() { + this._promise = new Promise((resolve, reject) => { + this._resolve = value => { + if (this._state === DeferredState.PENDING) { + this._state = DeferredState.RESOLVED; + resolve(value); + } + }; + this._reject = reason => { + if (this._state === DeferredState.PENDING) { + this._state = DeferredState.REJECTED; + reject(reason); + } + }; + }); + } + + get state(): DeferredState { + return this._state; + } + + get settled(): boolean { + return this._state !== DeferredState.PENDING; + } + + get promise(): Promise { + return this._promise; + } + + resolve(value: T): void { + this._resolve(value); + } + + reject(reason?: unknown): void { + this._reject(reason); + } +} diff --git a/src/next/split-async-iterable.ts b/src/next/split-async-iterable.ts new file mode 100644 index 0000000..9c1260b --- /dev/null +++ b/src/next/split-async-iterable.ts @@ -0,0 +1,32 @@ +import { SplitAsyncIterator } from './split-async-iterator.js'; + +export class SplitAsyncIterable + implements AsyncIterable +{ + private _iter: SplitAsyncIterator | null; + + constructor(source: AsyncIterator) { + this._iter = new SplitAsyncIterator(source); + } + + split(): [AsyncIterable, Promise] { + if (this._iter === null) { + throw Error('iterator already taken'); + } + const promise = this._iter.promise; + return [this, promise]; + } + + iter(): SplitAsyncIterator { + return this[Symbol.asyncIterator](); + } + + [Symbol.asyncIterator](): SplitAsyncIterator { + const { _iter } = this; + if (_iter === null) { + throw Error('iterator already taken'); + } + this._iter = null; + return _iter; + } +} diff --git a/src/next/split-async-iterator.ts b/src/next/split-async-iterator.ts new file mode 100644 index 0000000..28f8804 --- /dev/null +++ b/src/next/split-async-iterator.ts @@ -0,0 +1,147 @@ +import { Deferred } from './deferred.js'; + +export class IteratorCancelError extends Error { + constructor(message = 'iterator cancelled', options?: { cause?: unknown }) { + super(message, options); + this.name = 'IteratorCancelError'; + } +} + +export class SplitAsyncIterator + implements AsyncIterator +{ + private deferred: Deferred = new Deferred(); + private inFlight: boolean = false; + private completed?: { value: TReturn | undefined }; + + constructor(private source: AsyncIterator) {} + + get promise(): Promise { + return this.deferred.promise; + } + + private begin(): void { + if (this.inFlight) { + throw Error('concurrent iteration'); + } + this.inFlight = true; + } + + private end(): void { + this.inFlight = false; + } + + private complete(value: TReturn): void { + this.completed = { value }; + this.deferred.resolve(value); + } + + private completeFatally(error: unknown): void { + this.completed = { value: undefined }; + this.deferred.reject(error); + } + + next(...[value]: [] | [TNext]): Promise> { + return this._next(value as TNext); + } + + private async _next(value: TNext): Promise> { + if (this.completed) { + return { + done: true, + value: this.completed.value as TReturn, + }; + } + + this.begin(); + try { + const result = await this.source.next(value); + if (result.done) { + this.complete(result.value); + } + return result; + } catch (error) { + this.completeFatally(error); + throw error; + } finally { + this.end(); + } + } + + async throw(e?: unknown): Promise> { + if (this.completed) { + return { + done: true, + value: this.completed.value as TReturn, + }; + } + + this.begin(); + if (this.source.throw) { + try { + const result = await this.source.throw(e); + if (result.done) { + this.complete(result.value); + } + return result; + } catch (error) { + this.completeFatally(error); + throw error; + } finally { + this.end(); + } + } else { + const reason = + e === undefined + ? new IteratorCancelError('iterator cancelled via throw()') + : e; + + try { + this.completeFatally(reason); + } finally { + this.end(); + } + throw reason; + } + } + + async return( + value?: TReturn | PromiseLike, + ): Promise> { + if (this.completed) { + return { + done: true, + value: this.completed.value as TReturn, + }; + } + + this.begin(); + if (this.source.return) { + try { + let result = await this.source.return(value); + while (!result.done) { + result = await this.source.next(undefined as TNext); + } + const v = result.value; + this.complete(v); + return { done: true, value: v }; + } catch (error) { + this.completeFatally(error); + throw error; + } finally { + this.end(); + } + } else { + try { + const v = (await value) as TReturn; + this.complete(v); + return { done: true, value: v }; + } catch (error) { + this.completeFatally(error); + throw error; + } finally { + this.end(); + } + } + } +} From 0b80a6bbeb593ff56ad0c49aed29fde71b0e9bf4 Mon Sep 17 00:00:00 2001 From: Nate Buckareff Date: Tue, 11 Nov 2025 16:19:05 -0800 Subject: [PATCH 2/3] Flesh out persistent operations client API --- src/api.ts | 50 ++++++++- src/next/batch-scheduler-type.ts | 2 + src/next/client-dev.ts | 143 ++++++++++++++++++++++++++ src/next/client-prod.ts | 82 +++++++++++++++ src/next/client-type.ts | 53 ++++++++++ src/next/client.ts | 14 +++ src/next/plan-builder.ts | 28 +++++ src/next/proxy.ts | 171 +++++++++++++++++++++++++++++++ src/next/transport.ts | 22 ++++ src/rpc-type.ts | 1 + src/server.ts | 1 + 11 files changed, 565 insertions(+), 2 deletions(-) create mode 100644 src/next/batch-scheduler-type.ts create mode 100644 src/next/client-dev.ts create mode 100644 src/next/client-prod.ts create mode 100644 src/next/client-type.ts create mode 100644 src/next/client.ts create mode 100644 src/next/plan-builder.ts create mode 100644 src/next/proxy.ts create mode 100644 src/next/transport.ts diff --git a/src/api.ts b/src/api.ts index 5a9bf80..6c45d15 100644 --- a/src/api.ts +++ b/src/api.ts @@ -1,6 +1,13 @@ -import { type AnyCodec, tk, type VoidCodec } from 'typekind'; +import { type AnyCodec, TupleCodec, tk, type VoidCodec } from 'typekind'; -// TODO: handler/stream naming could be improved +// TODO: handler/stream naming could be improved - use method as general name +// for handler or stream ie service have methods + +export type AnyApi = + | AnyRouterApi + | AnyServiceApi + | AnyHandlerApi + | AnyStreamApi; export type AnyRouterApi = RouterApi; export type AnyServiceApi = ServiceApi; @@ -30,6 +37,10 @@ export class HandlerApi { public readonly input: Input, public readonly output: Output, ) {} + + getParamCodec(index: number): AnyCodec { + return getParamCodec(this, index); + } } export class StreamApi< @@ -44,8 +55,43 @@ export class StreamApi< public readonly value: Value, public readonly output: Output, ) {} + + getParamCodec(index: number): AnyCodec { + return getParamCodec(this, index); + } +} + +function getParamCodec( + api: AnyHandlerApi | AnyStreamApi, + index: number, +): AnyCodec { + const inputCodec = api.input; + + if (inputCodec instanceof TupleCodec) { + const argCodec = (inputCodec as TupleCodec).codecs[index]; + if (argCodec === undefined) { + throw Error('rpc method argument index out of bounds'); + } + return argCodec; + } + + if (index !== 0) { + throw Error('rpc method argument index out of bounds'); + } + + return inputCodec; } +export type InferApiType = T extends AnyRouterApi + ? InferRouterType + : T extends AnyServiceApi + ? InferServiceType + : T extends AnyHandlerApi + ? InferHandlerType + : T extends AnyStreamApi + ? InferStreamType + : never; + export type InferRouterType = { [K in keyof T['routes']]: T['routes'][K] extends AnyServiceApi ? InferServiceType diff --git a/src/next/batch-scheduler-type.ts b/src/next/batch-scheduler-type.ts new file mode 100644 index 0000000..3d882a3 --- /dev/null +++ b/src/next/batch-scheduler-type.ts @@ -0,0 +1,2 @@ +// TODO +export type IBatchScheduler = unknown; diff --git a/src/next/client-dev.ts b/src/next/client-dev.ts new file mode 100644 index 0000000..f2c837e --- /dev/null +++ b/src/next/client-dev.ts @@ -0,0 +1,143 @@ +/** biome-ignore-all lint/suspicious/noExplicitAny: need any for inference */ + +import type { AnyRouterApi, InferRouterType } from '../api.js'; +import { HandlerApi, StreamApi } from '../api.js'; +import type { Rpc } from '../rpc-type.js'; +import type { + ClientNextConfig, + IClient, + InferOperationResult, + OperationBuilderReturn, +} from './client-type.js'; +import { type OpId, PlanBuilder } from './plan-builder.js'; +import { isRpcProxy, ProxyState, unwrapOpProxy } from './proxy.js'; +import type { TransportPlanRequest } from './transport.js'; + +export type DefineCallback< + Router extends AnyRouterApi, + Params extends Record, + Return extends OperationBuilderReturn, +> = (api: Rpc>, params: Params) => Return; + +export class ClientDev implements IClient { + private definedOperations: string[] = []; + + constructor(public readonly config: ClientNextConfig) {} + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + callback: DefineCallback, + ): (...args: Args) => InferOperationResult; + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + keys: () => (keyof Params)[], + callback: DefineCallback, + ): (...args: Args) => InferOperationResult; + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + callbackOrKeysArg: + | DefineCallback + | (() => (keyof Params)[]), + callbackArg?: DefineCallback, + ): (...args: Args) => InferOperationResult { + const getKeys = + callbackArg === undefined + ? undefined + : (callbackOrKeysArg as () => (keyof Params)[]); + + const callback = callbackArg + ? (callbackArg as DefineCallback) + : (callbackOrKeysArg as DefineCallback); + + if (this.definedOperations.includes(name)) { + throw Error('operation already defined'); + } + + const operation = (...args: Args): InferOperationResult => { + const params = prepare(...args); + const builder = new PlanBuilder(); + const api = this.config.router; + const proxy = ProxyState.createProxy(undefined, builder, api); + // set T to unknown to prevent everything from collapsing to any and + // maintain some type safety here + const result = callback(proxy, params) as OperationBuilderReturn; + + let streams: TransportPlanRequest['streams']; + let outputs: TransportPlanRequest['outputs']; + + if (isRpcProxy(result)) { + const { api, op } = unwrapOpProxy(result); + + if (api instanceof HandlerApi) { + outputs = op.id; + } else if (api instanceof StreamApi) { + streams = op.id; + } else { + throw Error('expected handler or stream'); + } + } else { + let s: Record | undefined; + let o: Record | undefined; + + for (const [key, value] of Object.entries(result)) { + const { api, op } = unwrapOpProxy(value); + + if (api instanceof HandlerApi) { + o ??= {}; + o[key] = op.id; + } else if (api instanceof StreamApi) { + s ??= {}; + s[key] = op.id; + } else { + throw Error('expected handler or stream'); + } + } + + if (s) streams = s; + if (o) outputs = o; + } + + const keys = getKeys?.() as string[] | undefined; + + const request: TransportPlanRequest = { + operation: name, + params, + keys, + plan: {}, // XXX + streams, + outputs, + }; + + throw Error('todo: send to transport'); + }; + + this.definedOperations.push(name); + + return operation; + } + + get api(): Rpc> { + // TODO: create a proxy to construct a single, unbatched request + + // TODO: how to tree-shake + + throw Error('todo: implement unbatched requests'); + } +} diff --git a/src/next/client-prod.ts b/src/next/client-prod.ts new file mode 100644 index 0000000..f7b9825 --- /dev/null +++ b/src/next/client-prod.ts @@ -0,0 +1,82 @@ +/** biome-ignore-all lint/suspicious/noExplicitAny: need any for inference */ + +import type { AnyRouterApi, InferRouterType } from '../api.js'; +import type { Rpc } from '../rpc-type.js'; +import type { + ClientNextConfig, + IClient, + InferOperationResult, + OperationBuilderReturn, +} from './client-type.js'; +import type { TransportPersistentRequest } from './transport.js'; + +export type DefineCallback< + Router extends AnyRouterApi, + Params extends Record, + Return extends OperationBuilderReturn, +> = (api: Rpc>, params: Params) => Return; + +export class ClientProd + implements IClient +{ + private operations: Record any> = {}; + + constructor(public readonly config: ClientNextConfig) {} + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + callback: DefineCallback, + ): (...args: Args) => InferOperationResult; + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + keys: () => (keyof Params)[], + callback: DefineCallback, + ): (...args: Args) => InferOperationResult; + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + _callbackOrKeysArg: + | DefineCallback + | (() => (keyof Params)[]), + _callbackArg?: DefineCallback, + ): (...args: Args) => InferOperationResult { + let operation = this.operations[name]; + + if (operation) { + return operation; + } + + operation = (...args: Args): InferOperationResult => { + const params = prepare(...args); + const request: TransportPersistentRequest = { + operation: name, + params, + }; + throw Error('todo: send to transport'); + }; + + this.operations[name] = operation; + + return operation; + } + + get api(): Rpc> { + throw Error('todo: implement unbatched requests'); + } +} diff --git a/src/next/client-type.ts b/src/next/client-type.ts new file mode 100644 index 0000000..0a0be8e --- /dev/null +++ b/src/next/client-type.ts @@ -0,0 +1,53 @@ +/** biome-ignore-all lint/suspicious/noExplicitAny: need any for inference */ + +import type { Simplify } from 'typekind'; +import type { AnyRouterApi, InferRouterType } from '../api.js'; +import type { Rpc } from '../rpc-type.js'; +import type { DefineCallback } from './client-dev.js'; +import type { SplitAsyncIterable } from './split-async-iterable.js'; + +// need any to correctly infer return types at usage, but set T to unknown for +// typing internal code +export type OperationBuilderReturn = Rpc | Record>; + +export type InferOperationOutput = T extends Rpc + ? U extends AsyncGenerator + ? SplitAsyncIterable + : Promise + : never; + +export type InferOperationResult = T extends Rpc + ? U extends AsyncGenerator + ? SplitAsyncIterable + : Promise + : Simplify<{ [K in keyof T]: InferOperationOutput }>; + +export interface ClientNextConfig { + router: Router; +} + +export interface IClient { + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + callback: DefineCallback, + ): (...args: Args) => InferOperationResult; + + define< + Args extends any[], + Params extends Record, + Return extends OperationBuilderReturn, + >( + name: string, + prepare: (...args: Args) => Params, + keys: () => (keyof Params)[], + callback: DefineCallback, + ): (...args: Args) => InferOperationResult; + + // TODO: will probably remove for tree-shaking + get api(): Rpc>; +} diff --git a/src/next/client.ts b/src/next/client.ts new file mode 100644 index 0000000..9af4da4 --- /dev/null +++ b/src/next/client.ts @@ -0,0 +1,14 @@ +import type { AnyRouterApi } from '../api.js'; +import { ClientDev } from './client-dev.js'; +import { ClientProd } from './client-prod.js'; +import type { ClientNextConfig, IClient } from './client-type.js'; + +export function createClient( + config: ClientNextConfig, +): IClient { + if (import.meta.env.DEV) { + return new ClientDev(config); + } else { + return new ClientProd(config); + } +} diff --git a/src/next/plan-builder.ts b/src/next/plan-builder.ts new file mode 100644 index 0000000..f1b3c3f --- /dev/null +++ b/src/next/plan-builder.ts @@ -0,0 +1,28 @@ +import type { Json } from 'typekind'; + +export type Op = + | { kind: 'root'; id: 0 } + | { kind: 'get'; id: number; target: OpTarget } + | { kind: 'data'; id: number; data: Json } + | { kind: 'apply'; id: number; target: OpTarget; args: OpId[] }; + +export type OpId = number | string; +export type OpTarget = [OpId, ...string[]]; + +export class PlanBuilder { + private ops: Map = new Map(); + + constructor() { + this.ops.set(0, { kind: 'root', id: 0 }); + } + + get root(): Extract { + return this.ops.get(0) as Extract; + } + + addOp(op: Op): Op { + op.id = this.ops.size; + this.ops.set(op.id, op); + return op; + } +} diff --git a/src/next/proxy.ts b/src/next/proxy.ts new file mode 100644 index 0000000..fae0c7a --- /dev/null +++ b/src/next/proxy.ts @@ -0,0 +1,171 @@ +import type { AnyApi, AnyRouterApi, InferRouterType } from '../api.js'; +import { HandlerApi, RouterApi, ServiceApi, StreamApi } from '../api.js'; +import type { Rpc } from '../rpc-type.js'; +import type { Op, OpId, OpTarget, PlanBuilder } from './plan-builder.js'; + +const nothing = () => {}; +const proxyStateSymbol = Symbol('proxyState'); + +export function isRpcProxy(value: unknown): value is Rpc { + return isOpProxy(value); +} + +export function isOpProxy(value: unknown): value is OpProxy { + return typeof value === 'function' && proxyStateSymbol in value; +} + +export function unwrapOpProxy(value: Rpc): ProxyState; +export function unwrapOpProxy(value: OpProxy): ProxyState; +export function unwrapOpProxy(value: unknown): ProxyState | undefined; +export function unwrapOpProxy(value: unknown): ProxyState | undefined { + if (isOpProxy(value)) { + return value[proxyStateSymbol]; + } +} + +export interface OpProxy { + [proxyStateSymbol]: ProxyState; +} + +export class ProxyState { + // private cachedPromise?: Promise; // TODO + // private cachedGenerator?: unknown; // TODO + + private constructor( + public sched: {} | undefined, // TODO + public builder: PlanBuilder, + public api: AnyApi, + public op: Op, + ) {} + + static createProxy( + sched: {} | undefined, // TODO + builder: PlanBuilder, + router: Router, + ): Rpc> { + const root: Op = { kind: 'root', id: 0 }; + const state = new ProxyState(sched, builder, router, root); + return createProxy>(state); + } + + createProxy(op: Op): Rpc; + createProxy(api: AnyApi, op: Op): Rpc; + createProxy(opOrApi: Op | AnyApi, opArg?: Op): Rpc { + const api = opArg === undefined ? this.api : (opOrApi as AnyApi); + const op = opArg === undefined ? (opOrApi as Op) : opArg; + const clone = new ProxyState(this.sched, this.builder, api, op); + return createProxy(clone); + } +} + +function createProxy(state: ProxyState): Rpc { + return new Proxy(nothing as Rpc, { + has(_target, p) { + return trapHas(p); + }, + get(_target, p) { + return trapGet(state, p); + }, + apply(_target, _thisArg, argArray) { + return trapApply(state, argArray); + }, + }); +} + +function trapHas(p: string | symbol): boolean { + return p === proxyStateSymbol; +} + +function trapGet(state: ProxyState, p: string | symbol): unknown { + if (p === proxyStateSymbol) { + return state; + } + + if (p === 'then') { + if (!state.sched) { + throw Error('no batch scheduler'); + } + + if (state.op.id === -1) { + state.op = state.builder.addOp(state.op); + } + + // cachedPromise ??= state.sched?.schedulePromise(state.op.id); return + // cachedPromise.then.bind(cachedPromise); + } + + // TODO: handle generators + + if (typeof p !== 'string') { + throw Error('invalid property'); + } + + const target: OpTarget = + state.op.kind === 'get' && state.op.id === -1 + ? [...state.op.target, p] + : [state.op.id, p]; + + let api: AnyApi; + + if (state.api instanceof RouterApi) { + const route = state.api.routes[p]; + if (route === undefined) { + throw Error(`route not found`); + } + api = route; + } else if (state.api instanceof ServiceApi) { + const method = state.api.handlers[p]; + if (method === undefined) { + throw Error(`method not found`); + } + api = method; + } else { + throw Error(`cannot get property of ${state.api.constructor.name}`); + } + + return state.createProxy(api, { + kind: 'get', + id: -1, + target, + }); +} + +function trapApply(state: ProxyState, argArray: unknown[]): unknown { + const args: OpId[] = []; + + // TODO: handle .map + + if (!(state.api instanceof HandlerApi || state.api instanceof StreamApi)) { + throw Error('apply target is not a service method'); + } + + for (let i = 0; i < argArray.length; i++) { + const arg = argArray[i]; + const codec = state.api.getParamCodec(i); + + if (isOpProxy(arg)) { + args.push(arg[proxyStateSymbol].op.id); + } else { + const argOp = state.builder.addOp({ + kind: 'data', + id: -1, + data: codec.serialize(arg), // TODO: ref context + }); + args.push(argOp.id); + } + } + + const target: OpTarget = + state.op.kind === 'get' && state.op.id === -1 + ? state.op.target + : [state.op.id]; + + const applyOp = state.builder.addOp({ + kind: 'apply', + id: -1, + target, + args, + }); + + return state.createProxy(applyOp); +} diff --git a/src/next/transport.ts b/src/next/transport.ts new file mode 100644 index 0000000..153526e --- /dev/null +++ b/src/next/transport.ts @@ -0,0 +1,22 @@ +import type { OpId } from './plan-builder.js'; + +// TODO: split into transport-client.ts and transport-server.ts? + +export interface TransportPlanRequest { + operation: string; + params: Record; + keys?: string[]; + plan: unknown; // TODO: SerializedPlan; + streams?: OpId | Record; + outputs?: OpId | Record; +} + +export interface TransportPersistentRequest { + operation: string; + params: Record; +} + +export interface TransportUnbatchedRequest { + operation: `rpc#${string}`; + params: Record; +} diff --git a/src/rpc-type.ts b/src/rpc-type.ts index b44e663..15e77b1 100644 --- a/src/rpc-type.ts +++ b/src/rpc-type.ts @@ -19,6 +19,7 @@ export type IsPlainPrimitive = T extends ? T : never; +// TODO: router object should not be mappable export type Mappable = { map(callback: (value: Rpc) => U): Rpc; }; diff --git a/src/server.ts b/src/server.ts index 465fada..5c6a475 100644 --- a/src/server.ts +++ b/src/server.ts @@ -16,6 +16,7 @@ export type InferRouter = { : InferRouter>; }; +// TODO: poorly named...this is really InferServiceType export type InferApi = { [K in keyof T['handlers']]?: T['handlers'][K] extends HandlerApi< infer Input, From e5d35e40d3d387e8d07e46d54d5fe43e8d95e221 Mon Sep 17 00:00:00 2001 From: Nate Buckareff Date: Wed, 12 Nov 2025 21:37:12 -0800 Subject: [PATCH 3/3] Replace SplitAsyncIterator with `yield*` --- src/next/split-async-iterable.ts | 42 +++++---- src/next/split-async-iterator.ts | 147 ------------------------------- 2 files changed, 26 insertions(+), 163 deletions(-) delete mode 100644 src/next/split-async-iterator.ts diff --git a/src/next/split-async-iterable.ts b/src/next/split-async-iterable.ts index 9c1260b..dbbc62b 100644 --- a/src/next/split-async-iterable.ts +++ b/src/next/split-async-iterable.ts @@ -1,32 +1,42 @@ -import { SplitAsyncIterator } from './split-async-iterator.js'; +import { Deferred } from './deferred.js'; export class SplitAsyncIterable implements AsyncIterable { - private _iter: SplitAsyncIterator | null; + private deferred: Deferred = new Deferred(); + private iterator?: AsyncIterator; - constructor(source: AsyncIterator) { - this._iter = new SplitAsyncIterator(source); - } + constructor(source: AsyncGenerator) { + const { deferred } = this; - split(): [AsyncIterable, Promise] { - if (this._iter === null) { - throw Error('iterator already taken'); + async function* wrapper(): AsyncGenerator { + try { + const result = yield* source; + deferred.resolve(result); + return result; + } catch (error) { + deferred.reject(error); + throw error; + } } - const promise = this._iter.promise; - return [this, promise]; + + this.iterator = wrapper()[Symbol.asyncIterator](); + } + + split(): [SplitAsyncIterable, Promise] { + return [this, this.deferred.promise]; } - iter(): SplitAsyncIterator { + iter(): AsyncIterator { return this[Symbol.asyncIterator](); } - [Symbol.asyncIterator](): SplitAsyncIterator { - const { _iter } = this; - if (_iter === null) { + [Symbol.asyncIterator](): AsyncIterator { + const { iterator } = this; + if (iterator === undefined) { throw Error('iterator already taken'); } - this._iter = null; - return _iter; + this.iterator = undefined; + return iterator; } } diff --git a/src/next/split-async-iterator.ts b/src/next/split-async-iterator.ts deleted file mode 100644 index 28f8804..0000000 --- a/src/next/split-async-iterator.ts +++ /dev/null @@ -1,147 +0,0 @@ -import { Deferred } from './deferred.js'; - -export class IteratorCancelError extends Error { - constructor(message = 'iterator cancelled', options?: { cause?: unknown }) { - super(message, options); - this.name = 'IteratorCancelError'; - } -} - -export class SplitAsyncIterator - implements AsyncIterator -{ - private deferred: Deferred = new Deferred(); - private inFlight: boolean = false; - private completed?: { value: TReturn | undefined }; - - constructor(private source: AsyncIterator) {} - - get promise(): Promise { - return this.deferred.promise; - } - - private begin(): void { - if (this.inFlight) { - throw Error('concurrent iteration'); - } - this.inFlight = true; - } - - private end(): void { - this.inFlight = false; - } - - private complete(value: TReturn): void { - this.completed = { value }; - this.deferred.resolve(value); - } - - private completeFatally(error: unknown): void { - this.completed = { value: undefined }; - this.deferred.reject(error); - } - - next(...[value]: [] | [TNext]): Promise> { - return this._next(value as TNext); - } - - private async _next(value: TNext): Promise> { - if (this.completed) { - return { - done: true, - value: this.completed.value as TReturn, - }; - } - - this.begin(); - try { - const result = await this.source.next(value); - if (result.done) { - this.complete(result.value); - } - return result; - } catch (error) { - this.completeFatally(error); - throw error; - } finally { - this.end(); - } - } - - async throw(e?: unknown): Promise> { - if (this.completed) { - return { - done: true, - value: this.completed.value as TReturn, - }; - } - - this.begin(); - if (this.source.throw) { - try { - const result = await this.source.throw(e); - if (result.done) { - this.complete(result.value); - } - return result; - } catch (error) { - this.completeFatally(error); - throw error; - } finally { - this.end(); - } - } else { - const reason = - e === undefined - ? new IteratorCancelError('iterator cancelled via throw()') - : e; - - try { - this.completeFatally(reason); - } finally { - this.end(); - } - throw reason; - } - } - - async return( - value?: TReturn | PromiseLike, - ): Promise> { - if (this.completed) { - return { - done: true, - value: this.completed.value as TReturn, - }; - } - - this.begin(); - if (this.source.return) { - try { - let result = await this.source.return(value); - while (!result.done) { - result = await this.source.next(undefined as TNext); - } - const v = result.value; - this.complete(v); - return { done: true, value: v }; - } catch (error) { - this.completeFatally(error); - throw error; - } finally { - this.end(); - } - } else { - try { - const v = (await value) as TReturn; - this.complete(v); - return { done: true, value: v }; - } catch (error) { - this.completeFatally(error); - throw error; - } finally { - this.end(); - } - } - } -}