feat: Expose ReadableStream and make Blob more standardized (#4581)

Co-authored-by: crowlkats <crowlkats@gmail.com>

cli/js/globals.ts

@@ -18,6 +18,7 @@ import * as urlSearchParams from "./web/url_search_params.ts";
 import * as workers from "./web/workers.ts";
 import * as performanceUtil from "./web/performance.ts";
 import * as request from "./web/request.ts";
+import * as streams from "./web/streams/mod.ts";
 
 // These imports are not exposed and therefore are fine to just import the
 // symbols required.
@@ -226,6 +227,7 @@ export const windowOrWorkerGlobalScopeProperties = {
   FormData: nonEnumerable(formData.FormData),
   TextEncoder: nonEnumerable(textEncoding.TextEncoder),
   TextDecoder: nonEnumerable(textEncoding.TextDecoder),
+  ReadableStream: nonEnumerable(streams.ReadableStream),
   Request: nonEnumerable(request.Request),
   Response: nonEnumerable(fetchTypes.Response),
   performance: writable(new performanceUtil.Performance()),

cli/js/lib.deno.shared_globals.d.ts

@@ -34,6 +34,7 @@ declare interface WindowOrWorkerGlobalScope {
   FormData: __domTypes.FormDataConstructor;
   TextEncoder: typeof __textEncoding.TextEncoder;
   TextDecoder: typeof __textEncoding.TextDecoder;
+  ReadableStream: __domTypes.ReadableStreamConstructor;
   Request: __domTypes.RequestConstructor;
   Response: typeof __fetch.Response;
   performance: __performanceUtil.Performance;
@@ -250,6 +251,7 @@ declare const location: __domTypes.Location;
 declare const FormData: __domTypes.FormDataConstructor;
 declare const TextEncoder: typeof __textEncoding.TextEncoder;
 declare const TextDecoder: typeof __textEncoding.TextDecoder;
+declare const ReadableStream: __domTypes.ReadableStreamConstructor;
 declare const Request: __domTypes.RequestConstructor;
 declare const Response: typeof __fetch.Response;
 declare const performance: __performanceUtil.Performance;
@@ -282,6 +284,7 @@ declare type Headers = __domTypes.Headers;
 declare type FormData = __domTypes.FormData;
 declare type TextEncoder = __textEncoding.TextEncoder;
 declare type TextDecoder = __textEncoding.TextDecoder;
+declare type ReadableStream<R = any> = __domTypes.ReadableStream<R>;
 declare type Request = __domTypes.Request;
 declare type Response = __domTypes.Response;
 declare type Worker = __workers.Worker;
@@ -551,6 +554,27 @@ declare namespace __domTypes {
     preventClose?: boolean;
     signal?: AbortSignal;
   }
+  export interface UnderlyingSource<R = any> {
+    cancel?: ReadableStreamErrorCallback;
+    pull?: ReadableStreamDefaultControllerCallback<R>;
+    start?: ReadableStreamDefaultControllerCallback<R>;
+    type?: undefined;
+  }
+  export interface ReadableStreamErrorCallback {
+    (reason: any): void | PromiseLike<void>;
+  }
+
+  export interface ReadableStreamDefaultControllerCallback<R> {
+    (controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
+  }
+
+  export interface ReadableStreamDefaultController<R> {
+    readonly desiredSize: number;
+    enqueue(chunk?: R): void;
+    close(): void;
+    error(e?: any): void;
+  }
+
   /** 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. */
@@ -574,6 +598,12 @@ declare namespace __domTypes {
     */
     tee(): [ReadableStream<R>, ReadableStream<R>];
   }
+
+  export interface ReadableStreamConstructor<R = any> {
+    new (src?: UnderlyingSource<R>): ReadableStream<R>;
+    prototype: ReadableStream<R>;
+  }
+
   export interface ReadableStreamReader<R = any> {
     cancel(reason: any): Promise<void>;
     read(): Promise<ReadableStreamReadResult<R>>;
@@ -939,6 +969,9 @@ declare namespace __blob {
       options?: __domTypes.BlobPropertyBag
     );
     slice(start?: number, end?: number, contentType?: string): DenoBlob;
+    stream(): __domTypes.ReadableStream<Uint8Array>;
+    text(): Promise<string>;
+    arrayBuffer(): Promise<ArrayBuffer>;
   }
 }
 

cli/js/tests/blob_test.ts

@@ -1,5 +1,7 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 import { unitTest, assert, assertEquals } from "./test_util.ts";
+import { concat } from "../../../std/bytes/mod.ts";
+import { decode } from "../../../std/encoding/utf8.ts";
 
 unitTest(function blobString(): void {
   const b1 = new Blob(["Hello World"]);
@@ -67,4 +69,24 @@ unitTest(function nativeEndLine(): void {
   assertEquals(blob.size, Deno.build.os === "win" ? 12 : 11);
 });
 
-// TODO(qti3e) Test the stored data in a Blob after implementing FileReader API.
+unitTest(async function blobText(): Promise<void> {
+  const blob = new Blob(["Hello World"]);
+  assertEquals(await blob.text(), "Hello World");
+});
+
+unitTest(async function blobStream(): Promise<void> {
+  const blob = new Blob(["Hello World"]);
+  const stream = blob.stream();
+  assert(stream instanceof ReadableStream);
+  const reader = stream.getReader();
+  let bytes = new Uint8Array();
+  const read = async (): Promise<void> => {
+    const { done, value } = await reader.read();
+    if (!done && value) {
+      bytes = concat(bytes, value);
+      return read();
+    }
+  };
+  await read();
+  assertEquals(decode(bytes), "Hello World");
+});

cli/js/web/blob.ts

@@ -1,7 +1,8 @@
 // Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
 import * as domTypes from "./dom_types.ts";
-import { TextEncoder } from "./text_encoding.ts";
+import { TextDecoder, TextEncoder } from "./text_encoding.ts";
 import { build } from "../build.ts";
+import { ReadableStream } from "./streams/mod.ts";
 
 export const bytesSymbol = Symbol("bytes");
 
@@ -114,7 +115,6 @@ function processBlobParts(
     .reduce((a, b): number => a + b, 0);
   const ab = new ArrayBuffer(byteLength);
   const bytes = new Uint8Array(ab);
-
   let courser = 0;
   for (const u8 of uint8Arrays) {
     bytes.set(u8, courser);
@@ -124,6 +124,48 @@ function processBlobParts(
   return bytes;
 }
 
+function getStream(blobBytes: Uint8Array): domTypes.ReadableStream<Uint8Array> {
+  return new ReadableStream<Uint8Array>({
+    start: (
+      controller: domTypes.ReadableStreamDefaultController<Uint8Array>
+    ): void => {
+      controller.enqueue(blobBytes);
+      controller.close();
+    },
+  }) as domTypes.ReadableStream<Uint8Array>;
+}
+
+async function readBytes(
+  reader: domTypes.ReadableStreamReader<Uint8Array>
+): Promise<ArrayBuffer> {
+  const chunks: Uint8Array[] = [];
+  while (true) {
+    try {
+      const { done, value } = await reader.read();
+      if (!done && value instanceof Uint8Array) {
+        chunks.push(value);
+      } else if (done) {
+        const size = chunks.reduce((p, i) => p + i.byteLength, 0);
+        const bytes = new Uint8Array(size);
+        let offs = 0;
+        for (const chunk of chunks) {
+          bytes.set(chunk, offs);
+          offs += chunk.byteLength;
+        }
+        return Promise.resolve(bytes);
+      } else {
+        return Promise.reject(new TypeError());
+      }
+    } catch (e) {
+      return Promise.reject(e);
+    }
+  }
+}
+
+// A WeakMap holding blob to byte array mapping.
+// Ensures it does not impact garbage collection.
+export const blobBytesWeakMap = new WeakMap<domTypes.Blob, Uint8Array>();
+
 export class DenoBlob implements domTypes.Blob {
   [bytesSymbol]: Uint8Array;
   readonly size: number = 0;
@@ -167,4 +209,18 @@ export class DenoBlob implements domTypes.Blob {
       type: contentType || this.type,
     });
   }
+
+  stream(): domTypes.ReadableStream<Uint8Array> {
+    return getStream(this[bytesSymbol]);
+  }
+
+  async text(): Promise<string> {
+    const reader = getStream(this[bytesSymbol]).getReader();
+    const decoder = new TextDecoder();
+    return decoder.decode(await readBytes(reader));
+  }
+
+  arrayBuffer(): Promise<ArrayBuffer> {
+    return readBytes(getStream(this[bytesSymbol]).getReader());
+  }
 }

cli/js/web/dom_types.ts

@@ -277,6 +277,9 @@ export interface Blob {
   readonly size: number;
   readonly type: string;
   slice(start?: number, end?: number, contentType?: string): Blob;
+  stream(): ReadableStream;
+  text(): Promise<string>;
+  arrayBuffer(): Promise<ArrayBuffer>;
 }
 
 export interface Body {
@@ -317,6 +320,24 @@ export interface PipeOptions {
   signal?: AbortSignal;
 }
 
+export interface UnderlyingSource<R = any> {
+  cancel?: ReadableStreamErrorCallback;
+  pull?: ReadableStreamDefaultControllerCallback<R>;
+  start?: ReadableStreamDefaultControllerCallback<R>;
+  type?: undefined;
+}
+export interface ReadableStreamErrorCallback {
+  (reason: any): void | PromiseLike<void>;
+}
+
+export interface ReadableStreamDefaultControllerCallback<R> {
+  (controller: ReadableStreamDefaultController<R>): void | PromiseLike<void>;
+}
+
+export interface ReadableStreamConstructor {
+  new <R = any>(source?: UnderlyingSource<R>): ReadableStream<R>;
+}
+
 export interface ReadableStream<R = any> {
   readonly locked: boolean;
   cancel(reason?: any): Promise<void>;