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.

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.

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();
-});

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!");
+  },
+);

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!");
+  },
+);

ext/net/01_net.js

@@ -4,6 +4,7 @@
 ((window) => {
   const core = window.Deno.core;
   const { BadResourcePrototype, InterruptedPrototype } = core;
+  const { ReadableStream, WritableStream } = window.__bootstrap.streams;
   const {
     ObjectPrototypeIsPrototypeOf,
     PromiseResolve,
@@ -59,10 +60,75 @@
     return core.opAsync("op_dns_resolve", { query, recordType, options });
   }
 
+  const DEFAULT_CHUNK_SIZE = 16_640;
+
+  function tryClose(rid) {
+    try {
+      core.close(rid);
+    } catch {
+      // Ignore errors
+    }
+  }
+
+  function readableStreamForRid(rid) {
+    return new ReadableStream({
+      type: "bytes",
+      async pull(controller) {
+        const v = controller.byobRequest.view;
+        try {
+          const bytesRead = await read(rid, v);
+          if (bytesRead === null) {
+            tryClose(rid);
+            controller.close();
+            controller.byobRequest.respond(0);
+          } else {
+            controller.byobRequest.respond(bytesRead);
+          }
+        } catch (e) {
+          controller.error(e);
+          tryClose(rid);
+        }
+      },
+      cancel() {
+        tryClose(rid);
+      },
+      autoAllocateChunkSize: DEFAULT_CHUNK_SIZE,
+    });
+  }
+
+  function writableStreamForRid(rid) {
+    return new WritableStream({
+      async write(chunk, controller) {
+        try {
+          let nwritten = 0;
+          while (nwritten < chunk.length) {
+            nwritten += await write(
+              rid,
+              TypedArrayPrototypeSubarray(chunk, nwritten),
+            );
+          }
+        } catch (e) {
+          controller.error(e);
+          tryClose(rid);
+        }
+      },
+      close() {
+        tryClose(rid);
+      },
+      abort() {
+        tryClose(rid);
+      },
+    });
+  }
+
   class Conn {
     #rid = 0;
     #remoteAddr = null;
     #localAddr = null;
+
+    #readable;
+    #writable;
+
     constructor(rid, remoteAddr, localAddr) {
       this.#rid = rid;
       this.#remoteAddr = remoteAddr;
@@ -104,6 +170,20 @@
     setKeepAlive(keepalive = true) {
       return core.opSync("op_set_keepalive", this.rid, keepalive);
     }
+
+    get readable() {
+      if (this.#readable === undefined) {
+        this.#readable = readableStreamForRid(this.rid);
+      }
+      return this.#readable;
+    }
+
+    get writable() {
+      if (this.#writable === undefined) {
+        this.#writable = writableStreamForRid(this.rid);
+      }
+      return this.#writable;
+    }
   }
 
   class Listener {
@@ -252,4 +332,8 @@
     Datagram,
     resolveDns,
   };
+  window.__bootstrap.streamUtils = {
+    readableStreamForRid,
+    writableStreamForRid,
+  };
 })(this);

ext/net/lib.deno_net.d.ts

@@ -54,6 +54,9 @@ declare namespace Deno {
     setNoDelay(nodelay?: boolean): void;
     /** Enable/disable keep-alive functionality */
     setKeepAlive(keepalive?: boolean): void;
+
+    readonly readable: ReadableStream<Uint8Array>;
+    readonly writable: WritableStream<Uint8Array>;
   }
 
   // deno-lint-ignore no-empty-interface

runtime/js/40_files.js

@@ -6,10 +6,12 @@
   const { read, readSync, write, writeSync } = window.__bootstrap.io;
   const { ftruncate, ftruncateSync, fstat, fstatSync } = window.__bootstrap.fs;
   const { pathFromURL } = window.__bootstrap.util;
+  const { readableStreamForRid, writableStreamForRid } =
+    window.__bootstrap.streamUtils;
   const {
+    ArrayPrototypeFilter,
     Error,
     ObjectValues,
-    ArrayPrototypeFilter,
   } = window.__bootstrap.primordials;
 
   function seekSync(
@@ -77,6 +79,9 @@
   class File {
     #rid = 0;
 
+    #readable;
+    #writable;
+
     constructor(rid) {
       this.#rid = rid;
     }
@@ -128,9 +133,25 @@
     close() {
       core.close(this.rid);
     }
+
+    get readable() {
+      if (this.#readable === undefined) {
+        this.#readable = readableStreamForRid(this.rid);
+      }
+      return this.#readable;
+    }
+
+    get writable() {
+      if (this.#writable === undefined) {
+        this.#writable = writableStreamForRid(this.rid);
+      }
+      return this.#writable;
+    }
   }
 
   class Stdin {
+    #readable;
+
     constructor() {
     }
 
@@ -149,9 +170,18 @@
     close() {
       core.close(this.rid);
     }
+
+    get readable() {
+      if (this.#readable === undefined) {
+        this.#readable = readableStreamForRid(this.rid);
+      }
+      return this.#readable;
+    }
   }
 
   class Stdout {
+    #writable;
+
     constructor() {
     }
 
@@ -170,9 +200,18 @@
     close() {
       core.close(this.rid);
     }
+
+    get writable() {
+      if (this.#writable === undefined) {
+        this.#writable = writableStreamForRid(this.rid);
+      }
+      return this.#writable;
+    }
   }
 
   class Stderr {
+    #writable;
+
     constructor() {
     }
 
@@ -191,6 +230,13 @@
     close() {
       core.close(this.rid);
     }
+
+    get writable() {
+      if (this.#writable === undefined) {
+        this.#writable = writableStreamForRid(this.rid);
+      }
+      return this.#writable;
+    }
   }
 
   const stdin = new Stdin();