stream ReadableStream instances

src/flatten.ts

@@ -13,6 +13,7 @@ import {
   TYPE_NULL_OBJECT,
   TYPE_PREVIOUS_RESOLVED,
   TYPE_PROMISE,
+  TYPE_STREAM,
   TYPE_REGEXP,
   TYPE_SET,
   TYPE_SYMBOL,
@@ -39,7 +40,7 @@ export function flatten(this: ThisEncode, input: unknown): number | [number] {
 }
 
 function stringify(this: ThisEncode, input: unknown, index: number) {
-  const { deferred, plugins, postPlugins } = this;
+  const { deferred, plugins, postPlugins, streams } = this;
   const str = this.stringified;
 
   const stack: [unknown, number][] = [[input, index]];
@@ -136,6 +137,22 @@ function stringify(this: ThisEncode, input: unknown, index: number) {
           } else if (input instanceof Promise) {
             str[index] = `["${TYPE_PROMISE}",${index}]`;
             deferred[index] = input;
+          } else if (input instanceof ReadableStream) {
+            str[index] = `["${TYPE_STREAM}",${index}]`;
+            // during this, the stream will be read, which is destructive
+            // so we tee it, and replace the methods on the original stream
+            // with methods forwarding to the left teed stream, while
+            // processing the right teed stream ourselves
+            const [left, right] = input.tee();
+            input.getReader = left.getReader.bind(left);
+            input.cancel = left.cancel.bind(left);
+            input.pipeThrough = left.pipeThrough.bind(left);
+            input.pipeTo = left.pipeTo.bind(left);
+            input.tee = left.tee.bind(left);
+            Object.defineProperty(input, "locked", {
+              get: () => left.locked,
+            });
+            streams[index] = right;
           } else if (input instanceof Error) {
             str[index] = `["${TYPE_ERROR}",${JSON.stringify(input.message)}`;
             if (input.name !== "Error") {

src/turbo-stream.spec.ts

@@ -572,3 +572,196 @@ test("should allow many nested promises without a memory leak", async () => {
   expect(currentDecoded.i).toBe(depth - 1);
   await decoded.done;
 });
+
+test.describe("ReadableStream", () => {
+  test("basic usage", async () => {
+    const input = new ReadableStream({
+      start(controller) {
+        controller.enqueue("foo");
+        controller.enqueue("bar");
+        setTimeout(() => {
+          controller.enqueue("foo");
+          controller.enqueue({ loba: "boba" });
+          controller.close();
+        }, 10);
+      },
+    });
+
+    const decoded = await decode(encode(input));
+
+    const inputReader = input.getReader();
+    const decodedReader = (decoded.value as typeof input).getReader();
+
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: "foo", done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: "bar", done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: "foo", done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: { loba: "boba" }, done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: undefined, done: true });
+    }
+    {
+      const inputValue = await inputReader.closed;
+      const decodedValue = await decodedReader.closed;
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual(undefined);
+    }
+    await decoded.done;
+  });
+
+  test("errors", async () => {
+    const input = new ReadableStream({
+      start(controller) {
+        controller.enqueue("foo");
+        controller.enqueue("bar");
+        setTimeout(() => {
+          controller.enqueue("foo");
+          controller.enqueue({ loba: "boba" });
+          setTimeout(() => {
+            controller.error(new Error("baz"));
+          }, 1);
+        }, 10);
+      },
+    });
+
+    const decoded = await decode(encode(input));
+
+    const inputReader = input.getReader();
+    const decodedReader = (decoded.value as typeof input).getReader();
+
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: "foo", done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: "bar", done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: "foo", done: false });
+    }
+    {
+      const inputValue = await inputReader.read();
+      const decodedValue = await decodedReader.read();
+      expect(inputValue).toEqual(decodedValue);
+      expect(inputValue).toEqual({ value: { loba: "boba" }, done: false });
+    }
+    {
+      const expected = new Error("baz");
+      await expect(inputReader.closed).rejects.toEqual(expected);
+      await expect(decodedReader.closed).rejects.toEqual(expected);
+    }
+    await decoded.done;
+  });
+
+  test("streams in a stream", async () => {
+    const nested1 = new ReadableStream({
+      start(controller) {
+        controller.enqueue("a");
+        controller.enqueue("b");
+        controller.close();
+      },
+    });
+    const nested2 = new ReadableStream({
+      start(controller) {
+        controller.enqueue("1");
+        controller.enqueue("2");
+        controller.close();
+      },
+    });
+    const input = new ReadableStream({
+      start(controller) {
+        controller.enqueue(nested1);
+        controller.enqueue(nested2);
+        controller.close();
+      },
+    });
+
+    const decoded = await decode(encode(input));
+
+    const inputReader = input.getReader();
+    const decodedReader = (decoded.value as typeof input).getReader();
+    {
+      const inputValue = (await inputReader.read()).value as ReadableStream;
+      const decodedValue = (await decodedReader.read()).value as ReadableStream;
+      expect(inputValue).toBe(nested1);
+      const inputNestedReader = inputValue.getReader();
+      const decodedNestedReader = decodedValue.getReader();
+      {
+        const inputValue = await inputNestedReader.read();
+        const decodedValue = await decodedNestedReader.read();
+        expect(inputValue).toEqual(decodedValue);
+        expect(inputValue).toEqual({ value: "a", done: false });
+      }
+      {
+        const inputValue = await inputNestedReader.read();
+        const decodedValue = await decodedNestedReader.read();
+        expect(inputValue).toEqual(decodedValue);
+        expect(inputValue).toEqual({ value: "b", done: false });
+      }
+      {
+        const inputValue = await inputNestedReader.read();
+        const decodedValue = await decodedNestedReader.read();
+        expect(inputValue).toEqual(decodedValue);
+        expect(inputValue).toEqual({ value: undefined, done: true });
+      }
+    }
+
+    {
+      const inputValue = (await inputReader.read()).value as ReadableStream;
+      const decodedValue = (await decodedReader.read()).value as ReadableStream;
+      expect(inputValue).toBe(nested2);
+      const inputNestedReader = inputValue.getReader();
+      const decodedNestedReader = decodedValue.getReader();
+      {
+        const inputValue = await inputNestedReader.read();
+        const decodedValue = await decodedNestedReader.read();
+        expect(inputValue).toEqual(decodedValue);
+        expect(inputValue).toEqual({ value: "1", done: false });
+      }
+      {
+        const inputValue = await inputNestedReader.read();
+        const decodedValue = await decodedNestedReader.read();
+        expect(inputValue).toEqual(decodedValue);
+        expect(inputValue).toEqual({ value: "2", done: false });
+      }
+      {
+        const inputValue = await inputNestedReader.read();
+        const decodedValue = await decodedNestedReader.read();
+        expect(inputValue).toEqual(decodedValue);
+        expect(inputValue).toEqual({ value: undefined, done: true });
+      }
+    }
+
+    await decoded.done;
+  });
+});