feat: WebSocketStream test - stream can error when blocked on data

Author: amydevs

tests/WebSocketStream.test.ts

@@ -190,6 +190,7 @@ describe(WebSocketStream.name, () => {
       const stream2Writable = stream2.writable;
       await stream2Writable.abort(testReason);
       await expect(stream1Readable.getReader().read()).rejects.toBe(testReason);
+      await expect(stream2Writable.getWriter().write()).rejects.toBe(testReason);
     }
   );
   testProp(
@@ -440,4 +441,73 @@ describe(WebSocketStream.name, () => {
       }
     },
   );
+  test('streams can be cancelled after data sent', async () => {
+    const cancelReason = Symbol('CancelReason');
+    const codeToReason = (type, code: bigint) => {
+      switch (code) {
+        case 4001n:
+          return cancelReason;
+        default:
+          return new Error(`${type.toString()} ${code.toString()}`);
+      }
+    };
+    const reasonToCode = (_type, reason) => {
+      if (reason === cancelReason) return 4001n;
+      return 0n;
+    };
+    const [_stream1, stream2] = await createStreamPair({ codeToReason, reasonToCode });
+
+    const writer = stream2.writable.getWriter();
+    await writer.write(new Uint8Array(2));
+    writer.releaseLock();
+    await stream2.cancel(cancelReason);
+
+    await expect(stream2.readable.getReader().read()).rejects.toBe(cancelReason);
+    await expect(stream2.writable.getWriter().write()).rejects.toBe(cancelReason);
+  });
+  test('streams can be cancelled with no data sent', async () => {
+    const cancelReason = Symbol('CancelReason');
+    const codeToReason = (type, code: bigint) => {
+      switch (code) {
+        case 4001n:
+          return cancelReason;
+        default:
+          return new Error(`${type.toString()} ${code.toString()}`);
+      }
+    };
+    const reasonToCode = (_type, reason) => {
+      if (reason === cancelReason) return 4001n;
+      return 0n;
+    };
+    const [_stream1, stream2] = await createStreamPair({ codeToReason, reasonToCode });
+
+    await stream2.cancel(cancelReason);
+
+    await expect(stream2.readable.getReader().read()).rejects.toBe(cancelReason);
+    await expect(stream2.writable.getWriter().write()).rejects.toBe(cancelReason);
+  });
+  test('streams can be cancelled concurrently after data sent', async () => {
+    const [stream1, stream2] = await createStreamPair();
+
+    const writer = stream2.writable.getWriter();
+    await writer.write(new Uint8Array(2));
+
+    const reader = stream1.readable.getReader();
+    reader.releaseLock();
+
+    await Promise.all([writer.close(), stream1.writable.close()])
+  });
+  test('stream can error when blocked on data', async () => {
+    const [stream1, stream2] = await createStreamPair();
+
+    const message = new Uint8Array(STREAM_BUFFER_SIZE * 2);
+
+    const stream1Writer = stream1.writable.getWriter();
+    stream1Writer.write(message);
+
+    const stream2Writer = stream2.writable.getWriter();
+    stream2Writer.write(message);
+
+    Promise.all([stream1Writer.abort(Error('some error')), stream1Writer.abort(Error('some error'))]);
+  });
 });