Update the WritableIterable behavior

packages/connect-web-bench/README.md

@@ -10,5 +10,5 @@ it like a web server would usually do.
 
 | code generator | bundle size        | minified               | compressed           |
 |----------------|-------------------:|-----------------------:|---------------------:|
-| connect        | 113,771 b | 49,913 b | 13,375 b |
+| connect        | 113,771 b | 49,913 b | 13,377 b |
 | grpc-web       | 414,906 b    | 301,127 b    | 53,279 b |

packages/connect/src/protocol/async-iterable-story.spec.ts

@@ -160,7 +160,7 @@ describe("full story", function () {
         .then(() => writer.write({ value: "gamma", end: false }))
         .then(() => writer.write({ value: "delta", end: false }))
         .finally(() => {
-          void writer.close();
+          writer.close();
         });
 
       const resp = await readAll(readerIt);
@@ -194,22 +194,19 @@ describe("full story", function () {
         { value: "delta", end: false },
       ]);
     });
-    it("should throw if close is called on a writer that is already closed", function () {
-      void writer.close();
-
-      writer
-        .close()
-        .catch((e) =>
-          expect(e).toEqual(new ConnectError("cannot close, already closed"))
-        );
+    it("should allow close on a writer that is already closed", function () {
+      writer.close();
+      writer.close();
     });
     it("should throw if write is called on a writer that is closed", function () {
-      void writer.close();
+      writer.close();
 
       writer
         .write({ value: "alpha", end: false })
         .catch((e) =>
-          expect(e).toEqual(new ConnectError("cannot write, already closed"))
+          expect(e).toEqual(
+            new Error("cannot write, WritableIterable already closed")
+          )
         );
     });
     it("should correctly behave when consumer fails and throw is invoked", async function () {
@@ -230,11 +227,11 @@ describe("full story", function () {
       )[] = [];
       try {
         // Iterate over the reader and purposely fail after the first read.
-        for (;;) {
-          const result = await readerIt[Symbol.asyncIterator]().next();
-          resp.push(result.value as { end: false; value: string });
-          throw "READER_ERROR";
-        }
+        const itr = readerIt[Symbol.asyncIterator]();
+        const result = await itr.next();
+        resp.push(result.value as { end: false; value: string });
+        await itr.next();
+        throw "READER_ERROR";
       } catch (e) {
         // Verify we got the first send only and then verify we caught the expected error.
         expect(resp).toEqual([{ value: "alpha", end: false }]);
@@ -265,11 +262,16 @@ describe("full story", function () {
         .then(() => fail("send was unexpectedly resolved."))
         .catch((e) => expect(e).toBe("READER_ERROR"));
 
-      // The reader's internal writer is closed so any future reads will be rejected.
+      // The reader's internal writer is closed so any future reads should result in the
+      // done.
       readerIt[Symbol.asyncIterator]()
         .next()
-        .then(() => fail("reads were unexpectedly resolved"))
-        .catch((e) => expect(e).toBe("READER_ERROR"));
+        .then((result) =>
+          expect(result).toEqual({ done: true, value: undefined })
+        )
+        .catch(() =>
+          fail("expected successful done result but unexpectedly rejected")
+        );
     });
   });
 

packages/connect/src/protocol/async-iterable.spec.ts

@@ -14,6 +14,7 @@
 
 import {
   createAsyncIterable,
+  createWritableIterable,
   makeIterableAbortable,
   pipe,
   pipeTo,
@@ -1071,3 +1072,113 @@ describe("transforming asynchronous iterables", () => {
     });
   });
 });
+
+describe("createWritableIterable()", function () {
+  it("works like a regular iterable on the happy path", async () => {
+    const wIterable = createWritableIterable<number>();
+    let readCount = 0;
+    const read = (async () => {
+      for await (const next of wIterable) {
+        expect(next).toBe(readCount);
+        readCount++;
+      }
+    })();
+    const writCount = 5;
+    for (let i = 0; i < writCount; i++) {
+      await wIterable.write(i);
+    }
+    wIterable.close();
+    await read;
+    expect(readCount).toEqual(writCount);
+  });
+  it("write is interrupted when read fails", async () => {
+    const wIterable = createWritableIterable<number>();
+    const read = (async () => {
+      const itr = wIterable[Symbol.asyncIterator]();
+      const next = await itr.next();
+      if (next.done === true) {
+        fail("expected at least one value");
+      } else {
+        expect(await itr.throw?.(new Error("read failed"))).toEqual({
+          done: true,
+          value: undefined,
+        });
+      }
+      // All further calls to next should also result in done results.
+      expect(await itr.next()).toEqual({
+        done: true,
+        value: undefined,
+      });
+    })();
+    await expectAsync(wIterable.write(1)).toBeRejected();
+    await expectAsync(wIterable.write(2)).toBeRejected();
+    await read;
+  });
+  it("queues writes", async () => {
+    const wIterable = createWritableIterable<number>();
+    const writCount = 50;
+    const writes = [];
+    for (let i = 0; i < writCount; i++) {
+      writes.push(wIterable.write(i));
+    }
+    let readCount = 0;
+    const read = (async () => {
+      for await (const next of wIterable) {
+        expect(next).toBe(readCount);
+        readCount++;
+      }
+    })();
+    wIterable.close();
+    await read;
+    expect(readCount).toEqual(writCount);
+    await Promise.all(writes);
+  });
+  it("queues reads", async () => {
+    const wIterable = createWritableIterable<number>();
+    const writCount = 50;
+    const read = (async () => {
+      const itr = wIterable[Symbol.asyncIterator]();
+      const readPromises: Promise<IteratorResult<number, number>>[] = [];
+      for (let i = 0; i < writCount; i++) {
+        readPromises.push(itr.next());
+      }
+      const reads = await Promise.all(readPromises);
+      expect(reads.find((r) => r.done === true)).toBeUndefined();
+      expect(reads.map((r) => r.value)).toEqual([...Array(writCount).keys()]);
+      await expectAsync(itr.next()).toBeResolvedTo({
+        done: true,
+        value: undefined,
+      });
+    })();
+    for (let i = 0; i < writCount; i++) {
+      await wIterable.write(i);
+    }
+    wIterable.close();
+    await read;
+  });
+  it("queues reads and writes", async () => {
+    const wIterable = createWritableIterable<number>();
+    const writCount = 50;
+    const read = (async () => {
+      const itr = wIterable[Symbol.asyncIterator]();
+      const readPromises: Promise<IteratorResult<number, number>>[] = [];
+      for (let i = 0; i < writCount; i++) {
+        readPromises.push(itr.next());
+      }
+      const reads = await Promise.all(readPromises);
+      expect(reads.find((r) => r.done === true)).toBeUndefined();
+      expect(reads.map((r) => r.value)).toEqual([...Array(writCount).keys()]);
+      await expectAsync(itr.next()).toBeResolvedTo({
+        done: true,
+        value: undefined,
+      });
+    })();
+    const writes: Promise<void>[] = [];
+    for (let i = 0; i < writCount; i++) {
+      writes.push(wIterable.write(i));
+    }
+    wIterable.close();
+    await Promise.all(writes);
+    await read;
+  });
+});