feat(runtime): web streams in fs & net APIs (#13615)

This commit adds `readable` and `writable` properties to `Deno.File` and
`Deno.Conn`. This makes it very simple to use files and network sockets
with fetch or the native HTTP server.

4 files changed, 203 insertions, 83 deletions

diff --git a/cli/dts/lib.deno.ns.d.ts b/cli/dts/lib.deno.ns.d.ts
index f24e2feca..ff8fbb741 100644
--- a/cli/dts/lib.deno.ns.d.ts
+++ b/cli/dts/lib.deno.ns.d.ts
@@ -1092,14 +1092,26 @@ declare namespace Deno {
     stat(): Promise<FileInfo>;
     statSync(): FileInfo;
     close(): void;
+
+    readonly readable: ReadableStream<Uint8Array>;
+    readonly writable: WritableStream<Uint8Array>;
   }
 
   /** A handle for `stdin`. */
-  export const stdin: Reader & ReaderSync & Closer & { readonly rid: number };
+  export const stdin: Reader & ReaderSync & Closer & {
+    readonly rid: number;
+    readonly readable: ReadableStream<Uint8Array>;
+  };
   /** A handle for `stdout`. */
-  export const stdout: Writer & WriterSync & Closer & { readonly rid: number };
+  export const stdout: Writer & WriterSync & Closer & {
+    readonly rid: number;
+    readonly writable: WritableStream<Uint8Array>;
+  };
   /** A handle for `stderr`. */
-  export const stderr: Writer & WriterSync & Closer & { readonly rid: number };
+  export const stderr: Writer & WriterSync & Closer & {
+    readonly rid: number;
+    readonly writable: WritableStream<Uint8Array>;
+  };
 
   export interface OpenOptions {
     /** Sets the option for read access. This option, when `true`, means that the
@@ -2208,12 +2220,18 @@ declare namespace Deno {
   export class Process<T extends RunOptions = RunOptions> {
     readonly rid: number;
     readonly pid: number;
-    readonly stdin: T["stdin"] extends "piped" ? Writer & Closer
-      : (Writer & Closer) | null;
-    readonly stdout: T["stdout"] extends "piped" ? Reader & Closer
-      : (Reader & Closer) | null;
-    readonly stderr: T["stderr"] extends "piped" ? Reader & Closer
-      : (Reader & Closer) | null;
+    readonly stdin: T["stdin"] extends "piped" ? Writer & Closer & {
+      writable: WritableStream<Uint8Array>;
+    }
+      : (Writer & Closer & { writable: WritableStream<Uint8Array> }) | null;
+    readonly stdout: T["stdout"] extends "piped" ? Reader & Closer & {
+      readable: ReadableStream<Uint8Array>;
+    }
+      : (Reader & Closer & { readable: ReadableStream<Uint8Array> }) | null;
+    readonly stderr: T["stderr"] extends "piped" ? Reader & Closer & {
+      readable: ReadableStream<Uint8Array>;
+    }
+      : (Reader & Closer & { readable: ReadableStream<Uint8Array> }) | null;
     /** Wait for the process to exit and return its exit status.
      *
      * Calling this function multiple times will return the same status.
diff --git a/cli/tests/unit/file_test.ts b/cli/tests/unit/file_test.ts
index 8159e898c..a89496b28 100644
--- a/cli/tests/unit/file_test.ts
+++ b/cli/tests/unit/file_test.ts
@@ -99,77 +99,3 @@ Deno.test(function fileUsingNumberFileName() {
 Deno.test(function fileUsingEmptyStringFileName() {
   testSecondArgument("", "");
 });
-
-Deno.test(
-  { permissions: { read: true, write: true } },
-  function fileTruncateSyncSuccess() {
-    const filename = Deno.makeTempDirSync() + "/test_fileTruncateSync.txt";
-    const file = Deno.openSync(filename, {
-      create: true,
-      read: true,
-      write: true,
-    });
-
-    file.truncateSync(20);
-    assertEquals(Deno.readFileSync(filename).byteLength, 20);
-    file.truncateSync(5);
-    assertEquals(Deno.readFileSync(filename).byteLength, 5);
-    file.truncateSync(-5);
-    assertEquals(Deno.readFileSync(filename).byteLength, 0);
-
-    file.close();
-    Deno.removeSync(filename);
-  },
-);
-
-Deno.test(
-  { permissions: { read: true, write: true } },
-  async function fileTruncateSuccess() {
-    const filename = Deno.makeTempDirSync() + "/test_fileTruncate.txt";
-    const file = await Deno.open(filename, {
-      create: true,
-      read: true,
-      write: true,
-    });
-
-    await file.truncate(20);
-    assertEquals((await Deno.readFile(filename)).byteLength, 20);
-    await file.truncate(5);
-    assertEquals((await Deno.readFile(filename)).byteLength, 5);
-    await file.truncate(-5);
-    assertEquals((await Deno.readFile(filename)).byteLength, 0);
-
-    file.close();
-    await Deno.remove(filename);
-  },
-);
-
-Deno.test({ permissions: { read: true } }, function fileStatSyncSuccess() {
-  const file = Deno.openSync("README.md");
-  const fileInfo = file.statSync();
-  assert(fileInfo.isFile);
-  assert(!fileInfo.isSymlink);
-  assert(!fileInfo.isDirectory);
-  assert(fileInfo.size);
-  assert(fileInfo.atime);
-  assert(fileInfo.mtime);
-  // The `birthtime` field is not available on Linux before kernel version 4.11.
-  assert(fileInfo.birthtime || Deno.build.os === "linux");
-
-  file.close();
-});
-
-Deno.test({ permissions: { read: true } }, async function fileStatSuccess() {
-  const file = await Deno.open("README.md");
-  const fileInfo = await file.stat();
-  assert(fileInfo.isFile);
-  assert(!fileInfo.isSymlink);
-  assert(!fileInfo.isDirectory);
-  assert(fileInfo.size);
-  assert(fileInfo.atime);
-  assert(fileInfo.mtime);
-  // The `birthtime` field is not available on Linux before kernel version 4.11.
-  assert(fileInfo.birthtime || Deno.build.os === "linux");
-
-  file.close();
-});
diff --git a/cli/tests/unit/files_test.ts b/cli/tests/unit/files_test.ts
index 3e30fed9a..a509672c0 100644
--- a/cli/tests/unit/files_test.ts
+++ b/cli/tests/unit/files_test.ts
@@ -671,3 +671,123 @@ Deno.test({ permissions: { read: true } }, async function seekMode() {
   assertEquals(new TextDecoder().decode(buf), "H");
   file.close();
 });
+
+Deno.test(
+  { permissions: { read: true, write: true } },
+  function fileTruncateSyncSuccess() {
+    const filename = Deno.makeTempDirSync() + "/test_fileTruncateSync.txt";
+    const file = Deno.openSync(filename, {
+      create: true,
+      read: true,
+      write: true,
+    });
+
+    file.truncateSync(20);
+    assertEquals(Deno.readFileSync(filename).byteLength, 20);
+    file.truncateSync(5);
+    assertEquals(Deno.readFileSync(filename).byteLength, 5);
+    file.truncateSync(-5);
+    assertEquals(Deno.readFileSync(filename).byteLength, 0);
+
+    file.close();
+    Deno.removeSync(filename);
+  },
+);
+
+Deno.test(
+  { permissions: { read: true, write: true } },
+  async function fileTruncateSuccess() {
+    const filename = Deno.makeTempDirSync() + "/test_fileTruncate.txt";
+    const file = await Deno.open(filename, {
+      create: true,
+      read: true,
+      write: true,
+    });
+
+    await file.truncate(20);
+    assertEquals((await Deno.readFile(filename)).byteLength, 20);
+    await file.truncate(5);
+    assertEquals((await Deno.readFile(filename)).byteLength, 5);
+    await file.truncate(-5);
+    assertEquals((await Deno.readFile(filename)).byteLength, 0);
+
+    file.close();
+    await Deno.remove(filename);
+  },
+);
+
+Deno.test({ permissions: { read: true } }, function fileStatSyncSuccess() {
+  const file = Deno.openSync("README.md");
+  const fileInfo = file.statSync();
+  assert(fileInfo.isFile);
+  assert(!fileInfo.isSymlink);
+  assert(!fileInfo.isDirectory);
+  assert(fileInfo.size);
+  assert(fileInfo.atime);
+  assert(fileInfo.mtime);
+  // The `birthtime` field is not available on Linux before kernel version 4.11.
+  assert(fileInfo.birthtime || Deno.build.os === "linux");
+
+  file.close();
+});
+
+Deno.test(async function fileStatSuccess() {
+  const file = await Deno.open("README.md");
+  const fileInfo = await file.stat();
+  assert(fileInfo.isFile);
+  assert(!fileInfo.isSymlink);
+  assert(!fileInfo.isDirectory);
+  assert(fileInfo.size);
+  assert(fileInfo.atime);
+  assert(fileInfo.mtime);
+  // The `birthtime` field is not available on Linux before kernel version 4.11.
+  assert(fileInfo.birthtime || Deno.build.os === "linux");
+
+  file.close();
+});
+
+Deno.test({ permissions: { read: true } }, async function readableStream() {
+  const filename = "cli/tests/testdata/hello.txt";
+  const file = await Deno.open(filename);
+  assert(file.readable instanceof ReadableStream);
+  const chunks = [];
+  for await (const chunk of file.readable) {
+    chunks.push(chunk);
+  }
+  assertEquals(chunks.length, 1);
+  assertEquals(chunks[0].byteLength, 12);
+});
+
+Deno.test(
+  { permissions: { read: true } },
+  async function readableStreamTextEncoderPipe() {
+    const filename = "cli/tests/testdata/hello.txt";
+    const file = await Deno.open(filename);
+    const readable = file.readable.pipeThrough(new TextDecoderStream());
+    const chunks = [];
+    for await (const chunk of readable) {
+      chunks.push(chunk);
+    }
+    assertEquals(chunks.length, 1);
+    assertEquals(chunks[0].length, 12);
+  },
+);
+
+Deno.test(
+  { permissions: { read: true, write: true } },
+  async function writableStream() {
+    const path = await Deno.makeTempFile();
+    const file = await Deno.open(path, { write: true });
+    assert(file.writable instanceof WritableStream);
+    const readable = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("hello "));
+        controller.enqueue(new TextEncoder().encode("world!"));
+        controller.close();
+      },
+    });
+    await readable.pipeTo(file.writable);
+    const res = await Deno.readTextFile(path);
+    assertEquals(res, "hello world!");
+  },
+);
diff --git a/cli/tests/unit/net_test.ts b/cli/tests/unit/net_test.ts
index 052202676..3953d5866 100644
--- a/cli/tests/unit/net_test.ts
+++ b/cli/tests/unit/net_test.ts
@@ -751,3 +751,59 @@ Deno.test(
     listener.close();
   },
 );
+
+Deno.test({ permissions: { net: true } }, async function whatwgStreams() {
+  (async () => {
+    const listener = Deno.listen({ hostname: "127.0.0.1", port: 3500 });
+    const conn = await listener.accept();
+    await conn.readable.pipeTo(conn.writable);
+    listener.close();
+  })();
+
+  const conn = await Deno.connect({ hostname: "127.0.0.1", port: 3500 });
+  const reader = conn.readable.getReader();
+  const writer = conn.writable.getWriter();
+  const encoder = new TextEncoder();
+  const decoder = new TextDecoder();
+  const data = encoder.encode("Hello World");
+
+  await writer.write(data);
+  const { value, done } = await reader.read();
+  assert(!done);
+  assertEquals(decoder.decode(value), "Hello World");
+  await reader.cancel();
+});
+
+Deno.test(
+  { permissions: { read: true } },
+  async function readableStreamTextEncoderPipe() {
+    const filename = "cli/tests/testdata/hello.txt";
+    const file = await Deno.open(filename);
+    const readable = file.readable.pipeThrough(new TextDecoderStream());
+    const chunks = [];
+    for await (const chunk of readable) {
+      chunks.push(chunk);
+    }
+    assertEquals(chunks.length, 1);
+    assertEquals(chunks[0].length, 12);
+  },
+);
+
+Deno.test(
+  { permissions: { read: true, write: true } },
+  async function writableStream() {
+    const path = await Deno.makeTempFile();
+    const file = await Deno.open(path, { write: true });
+    assert(file.writable instanceof WritableStream);
+    const readable = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("hello "));
+        controller.enqueue(new TextEncoder().encode("world!"));
+        controller.close();
+      },
+    });
+    await readable.pipeTo(file.writable);
+    const res = await Deno.readTextFile(path);
+    assertEquals(res, "hello world!");
+  },
+);