refactor: move streams implementation to deno_web crate (#10935)

extensions/fetch/20_headers.js

@@ -5,7 +5,7 @@
 /// <reference path="../web/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
 /// <reference path="./internal.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/fetch/21_formdata.js

@@ -5,7 +5,7 @@
 /// <reference path="../web/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
 /// <reference path="./internal.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/fetch/22_body.js

@@ -7,7 +7,7 @@
 /// <reference path="../web/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
 /// <reference path="./internal.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/fetch/22_http_client.js

@@ -6,7 +6,7 @@
 /// <reference path="../url/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
 /// <reference path="./internal.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/fetch/23_request.js

@@ -5,7 +5,7 @@
 /// <reference path="../web/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
 /// <reference path="./internal.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/fetch/23_response.js

@@ -6,7 +6,7 @@
 /// <reference path="../url/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
 /// <reference path="./internal.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/fetch/26_fetch.js

@@ -5,7 +5,7 @@
 /// <reference path="../web/internal.d.ts" />
 /// <reference path="../url/internal.d.ts" />
 /// <reference path="../web/lib.deno_web.d.ts" />
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="../web/06_streams_types.d.ts" />
 /// <reference path="./internal.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />

extensions/fetch/internal.d.ts

@@ -52,11 +52,6 @@ declare namespace globalThis {
       declare function formDataFromEntries(entries: FormDataEntry[]): FormData;
     }
 
-    declare var streams: {
-      ReadableStream: typeof ReadableStream;
-      isReadableStreamDisturbed(stream: ReadableStream): boolean;
-    };
-
     declare namespace fetchBody {
       function mixinBody(
         prototype: any,

extensions/fetch/lib.deno_fetch.d.ts

@@ -16,277 +16,6 @@ interface DomIterable<K, V> {
   ): void;
 }
 
-interface ReadableStreamReadDoneResult<T> {
-  done: true;
-  value?: T;
-}
-
-interface ReadableStreamReadValueResult<T> {
-  done: false;
-  value: T;
-}
-
-type ReadableStreamReadResult<T> =
-  | ReadableStreamReadValueResult<T>
-  | ReadableStreamReadDoneResult<T>;
-
-interface ReadableStreamDefaultReader<R = any> {
-  readonly closed: Promise<void>;
-  cancel(reason?: any): Promise<void>;
-  read(): Promise<ReadableStreamReadResult<R>>;
-  releaseLock(): void;
-}
-
-declare var ReadableStreamDefaultReader: {
-  prototype: ReadableStreamDefaultReader;
-  new <R>(stream: ReadableStream<R>): ReadableStreamDefaultReader<R>;
-};
-
-interface ReadableStreamReader<R = any> {
-  cancel(): Promise<void>;
-  read(): Promise<ReadableStreamReadResult<R>>;
-  releaseLock(): void;
-}
-
-declare var ReadableStreamReader: {
-  prototype: ReadableStreamReader;
-  new (): ReadableStreamReader;
-};
-
-interface ReadableByteStreamControllerCallback {
-  (controller: ReadableByteStreamController): void | PromiseLike<void>;
-}
-
-interface UnderlyingByteSource {
-  autoAllocateChunkSize?: number;
-  cancel?: ReadableStreamErrorCallback;
-  pull?: ReadableByteStreamControllerCallback;
-  start?: ReadableByteStreamControllerCallback;
-  type: "bytes";
-}
-
-interface UnderlyingSink<W = any> {
-  abort?: WritableStreamErrorCallback;
-  close?: WritableStreamDefaultControllerCloseCallback;
-  start?: WritableStreamDefaultControllerStartCallback;
-  type?: undefined;
-  write?: WritableStreamDefaultControllerWriteCallback<W>;
-}
-
-interface UnderlyingSource<R = any> {
-  cancel?: ReadableStreamErrorCallback;
-  pull?: ReadableStreamDefaultControllerCallback<R>;
-  start?: ReadableStreamDefaultControllerCallback<R>;
-  type?: undefined;
-}
-
-interface ReadableStreamErrorCallback {
-  (reason: any): void | PromiseLike<void>;
-}
-
-interface ReadableStreamDefaultControllerCallback<R> {
-  (controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
-}
-
-interface ReadableStreamDefaultController<R = any> {
-  readonly desiredSize: number | null;
-  close(): void;
-  enqueue(chunk: R): void;
-  error(error?: any): void;
-}
-
-declare var ReadableStreamDefaultController: {
-  prototype: ReadableStreamDefaultController;
-  new (): ReadableStreamDefaultController;
-};
-
-interface ReadableByteStreamController {
-  readonly byobRequest: undefined;
-  readonly desiredSize: number | null;
-  close(): void;
-  enqueue(chunk: ArrayBufferView): void;
-  error(error?: any): void;
-}
-
-declare var ReadableByteStreamController: {
-  prototype: ReadableByteStreamController;
-  new (): ReadableByteStreamController;
-};
-
-interface PipeOptions {
-  preventAbort?: boolean;
-  preventCancel?: boolean;
-  preventClose?: boolean;
-  signal?: AbortSignal;
-}
-
-interface QueuingStrategySizeCallback<T = any> {
-  (chunk: T): number;
-}
-
-interface QueuingStrategy<T = any> {
-  highWaterMark?: number;
-  size?: QueuingStrategySizeCallback<T>;
-}
-
-/** This Streams API interface provides a built-in byte length queuing strategy
- * that can be used when constructing streams. */
-declare class CountQueuingStrategy implements QueuingStrategy {
-  constructor(options: { highWaterMark: number });
-  highWaterMark: number;
-  size(chunk: any): 1;
-}
-
-declare class ByteLengthQueuingStrategy
-  implements QueuingStrategy<ArrayBufferView> {
-  constructor(options: { highWaterMark: number });
-  highWaterMark: number;
-  size(chunk: ArrayBufferView): number;
-}
-
-/** This Streams API interface represents a readable stream of byte data. The
- * Fetch API offers a concrete instance of a ReadableStream through the body
- * property of a Response object. */
-interface ReadableStream<R = any> {
-  readonly locked: boolean;
-  cancel(reason?: any): Promise<void>;
-  /**
-   * @deprecated This is no longer part of the Streams standard and the async
-   *             iterable should be obtained by just using the stream as an
-   *             async iterator.
-   */
-  getIterator(options?: { preventCancel?: boolean }): AsyncIterableIterator<R>;
-  getReader(): ReadableStreamDefaultReader<R>;
-  pipeThrough<T>(
-    { writable, readable }: {
-      writable: WritableStream<R>;
-      readable: ReadableStream<T>;
-    },
-    options?: PipeOptions,
-  ): ReadableStream<T>;
-  pipeTo(dest: WritableStream<R>, options?: PipeOptions): Promise<void>;
-  tee(): [ReadableStream<R>, ReadableStream<R>];
-  [Symbol.asyncIterator](options?: {
-    preventCancel?: boolean;
-  }): AsyncIterableIterator<R>;
-}
-
-declare var ReadableStream: {
-  prototype: ReadableStream;
-  new (
-    underlyingSource: UnderlyingByteSource,
-    strategy?: { highWaterMark?: number; size?: undefined },
-  ): ReadableStream<Uint8Array>;
-  new <R = any>(
-    underlyingSource?: UnderlyingSource<R>,
-    strategy?: QueuingStrategy<R>,
-  ): ReadableStream<R>;
-};
-
-interface WritableStreamDefaultControllerCloseCallback {
-  (): void | PromiseLike<void>;
-}
-
-interface WritableStreamDefaultControllerStartCallback {
-  (controller: WritableStreamDefaultController): void | PromiseLike<void>;
-}
-
-interface WritableStreamDefaultControllerWriteCallback<W> {
-  (chunk: W, controller: WritableStreamDefaultController):
-    | void
-    | PromiseLike<
-      void
-    >;
-}
-
-interface WritableStreamErrorCallback {
-  (reason: any): void | PromiseLike<void>;
-}
-
-/** This Streams API interface provides a standard abstraction for writing
- * streaming data to a destination, known as a sink. This object comes with
- * built-in backpressure and queuing. */
-interface WritableStream<W = any> {
-  readonly locked: boolean;
-  abort(reason?: any): Promise<void>;
-  getWriter(): WritableStreamDefaultWriter<W>;
-}
-
-declare var WritableStream: {
-  prototype: WritableStream;
-  new <W = any>(
-    underlyingSink?: UnderlyingSink<W>,
-    strategy?: QueuingStrategy<W>,
-  ): WritableStream<W>;
-};
-
-/** This Streams API interface represents a controller allowing control of a
- * WritableStream's state. When constructing a WritableStream, the underlying
- * sink is given a corresponding WritableStreamDefaultController instance to
- * manipulate. */
-interface WritableStreamDefaultController {
-  error(error?: any): void;
-}
-
-/** This Streams API interface is the object returned by
- * WritableStream.getWriter() and once created locks the < writer to the
- * WritableStream ensuring that no other streams can write to the underlying
- * sink. */
-interface WritableStreamDefaultWriter<W = any> {
-  readonly closed: Promise<void>;
-  readonly desiredSize: number | null;
-  readonly ready: Promise<void>;
-  abort(reason?: any): Promise<void>;
-  close(): Promise<void>;
-  releaseLock(): void;
-  write(chunk: W): Promise<void>;
-}
-
-declare var WritableStreamDefaultWriter: {
-  prototype: WritableStreamDefaultWriter;
-  new (): WritableStreamDefaultWriter;
-};
-
-interface TransformStream<I = any, O = any> {
-  readonly readable: ReadableStream<O>;
-  readonly writable: WritableStream<I>;
-}
-
-declare var TransformStream: {
-  prototype: TransformStream;
-  new <I = any, O = any>(
-    transformer?: Transformer<I, O>,
-    writableStrategy?: QueuingStrategy<I>,
-    readableStrategy?: QueuingStrategy<O>,
-  ): TransformStream<I, O>;
-};
-
-interface TransformStreamDefaultController<O = any> {
-  readonly desiredSize: number | null;
-  enqueue(chunk: O): void;
-  error(reason?: any): void;
-  terminate(): void;
-}
-
-interface Transformer<I = any, O = any> {
-  flush?: TransformStreamDefaultControllerCallback<O>;
-  readableType?: undefined;
-  start?: TransformStreamDefaultControllerCallback<O>;
-  transform?: TransformStreamDefaultControllerTransformCallback<I, O>;
-  writableType?: undefined;
-}
-
-interface TransformStreamDefaultControllerCallback<O> {
-  (controller: TransformStreamDefaultController<O>): void | PromiseLike<void>;
-}
-
-interface TransformStreamDefaultControllerTransformCallback<I, O> {
-  (
-    chunk: I,
-    controller: TransformStreamDefaultController<O>,
-  ): void | PromiseLike<void>;
-}
-
 type FormDataEntryValue = File | string;
 
 /** Provides a way to easily construct a set of key/value pairs representing

extensions/fetch/lib.rs

@@ -61,7 +61,6 @@ pub fn init<P: FetchPermissions + 'static>(
     .js(include_js_files!(
       prefix "deno:extensions/fetch",
       "01_fetch_util.js",
-      "11_streams.js",
       "20_headers.js",
       "21_formdata.js",
       "22_body.js",

extensions/web/06_streams.js

@@ -1,7 +1,7 @@
 // Copyright 2018-2021 the Deno authors. All rights reserved. MIT license.
 
 // @ts-check
-/// <reference path="./11_streams_types.d.ts" />
+/// <reference path="./06_streams_types.d.ts" />
 /// <reference path="./lib.deno_fetch.d.ts" />
 /// <reference lib="esnext" />
 "use strict";

extensions/web/internal.d.ts

@@ -80,5 +80,10 @@ declare namespace globalThis {
         [globalThis.__bootstrap.file._byteSequence]: Uint8Array;
       };
     };
+
+    declare var streams: {
+      ReadableStream: typeof ReadableStream;
+      isReadableStreamDisturbed(stream: ReadableStream): boolean;
+    };
   }
 }

extensions/web/lib.deno_web.d.ts

@@ -377,3 +377,274 @@ declare class File extends Blob {
   readonly lastModified: number;
   readonly name: string;
 }
+
+interface ReadableStreamReadDoneResult<T> {
+  done: true;
+  value?: T;
+}
+
+interface ReadableStreamReadValueResult<T> {
+  done: false;
+  value: T;
+}
+
+type ReadableStreamReadResult<T> =
+  | ReadableStreamReadValueResult<T>
+  | ReadableStreamReadDoneResult<T>;
+
+interface ReadableStreamDefaultReader<R = any> {
+  readonly closed: Promise<void>;
+  cancel(reason?: any): Promise<void>;
+  read(): Promise<ReadableStreamReadResult<R>>;
+  releaseLock(): void;
+}
+
+declare var ReadableStreamDefaultReader: {
+  prototype: ReadableStreamDefaultReader;
+  new <R>(stream: ReadableStream<R>): ReadableStreamDefaultReader<R>;
+};
+
+interface ReadableStreamReader<R = any> {
+  cancel(): Promise<void>;
+  read(): Promise<ReadableStreamReadResult<R>>;
+  releaseLock(): void;
+}
+
+declare var ReadableStreamReader: {
+  prototype: ReadableStreamReader;
+  new (): ReadableStreamReader;
+};
+
+interface ReadableByteStreamControllerCallback {
+  (controller: ReadableByteStreamController): void | PromiseLike<void>;
+}
+
+interface UnderlyingByteSource {
+  autoAllocateChunkSize?: number;
+  cancel?: ReadableStreamErrorCallback;
+  pull?: ReadableByteStreamControllerCallback;
+  start?: ReadableByteStreamControllerCallback;
+  type: "bytes";
+}
+
+interface UnderlyingSink<W = any> {
+  abort?: WritableStreamErrorCallback;
+  close?: WritableStreamDefaultControllerCloseCallback;
+  start?: WritableStreamDefaultControllerStartCallback;
+  type?: undefined;
+  write?: WritableStreamDefaultControllerWriteCallback<W>;
+}
+
+interface UnderlyingSource<R = any> {
+  cancel?: ReadableStreamErrorCallback;
+  pull?: ReadableStreamDefaultControllerCallback<R>;
+  start?: ReadableStreamDefaultControllerCallback<R>;
+  type?: undefined;
+}
+
+interface ReadableStreamErrorCallback {
+  (reason: any): void | PromiseLike<void>;
+}
+
+interface ReadableStreamDefaultControllerCallback<R> {
+  (controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
+}
+
+interface ReadableStreamDefaultController<R = any> {
+  readonly desiredSize: number | null;
+  close(): void;
+  enqueue(chunk: R): void;
+  error(error?: any): void;
+}
+
+declare var ReadableStreamDefaultController: {
+  prototype: ReadableStreamDefaultController;
+  new (): ReadableStreamDefaultController;
+};
+
+interface ReadableByteStreamController {
+  readonly byobRequest: undefined;
+  readonly desiredSize: number | null;
+  close(): void;
+  enqueue(chunk: ArrayBufferView): void;
+  error(error?: any): void;
+}
+
+declare var ReadableByteStreamController: {
+  prototype: ReadableByteStreamController;
+  new (): ReadableByteStreamController;
+};
+
+interface PipeOptions {
+  preventAbort?: boolean;
+  preventCancel?: boolean;
+  preventClose?: boolean;
+  signal?: AbortSignal;
+}
+
+interface QueuingStrategySizeCallback<T = any> {
+  (chunk: T): number;
+}
+
+interface QueuingStrategy<T = any> {
+  highWaterMark?: number;
+  size?: QueuingStrategySizeCallback<T>;
+}
+
+/** This Streams API interface provides a built-in byte length queuing strategy
+ * that can be used when constructing streams. */
+declare class CountQueuingStrategy implements QueuingStrategy {
+  constructor(options: { highWaterMark: number });
+  highWaterMark: number;
+  size(chunk: any): 1;
+}
+
+declare class ByteLengthQueuingStrategy
+  implements QueuingStrategy<ArrayBufferView> {
+  constructor(options: { highWaterMark: number });
+  highWaterMark: number;
+  size(chunk: ArrayBufferView): number;
+}
+
+/** This Streams API interface represents a readable stream of byte data. The
+ * Fetch API offers a concrete instance of a ReadableStream through the body
+ * property of a Response object. */
+interface ReadableStream<R = any> {
+  readonly locked: boolean;
+  cancel(reason?: any): Promise<void>;
+  /**
+   * @deprecated This is no longer part of the Streams standard and the async
+   *             iterable should be obtained by just using the stream as an
+   *             async iterator.
+   */
+  getIterator(options?: { preventCancel?: boolean }): AsyncIterableIterator<R>;
+  getReader(): ReadableStreamDefaultReader<R>;
+  pipeThrough<T>(
+    { writable, readable }: {
+      writable: WritableStream<R>;
+      readable: ReadableStream<T>;
+    },
+    options?: PipeOptions,
+  ): ReadableStream<T>;
+  pipeTo(dest: WritableStream<R>, options?: PipeOptions): Promise<void>;
+  tee(): [ReadableStream<R>, ReadableStream<R>];
+  [Symbol.asyncIterator](options?: {
+    preventCancel?: boolean;
+  }): AsyncIterableIterator<R>;
+}
+
+declare var ReadableStream: {
+  prototype: ReadableStream;
+  new (
+    underlyingSource: UnderlyingByteSource,
+    strategy?: { highWaterMark?: number; size?: undefined },
+  ): ReadableStream<Uint8Array>;
+  new <R = any>(
+    underlyingSource?: UnderlyingSource<R>,
+    strategy?: QueuingStrategy<R>,
+  ): ReadableStream<R>;
+};
+
+interface WritableStreamDefaultControllerCloseCallback {
+  (): void | PromiseLike<void>;
+}
+
+interface WritableStreamDefaultControllerStartCallback {
+  (controller: WritableStreamDefaultController): void | PromiseLike<void>;
+}
+
+interface WritableStreamDefaultControllerWriteCallback<W> {
+  (chunk: W, controller: WritableStreamDefaultController):
+    | void
+    | PromiseLike<
+      void
+    >;
+}
+
+interface WritableStreamErrorCallback {
+  (reason: any): void | PromiseLike<void>;
+}
+
+/** This Streams API interface provides a standard abstraction for writing
+ * streaming data to a destination, known as a sink. This object comes with
+ * built-in backpressure and queuing. */
+interface WritableStream<W = any> {
+  readonly locked: boolean;
+  abort(reason?: any): Promise<void>;
+  getWriter(): WritableStreamDefaultWriter<W>;
+}
+
+declare var WritableStream: {
+  prototype: WritableStream;
+  new <W = any>(
+    underlyingSink?: UnderlyingSink<W>,
+    strategy?: QueuingStrategy<W>,
+  ): WritableStream<W>;
+};
+
+/** This Streams API interface represents a controller allowing control of a
+ * WritableStream's state. When constructing a WritableStream, the underlying
+ * sink is given a corresponding WritableStreamDefaultController instance to
+ * manipulate. */
+interface WritableStreamDefaultController {
+  error(error?: any): void;
+}
+
+/** This Streams API interface is the object returned by
+ * WritableStream.getWriter() and once created locks the < writer to the
+ * WritableStream ensuring that no other streams can write to the underlying
+ * sink. */
+interface WritableStreamDefaultWriter<W = any> {
+  readonly closed: Promise<void>;
+  readonly desiredSize: number | null;
+  readonly ready: Promise<void>;
+  abort(reason?: any): Promise<void>;
+  close(): Promise<void>;
+  releaseLock(): void;
+  write(chunk: W): Promise<void>;
+}
+
+declare var WritableStreamDefaultWriter: {
+  prototype: WritableStreamDefaultWriter;
+  new (): WritableStreamDefaultWriter;
+};
+
+interface TransformStream<I = any, O = any> {
+  readonly readable: ReadableStream<O>;
+  readonly writable: WritableStream<I>;
+}
+
+declare var TransformStream: {
+  prototype: TransformStream;
+  new <I = any, O = any>(
+    transformer?: Transformer<I, O>,
+    writableStrategy?: QueuingStrategy<I>,
+    readableStrategy?: QueuingStrategy<O>,
+  ): TransformStream<I, O>;
+};
+
+interface TransformStreamDefaultController<O = any> {
+  readonly desiredSize: number | null;
+  enqueue(chunk: O): void;
+  error(reason?: any): void;
+  terminate(): void;
+}
+
+interface Transformer<I = any, O = any> {
+  flush?: TransformStreamDefaultControllerCallback<O>;
+  readableType?: undefined;
+  start?: TransformStreamDefaultControllerCallback<O>;
+  transform?: TransformStreamDefaultControllerTransformCallback<I, O>;
+  writableType?: undefined;
+}
+
+interface TransformStreamDefaultControllerCallback<O> {
+  (controller: TransformStreamDefaultController<O>): void | PromiseLike<void>;
+}
+
+interface TransformStreamDefaultControllerTransformCallback<I, O> {
+  (
+    chunk: I,
+    controller: TransformStreamDefaultController<O>,
+  ): void | PromiseLike<void>;
+}

extensions/web/lib.rs

@@ -46,6 +46,7 @@ pub fn init(
       "03_abort_signal.js",
       "04_global_interfaces.js",
       "05_base64.js",
+      "06_streams.js",
       "08_text_encoding.js",
       "09_file.js",
       "10_filereader.js",