feat(streams/unstable): add AbortStream (#6708)

Author: BlackAsLight

streams/deno.json

@@ -3,6 +3,7 @@
   "version": "1.0.9",
   "exports": {
     ".": "./mod.ts",
+    "./unstable-abort-stream": "./unstable_abort_stream.ts",
     "./buffer": "./buffer.ts",
     "./byte-slice-stream": "./byte_slice_stream.ts",
     "./concat-readable-streams": "./concat_readable_streams.ts",

streams/unstable_abort_stream.ts

@@ -0,0 +1,48 @@
+// Copyright 2018-2025 the Deno authors. MIT license.
+// This module is browser compatible.
+
+/**
+ * A transform stream that accepts a {@linkcode AbortSignal} to easily abort a
+ * stream pipeThrough.
+ *
+ * @experimental **UNSTABLE**: New API, yet to be vetted.
+ *
+ * @typeparam T The type of the chunks passing through the AbortStream.
+ *
+ * @example Usage
+ * ```ts
+ * import { AbortStream } from "@std/streams/unstable-abort-stream";
+ * import { assertRejects } from "@std/assert/rejects";
+ *
+ * const controller = new AbortController();
+ * controller.abort(new Error("STOP"));
+ *
+ * await assertRejects(
+ *   async function () {
+ *     await new Response(
+ *       (await Deno.open("./deno.json"))
+ *         .readable
+ *         .pipeThrough(new AbortStream(controller.signal)),
+ *     )
+ *       .bytes();
+ *   },
+ *   Error,
+ *   "STOP",
+ * );
+ * ```
+ */
+export class AbortStream<T> extends TransformStream<T, T> {
+  /**
+   * Constructs a new instance.
+   *
+   * @param signal The {@linkcode AbortSignal}.
+   */
+  constructor(signal: AbortSignal) {
+    super({
+      transform(chunk, controller) {
+        if (signal.aborted) controller.error(signal.reason);
+        else controller.enqueue(chunk);
+      },
+    });
+  }
+}

streams/unstable_abort_stream_test.ts

@@ -0,0 +1,31 @@
+// Copyright 2018-2025 the Deno authors. MIT license.
+
+import { assertRejects } from "@std/assert";
+import { FixedChunkStream } from "@std/streams/unstable-fixed-chunk-stream";
+import { AbortStream } from "./unstable_abort_stream.ts";
+
+Deno.test("AbortStream", async () => {
+  const controller = new AbortController();
+
+  const file = "./deno.json";
+  const chunkSize = (await Deno.stat(file)).size / 4;
+  const promise = new Response(
+    (await Deno.open(file))
+      .readable
+      .pipeThrough(new FixedChunkStream(chunkSize)) // To make sure we aren't only given 1 chunk of the entire contents.
+      .pipeThrough( // To slow the process down for the controller.abort()
+        new TransformStream({
+          async transform(chunk, controller) {
+            await new Promise((a) => setTimeout(a, 50));
+            controller.enqueue(chunk);
+          },
+        }),
+      )
+      .pipeThrough(new AbortStream(controller.signal)),
+  )
+    .bytes();
+
+  setTimeout(() => controller.abort(), 75);
+
+  await assertRejects(() => promise);
+});