feat: maxReadableStreamBytes config option for readable stream in QUICStream

* Related #5

[ci skip]

Author: tegefaulkes

src/QUICClient.ts

@@ -55,6 +55,7 @@ class QUICClient extends EventTarget {
     resolveHostname = utils.resolveHostname,
     reasonToCode,
     codeToReason,
+    maxReadableStreamBytes,
     logger = new Logger(`${this.name}`),
     config = {},
   }: {
@@ -70,6 +71,7 @@ class QUICClient extends EventTarget {
     resolveHostname?: (hostname: Hostname) => Host | PromiseLike<Host>;
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
+    maxReadableStreamBytes?: number;
     logger?: Logger;
     config?: Partial<QUICConfig>;
   }) {
@@ -153,10 +155,9 @@ class QUICClient extends EventTarget {
       config: quicConfig,
       reasonToCode,
       codeToReason,
+      maxReadableStreamBytes,
       logger: logger.getChild(
-        `${QUICConnection.name} ${scid.toString().slice(32)}-${Math.floor(
-          Math.random() * 100,
-        )}`,
+        `${QUICConnection.name} ${scid.toString().slice(32)}`,
       ),
     });
     connection.addEventListener('error', handleConnectionError, { once: true });

src/QUICConnection.ts

@@ -43,6 +43,7 @@ class QUICConnection extends EventTarget {
   public streamMap: Map<StreamId, QUICStream> = new Map();
   protected reasonToCode: StreamReasonToCode;
   protected codeToReason: StreamCodeToReason;
+  protected maxReadableStreamBytes: number | undefined;
   protected destroyingMap: Map<StreamId, QUICStream> = new Map();
 
   // This basically allows one to await this promise
@@ -102,6 +103,7 @@ class QUICConnection extends EventTarget {
     reasonToCode = () => 0,
     codeToReason = (type, code) =>
       new Error(`${type.toString()} ${code.toString()}`),
+    maxReadableStreamBytes,
     logger = new Logger(`${this.name} ${scid}`),
   }: {
     scid: QUICConnectionId;
@@ -110,6 +112,7 @@ class QUICConnection extends EventTarget {
     config: QUICConfig;
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
+    maxReadableStreamBytes?: number;
     logger?: Logger;
   }) {
     logger.info(`Connect ${this.name}`);
@@ -139,6 +142,7 @@ class QUICConnection extends EventTarget {
       remoteInfo,
       reasonToCode,
       codeToReason,
+      maxReadableStreamBytes,
       logger,
     });
     socket.connectionMap.set(connection.connectionId, connection);
@@ -158,6 +162,7 @@ class QUICConnection extends EventTarget {
     reasonToCode = () => 0,
     codeToReason = (type, code) =>
       new Error(`${type.toString()} ${code.toString()}`),
+    maxReadableStreamBytes,
     logger = new Logger(`${this.name} ${scid}`),
   }: {
     scid: QUICConnectionId;
@@ -167,6 +172,7 @@ class QUICConnection extends EventTarget {
     config: QUICConfig;
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
+    maxReadableStreamBytes?: number;
     logger?: Logger;
   }): Promise<QUICConnection> {
     logger.info(`Accept ${this.name}`);
@@ -196,6 +202,7 @@ class QUICConnection extends EventTarget {
       remoteInfo,
       reasonToCode,
       codeToReason,
+      maxReadableStreamBytes,
       logger,
     });
     socket.connectionMap.set(connection.connectionId, connection);
@@ -211,6 +218,7 @@ class QUICConnection extends EventTarget {
     remoteInfo,
     reasonToCode,
     codeToReason,
+    maxReadableStreamBytes,
     logger,
   }: {
     type: 'client' | 'server';
@@ -220,6 +228,7 @@ class QUICConnection extends EventTarget {
     remoteInfo: RemoteInfo;
     reasonToCode: StreamReasonToCode;
     codeToReason: StreamCodeToReason;
+    maxReadableStreamBytes: number | undefined;
     logger: Logger;
   }) {
     super();
@@ -233,6 +242,7 @@ class QUICConnection extends EventTarget {
     this._remotePort = remoteInfo.port;
     this.reasonToCode = reasonToCode;
     this.codeToReason = codeToReason;
+    this.maxReadableStreamBytes = maxReadableStreamBytes;
     // Sets the timeout on the first
     this.checkTimeout();
 
@@ -434,11 +444,8 @@ class QUICConnection extends EventTarget {
               destroyingMap: this.destroyingMap,
               codeToReason: this.codeToReason,
               reasonToCode: this.reasonToCode,
-              logger: this.logger.getChild(
-                `${QUICStream.name} ${streamId}-${Math.floor(
-                  Math.random() * 100,
-                )}`,
-              ),
+              maxReadableStreamBytes: this.maxReadableStreamBytes,
+              logger: this.logger.getChild(`${QUICStream.name} ${streamId}`),
             });
             this.dispatchEvent(
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
@@ -456,11 +463,8 @@ class QUICConnection extends EventTarget {
               codeToReason: this.codeToReason,
               reasonToCode: this.reasonToCode,
               destroyingMap: this.destroyingMap,
-              logger: this.logger.getChild(
-                `${QUICStream.name} ${streamId}-${Math.floor(
-                  Math.random() * 100,
-                )}`,
-              ),
+              maxReadableStreamBytes: this.maxReadableStreamBytes,
+              logger: this.logger.getChild(`${QUICStream.name} ${streamId}`),
             });
             this.dispatchEvent(
               new events.QUICConnectionStreamEvent({ detail: quicStream }),
@@ -659,9 +663,8 @@ class QUICConnection extends EventTarget {
         codeToReason: this.codeToReason,
         reasonToCode: this.reasonToCode,
         destroyingMap: this.destroyingMap,
-        logger: this.logger.getChild(
-          `${QUICStream.name} ${streamId!}-${Math.floor(Math.random() * 100)}`,
-        ),
+        maxReadableStreamBytes: this.maxReadableStreamBytes,
+        logger: this.logger.getChild(`${QUICStream.name} ${streamId!}`),
       });
       // Ok the stream is opened and working
       if (this.type === 'client' && streamType === 'bidi') {

src/QUICServer.ts

@@ -38,6 +38,7 @@ class QUICServer extends EventTarget {
   protected socket: QUICSocket;
   protected reasonToCode: StreamReasonToCode | undefined;
   protected codeToReason: StreamCodeToReason | undefined;
+  protected maxReadableStreamBytes?: number | undefined;
   protected connectionMap: QUICConnectionMap;
 
   /**
@@ -62,6 +63,7 @@ class QUICServer extends EventTarget {
     resolveHostname = utils.resolveHostname,
     reasonToCode,
     codeToReason,
+    maxReadableStreamBytes,
     logger,
   }: {
     crypto: {
@@ -76,6 +78,7 @@ class QUICServer extends EventTarget {
     resolveHostname?: (hostname: Hostname) => Host | PromiseLike<Host>;
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
+    maxReadableStreamBytes?: number;
     logger?: Logger;
   }) {
     super();
@@ -103,6 +106,7 @@ class QUICServer extends EventTarget {
     this.config = quicConfig;
     this.reasonToCode = reasonToCode;
     this.codeToReason = codeToReason;
+    this.maxReadableStreamBytes = maxReadableStreamBytes;
   }
 
   @ready(new errors.ErrorQUICServerNotRunning())
@@ -276,10 +280,9 @@ class QUICServer extends EventTarget {
       config: this.config,
       reasonToCode: this.reasonToCode,
       codeToReason: this.codeToReason,
+      maxReadableStreamBytes: this.maxReadableStreamBytes,
       logger: this.logger.getChild(
-        `${QUICConnection.name} ${scid.toString().slice(32)}-${Math.floor(
-          Math.random() * 100,
-        )}`,
+        `${QUICConnection.name} ${scid.toString().slice(32)}`,
       ),
     });
 

src/QUICStream.ts

@@ -69,13 +69,15 @@ class QUICStream
     reasonToCode = () => 0,
     codeToReason = (type, code) =>
       new Error(`${type.toString()} ${code.toString()}`),
+    maxReadableStreamBytes = 1_000_000, // About 1KB
     logger = new Logger(`${this.name} ${streamId}`),
   }: {
     streamId: StreamId;
     connection: QUICConnection;
     destroyingMap: Map<StreamId, QUICStream>;
     reasonToCode?: StreamReasonToCode;
     codeToReason?: StreamCodeToReason;
+    maxReadableStreamBytes?: number;
     logger?: Logger;
   }): Promise<QUICStream> {
     logger.info(`Create ${this.name}`);
@@ -92,6 +94,7 @@ class QUICStream
       reasonToCode,
       codeToReason,
       destroyingMap,
+      maxReadableStreamBytes,
       logger,
     });
     connection.streamMap.set(stream.streamId, stream);
@@ -105,13 +108,15 @@ class QUICStream
     reasonToCode,
     codeToReason,
     destroyingMap,
+    maxReadableStreamBytes,
     logger,
   }: {
     streamId: StreamId;
     connection: QUICConnection;
     reasonToCode: StreamReasonToCode;
     codeToReason: StreamCodeToReason;
     destroyingMap: Map<StreamId, QUICStream>;
+    maxReadableStreamBytes: number;
     logger: Logger;
   }) {
     super();
@@ -124,8 +129,6 @@ class QUICStream
     this.codeToReason = codeToReason;
     this.destroyingMap = destroyingMap;
 
-    // Try the BYOB later, it seems more performant
-
     this.readable = new ReadableStream(
       {
         type: 'bytes',
@@ -142,7 +145,8 @@ class QUICStream
         },
       },
       {
-        highWaterMark: 20,
+        highWaterMark: maxReadableStreamBytes,
+        size: (chunk) => chunk?.byteLength ?? 0,
       },
     );