Simplify WorkerReader/WorkerWriter and add readableStreamFromWorker/writableStreamFromWorker

deno.jsonc

@@ -1,7 +1,7 @@
 {
   "tasks": {
     "test": "deno test -A --parallel",
-    "bench": "deno run -A benchmark/benchmark.ts",
+    "bench": "deno bench -A",
     "check": "deno check $(find . -name '*.ts')",
     "upgrade": "deno run -A https://deno.land/x/udd/main.ts $(find . -name '*.ts')"
   }

mod.ts

@@ -1,3 +1,4 @@
+export * from "./readable_stream.ts";
 export * from "./reader.ts";
+export * from "./writable_stream.ts";
 export * from "./writer.ts";
-export * from "./types.ts";

readable_stream.ts

@@ -0,0 +1,74 @@
+const DEFAULT_CHUNK_SIZE = 16 * 1024;
+
+/**
+ * Options for creating a readable stream from a worker.
+ */
+export interface ReadableStreamFromWorkerOptions {
+  /**
+   * The size of chunks to allocate to read. The default is ~16KiB, which is
+   * the maximum size that Deno operations can currently support.
+   */
+  chunkSize?: number;
+
+  /**
+   * The queuing strategy to create the `ReadableStream` with.
+   */
+  strategy?: {
+    /**
+     * A number representing the total number of bytes that can be stored in the
+     * stream's internal queue before backpressure is applied.
+     *
+     * If not specified, it defaults to 1.
+     */
+    highWaterMark?: number | undefined;
+    /**
+     * This value should always be left undefined as the stream's underlying
+     * source is the worker.
+     */
+    size?: undefined;
+  };
+}
+
+/**
+ * Creates a readable stream that reads data from a worker's `postMessage` event.
+ *
+ * @param worker The worker to read data from.
+ * @param options The options to configure the behavior of the stream. Defaults to
+ * 16 KiB chunk size and a queuing strategy with highWaterMark of 1 and undefined size.
+ * @returns A readable stream that can be used to read the data.
+ */
+export function readableStreamFromWorker(
+  worker: Worker,
+  options: ReadableStreamFromWorkerOptions = {},
+): ReadableStream<Uint8Array> {
+  const {
+    chunkSize = DEFAULT_CHUNK_SIZE,
+    strategy,
+  } = options;
+  let onmessage: (e: MessageEvent<Uint8Array | null>) => void;
+  return new ReadableStream({
+    start(controller) {
+      onmessage = (ev) => {
+        if (ev.data === null) {
+          worker.removeEventListener("message", onmessage);
+          controller.close();
+          return;
+        } else if (ev.data instanceof Uint8Array) {
+          const data = ev.data;
+          let offset = 0;
+          while (offset < data.length) {
+            const end = offset + chunkSize;
+            controller.enqueue(data.subarray(offset, end));
+            offset = end;
+          }
+        } else {
+          throw new Error("Unexpected data posted");
+        }
+      };
+      worker.addEventListener("message", onmessage);
+    },
+    cancel() {
+      worker.removeEventListener("message", onmessage);
+    },
+  }, strategy);
+}

readable_stream_bench.ts

@@ -0,0 +1,59 @@
+import { assertEquals } from "https://deno.land/[email protected]/testing/asserts.ts";
+import { concat } from "https://deno.land/[email protected]/bytes/mod.ts";
+import * as streams from "https://deno.land/[email protected]/streams/mod.ts";
+import { readableStreamFromWorker, WorkerReader } from "./mod.ts";
+import { MockWorker } from "./test_util.ts";
+
+const count = 100;
+const sizes = [
+  64,
+  128,
+  256,
+  512,
+  1024,
+];
+
+for (const size of sizes) {
+  const data = new Uint8Array(size);
+
+  Deno.bench(
+    `readableStreamFromWorker (${
+      size.toString().padStart(2)
+    } Bytes x ${count})`,
+    {
+      group: size.toString(),
+      baseline: true,
+    },
+    async () => {
+      const worker = new MockWorker();
+      const rstream = readableStreamFromWorker(worker);
+      for (let i = 0; i < count; i++) {
+        worker.postMessage(data);
+      }
+      worker.postMessage(null);
+      const chunks = [];
+      for await (const chunk of rstream) {
+        chunks.push(chunk);
+      }
+      const content = concat(...chunks);
+      assertEquals(content.length, size * count);
+    },
+  );
+
+  Deno.bench(
+    `WorkerReader (${size.toString().padStart(2)} Bytes x ${count})`,
+    {
+      group: size.toString(),
+    },
+    async () => {
+      const worker = new MockWorker();
+      const reader = new WorkerReader(worker);
+      for (let i = 0; i < count; i++) {
+        worker.postMessage(data);
+      }
+      worker.postMessage(null);
+      const content = await streams.readAll(reader);
+      assertEquals(content.length, size * count);
+    },
+  );
+}

readable_stream_test.ts

@@ -0,0 +1,38 @@
+import { assertEquals } from "https://deno.land/[email protected]/testing/asserts.ts";
+import { concat } from "https://deno.land/[email protected]/bytes/mod.ts";
+import { readableStreamFromWorker } from "./mod.ts";
+import { MockWorker } from "./test_util.ts";
+
+Deno.test(
+  "readableStreamFromWorker returns ReadableStream that yields data from worker",
+  async () => {
+    const worker = new MockWorker();
+    const rstream = readableStreamFromWorker(worker);
+    worker.postMessage(new Uint8Array([0, 1, 2, 3, 4]));
+    worker.postMessage(new Uint8Array([5, 6, 7, 8, 9]));
+    worker.postMessage(null);
+    const chunks = [];
+    for await (const chunk of rstream) {
+      chunks.push(chunk);
+    }
+    const content = concat(...chunks);
+    assertEquals(content, new Uint8Array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]));
+  },
+);
+
+Deno.test(
+  "readableStreamFromWorker returns ReadableStream that yields data from worker prior to close",
+  async () => {
+    const worker = new MockWorker();
+    const rstream = readableStreamFromWorker(worker);
+    worker.postMessage(new Uint8Array([0, 1, 2, 3, 4]));
+    worker.postMessage(null);
+    worker.postMessage(new Uint8Array([5, 6, 7, 8, 9]));
+    const chunks = [];
+    for await (const chunk of rstream) {
+      chunks.push(chunk);
+    }
+    const content = concat(...chunks);
+    assertEquals(content, new Uint8Array([0, 1, 2, 3, 4]));
+  },
+);