From 9843b2ec1806f42e4b75bacae30d7784a18c8c12 Mon Sep 17 00:00:00 2001 From: Okiki Ojo Date: Sat, 21 Mar 2026 04:12:06 -0400 Subject: [PATCH] feat: add cancellation handling and stream wrapping for operators Signed-off-by: Okiki Ojo --- helpers/_types.ts | 29 ++++++++ helpers/operators.ts | 164 +++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 189 insertions(+), 4 deletions(-) diff --git a/helpers/_types.ts b/helpers/_types.ts index 6eab603..5c114bf 100644 --- a/helpers/_types.ts +++ b/helpers/_types.ts @@ -1,6 +1,7 @@ // deno-lint-ignore-file no-explicit-any import type { ObservableError } from "../error.ts"; import type { Observable } from "../observable.ts"; +import type { SpecObservable } from "../_spec.ts"; /** * Type representing a stream operator function @@ -94,6 +95,34 @@ export interface BaseTransformOptions { name?: string; } +/** + * Structural readable/writable pair used by platform transforms such as + * `CompressionStream`. + */ +export interface StreamPair { + /** Readable side exposed by the transform-like object. */ + readable: ReadableStream; + /** Writable side exposed by the transform-like object. */ + writable: WritableStream; +} + +/** + * Observable-like values that can be converted through `Observable.from()`. + */ +export type ObservableInputLike = + | SpecObservable + | AsyncIterable + | Iterable + | PromiseLike; + +/** + * Foreign operator shape used by libraries that transform one Observable-like + * source into another Observable-like result. + */ +export type ObservableOperatorInterop = ( + source: SpecObservable, +) => ObservableInputLike; + // ======================================== // 2. CREATEOPERATOR INTERFACES // ======================================== diff --git a/helpers/operators.ts b/helpers/operators.ts index 2aab8b4..2cdea68 100644 --- a/helpers/operators.ts +++ b/helpers/operators.ts @@ -377,6 +377,7 @@ export function createOperator< const transform = (options as TransformFunctionOptions)?.transform; const start = (options as TransformFunctionOptions)?.start; const flush = (options as TransformFunctionOptions)?.flush; + const cancel = (options as TransformFunctionOptions)?.cancel; return (source) => { try { @@ -398,8 +399,10 @@ export function createOperator< { highWaterMark: 0 }, ); - // Pipe the source through the transform - return source.pipeThrough(transformStream); + // Pipe the source through the transform and bind downstream cancellation + return wrapStreamWithCancel(source.pipeThrough(transformStream), { + cancel: handleCancel(errorMode, cancel, { operatorName }), + }); } catch (err) { // If setup fails, return a stream that errors immediately return source.pipeThrough( @@ -754,6 +757,152 @@ export function handleFlush( }; } +/** + * Lifecycle handling for downstream cancellation cleanup. + * + * `flush()` runs when the source ends normally. `cancel()` runs when a consumer + * stops early, for example by unsubscribing, breaking out of `for await`, or + * otherwise cancelling the pipeline before the source completes. + * + * ```text + * source completes normally -> flush() + * consumer stops early -> cancel(reason) + * ``` + * + * That distinction matters for operators that hold timers, inner + * subscriptions, or abort controllers. Those resources should be released when + * the consumer goes away, even if the source never reached completion. + * + * Cleanup here is intentionally not treated like transform output. If the + * cleanup callback fails, the cancellation promise rejects, but no + * `ObservableError` is emitted into the stream because the consumer has already + * said it is no longer interested in more values. + * + * @typeParam T - Input chunk type + * @typeParam O - Output chunk type + * @typeParam S - State type (for stateful operators) + * @param _errorMode - Unused here. Cancellation cleanup does not route through + * the operator error modes because it is teardown, not part of the data path. + * @param cancel - Your cancellation handler (stateless or stateful, optional) + * @param context - Info about the operator, including name and state + * @returns A function suitable for a ReadableStream `cancel` hook, or undefined + */ +export function handleCancel( + _errorMode: OperatorErrorMode, + cancel?: + | TransformFunctionOptions["cancel"] + | StatefulTransformFunctionOptions["cancel"], + context: TransformHandlerContext = {}, +): ((reason?: unknown) => Promise) | undefined { + if (!cancel) return; + + const operatorName = context.operatorName || `operator:unknown`; + const isStateful = context.isStateful || false; + const state = context.state; + + return async function (reason?: unknown): Promise { + try { + if (isStateful) { + await (cancel as StatefulTransformFunctionOptions["cancel"])!( + state as S, + reason, + ); + return; + } + + await (cancel as TransformFunctionOptions["cancel"])!(reason); + } catch (err) { + throw ObservableError.from(err, `${operatorName}:cancel`, reason); + } + }; +} + +/** + * Wraps a transformed stream so operator-level cleanup runs when a downstream + * consumer stops early. + * + * This wrapper has a deliberately narrow job: it forwards readable-side + * cancellation to an operator cleanup callback while preserving the chunks + * coming from the wrapped stream. It does not try to redefine TransformStream + * semantics or reinterpret cancellation as a normal data-path error. + * + * ```text + * source -> TransformStream -> wrapped readable -> consumer + * | + * +-> cancel(reason) + * ``` + * + * @typeParam T - Output chunk type of the wrapped stream + * @param stream - The transformed stream to expose downstream + * @param options - Cancellation behavior for the wrapped stream + * @returns A readable stream that preserves output values and forwards cancel + */ +export function wrapStreamWithCancel( + stream: ReadableStream, + options: { + /** + * Cleanup to run if the consumer cancels the stream before it completes. + */ + cancel?: (reason?: unknown) => void | Promise; + } = {}, +): ReadableStream { + if (!options.cancel) return stream; + + const reader = stream.getReader(); + let settled = false; + + // The wrapper acquires an exclusive reader, so releasing it exactly once + // keeps teardown predictable even when both cancel and read completion race. + const release = (): void => { + if (settled) return; + settled = true; + + try { + reader.releaseLock(); + } catch { + // Releasing the reader is best-effort during teardown. + } + }; + + return new ReadableStream({ + async pull(controller) { + try { + // Forward chunks one-by-one so downstream backpressure still controls + // how quickly we read from the wrapped stream. + const { done, value } = await reader.read(); + + if (done) { + release(); + controller.close(); + return; + } + + controller.enqueue(value); + } catch (err) { + release(); + controller.error(err); + } + }, + + async cancel(reason) { + try { + // Run operator cleanup first so resources such as timers or inner + // subscriptions are released before we cancel the wrapped reader. + await options.cancel?.(reason); + } finally { + try { + // Cancelling the reader tells the wrapped stream that nobody wants + // more output. Any rejection here is surfaced through the returned + // cancellation promise instead of being emitted as a chunk. + await reader.cancel(reason); + } finally { + release(); + } + } + }, + }); +} + /** * Creates operators that maintain state across stream chunks * @@ -881,6 +1030,7 @@ export function createStatefulOperator< ?.transform; const start = (options as StatefulTransformFunctionOptions)?.start; const flush = (options as StatefulTransformFunctionOptions)?.flush; + const cancel = (options as StatefulTransformFunctionOptions)?.cancel; return (source) => { try { @@ -933,8 +1083,14 @@ export function createStatefulOperator< { highWaterMark: 0 }, ); - // Pipe the source through the transform - return source.pipeThrough(transformStream); + // Pipe the source through the transform and bind downstream cancellation + return wrapStreamWithCancel(source.pipeThrough(transformStream), { + cancel: handleCancel(errorMode, cancel, { + operatorName, + isStateful: true, + state, + }), + }); } catch (err) { // If setup fails, return a stream that errors immediately return source.pipeThrough( -- 2.51.2