[Fizz] write chunks to a buffer with no re-use

packages/react-server/src/ReactServerStreamConfigBrowser.js

@@ -21,24 +21,84 @@ export function flushBuffered(destination: Destination) {
   // transform streams. https://github.com/whatwg/streams/issues/960
 }
 
-export function beginWriting(destination: Destination) {}
+const VIEW_SIZE = 512;
+let currentView = null;
+let writtenBytes = 0;
+
+export function beginWriting(destination: Destination) {
+  currentView = new Uint8Array(VIEW_SIZE);
+  writtenBytes = 0;
+}
 
 export function writeChunk(
   destination: Destination,
   chunk: PrecomputedChunk | Chunk,
 ): void {
-  destination.enqueue(chunk);
+  if (chunk.length === 0) {
+    return;
+  }
+
+  if (chunk.length > VIEW_SIZE) {
+    // this chunk may overflow a single view which implies it was not
+    // one that is cached by the streaming renderer. We will enqueu
+    // it directly and expect it is not re-used
+    if (writtenBytes > 0) {
+      destination.enqueue(
+        new Uint8Array(
+          ((currentView: any): Uint8Array).buffer,
+          0,
+          writtenBytes,
+        ),
+      );
+      currentView = new Uint8Array(VIEW_SIZE);
+      writtenBytes = 0;
+    }
+    destination.enqueue(chunk);
+    return;
+  }
+
+  let bytesToWrite = chunk;
+  const allowableBytes = ((currentView: any): Uint8Array).length - writtenBytes;
+  if (allowableBytes < bytesToWrite.length) {
+    // this chunk would overflow the current view. We enqueue a full view
+    // and start a new view with the remaining chunk
+    if (allowableBytes === 0) {
+      // the current view is already full, send it
+      destination.enqueue(currentView);
+    } else {
+      // fill up the current view and apply the remaining chunk bytes
+      // to a new view.
+      ((currentView: any): Uint8Array).set(
+        bytesToWrite.subarray(0, allowableBytes),
+        writtenBytes,
+      );
+      // writtenBytes += allowableBytes; // this can be skipped because we are going to immediately reset the view
+      destination.enqueue(currentView);
+      bytesToWrite = bytesToWrite.subarray(allowableBytes);
+    }
+    currentView = new Uint8Array(VIEW_SIZE);
+    writtenBytes = 0;
+  }
+  ((currentView: any): Uint8Array).set(bytesToWrite, writtenBytes);
+  writtenBytes += bytesToWrite.length;
 }
 
 export function writeChunkAndReturn(
   destination: Destination,
   chunk: PrecomputedChunk | Chunk,
 ): boolean {
-  destination.enqueue(chunk);
-  return destination.desiredSize > 0;
+  writeChunk(destination, chunk);
+  // in web streams there is no backpressure so we can alwas write more
+  return true;
 }
 
-export function completeWriting(destination: Destination) {}
+export function completeWriting(destination: Destination) {
+  if (currentView && writtenBytes > 0) {
+    destination.enqueue(new Uint8Array(currentView.buffer, 0, writtenBytes));
+    currentView = null;
+    writtenBytes = 0;
+  }
+}
 
 export function close(destination: Destination) {
   destination.close();