1 files changed, 76 insertions, 1 deletions

diff --git a/cli/js/body.ts b/cli/js/body.ts
index 6567b1934..e00cd30b9 100644
--- a/cli/js/body.ts
+++ b/cli/js/body.ts
@@ -3,6 +3,7 @@ import * as blob from "./blob.ts";
 import * as encoding from "./text_encoding.ts";
 import * as headers from "./headers.ts";
 import * as domTypes from "./dom_types.ts";
+import { ReadableStream } from "./streams/mod.ts";
 
 const { Headers } = headers;
 
@@ -12,6 +13,13 @@ const { TextEncoder, TextDecoder } = encoding;
 const Blob = blob.DenoBlob;
 const DenoBlob = blob.DenoBlob;
 
+type ReadableStreamReader = domTypes.ReadableStreamReader;
+
+interface ReadableStreamController {
+  enqueue(chunk: string | ArrayBuffer): void;
+  close(): void;
+}
+
 export type BodySource =
   | domTypes.Blob
   | domTypes.BufferSource
@@ -37,6 +45,8 @@ function validateBodyType(owner: Body, bodySource: BodySource): boolean {
     return true;
   } else if (typeof bodySource === "string") {
     return true;
+  } else if (bodySource instanceof ReadableStream) {
+    return true;
   } else if (bodySource instanceof FormData) {
     return true;
   } else if (!bodySource) {
@@ -47,6 +57,58 @@ function validateBodyType(owner: Body, bodySource: BodySource): boolean {
   );
 }
 
+function concatenate(...arrays: Uint8Array[]): ArrayBuffer {
+  let totalLength = 0;
+  for (const arr of arrays) {
+    totalLength += arr.length;
+  }
+  const result = new Uint8Array(totalLength);
+  let offset = 0;
+  for (const arr of arrays) {
+    result.set(arr, offset);
+    offset += arr.length;
+  }
+  return result.buffer as ArrayBuffer;
+}
+
+function bufferFromStream(stream: ReadableStreamReader): Promise<ArrayBuffer> {
+  return new Promise(
+    (resolve, reject): void => {
+      const parts: Uint8Array[] = [];
+      const encoder = new TextEncoder();
+      // recurse
+      (function pump(): void {
+        stream
+          .read()
+          .then(
+            ({ done, value }): void => {
+              if (done) {
+                return resolve(concatenate(...parts));
+              }
+
+              if (typeof value === "string") {
+                parts.push(encoder.encode(value));
+              } else if (value instanceof ArrayBuffer) {
+                parts.push(new Uint8Array(value));
+              } else if (!value) {
+                // noop for undefined
+              } else {
+                reject("unhandled type on stream read");
+              }
+
+              return pump();
+            }
+          )
+          .catch(
+            (err): void => {
+              reject(err);
+            }
+          );
+      })();
+    }
+  );
+}
+
 function getHeaderValueParams(value: string): Map<string, string> {
   const params = new Map();
   // Forced to do so for some Map constructor param mismatch
@@ -81,8 +143,18 @@ export class Body implements domTypes.Body {
     if (this._stream) {
       return this._stream;
     }
+
+    if (this._bodySource instanceof ReadableStream) {
+      // @ts-ignore
+      this._stream = this._bodySource;
+    }
     if (typeof this._bodySource === "string") {
-      throw Error("not implemented");
+      this._stream = new ReadableStream({
+        start(controller: ReadableStreamController): void {
+          controller.enqueue(this._bodySource);
+          controller.close();
+        }
+      });
     }
     return this._stream;
   }
@@ -259,6 +331,9 @@ export class Body implements domTypes.Body {
     } else if (typeof this._bodySource === "string") {
       const enc = new TextEncoder();
       return enc.encode(this._bodySource).buffer as ArrayBuffer;
+    } else if (this._bodySource instanceof ReadableStream) {
+      // @ts-ignore
+      return bufferFromStream(this._bodySource.getReader());
     } else if (this._bodySource instanceof FormData) {
       const enc = new TextEncoder();
       return enc.encode(this._bodySource.toString()).buffer as ArrayBuffer;