refactor(@angular/ssr): drain node stream

Author: alexfriesen

packages/angular/ssr/node/src/response.ts

@@ -8,6 +8,7 @@
 
 import type { ServerResponse } from 'node:http';
 import type { Http2ServerResponse } from 'node:http2';
+import type { ReadableStreamReadResult } from 'node:stream/web';
 
 /**
  * Streams a web-standard `Response` into a Node.js `ServerResponse`
@@ -52,27 +53,49 @@ export async function writeResponseToNodeResponse(
 
   try {
     const reader = body.getReader();
-
-    destination.on('close', () => {
-      reader.cancel().catch((error) => {
+    function cancelStream(error?: Error) {
+      reader.cancel(error).catch(() => {
         // eslint-disable-next-line no-console
         console.error(
           `An error occurred while writing the response body for: ${destination.req.url}.`,
           error,
         );
       });
-    });
-
-    // eslint-disable-next-line no-constant-condition
-    while (true) {
-      const { done, value } = await reader.read();
-      if (done) {
-        destination.end();
-        break;
+      if (error) {
+        destination.destroy(error);
       }
+    }
 
-      (destination as ServerResponse).write(value);
+    function handleStream({
+      done,
+      value,
+    }: ReadableStreamReadResult<Uint8Array>): void | Promise<void> {
+      try {
+        if (done) {
+          // End the response
+          destination.end();
+        } else if ((destination as ServerResponse).write(value)) {
+          // Continue reading recursively
+          reader.read().then(handleStream, cancelStream);
+        } else {
+          // Wait for the drain event to continue reading
+          destination.once('drain', () => reader.read().then(handleStream, cancelStream));
+        }
+      } catch (error) {
+        cancelStream(error instanceof Error ? error : undefined);
+      }
     }
+
+    destination.on('close', cancelStream);
+    destination.on('error', cancelStream);
+    reader.read().then(handleStream, cancelStream);
+
+    // Return a promise that resolves when the stream is closed
+    return reader.closed.finally(() => {
+      // cleanup listeners
+      (destination as ServerResponse).off('close', cancelStream);
+      (destination as ServerResponse).off('error', cancelStream);
+    });
   } catch {
     destination.end('Internal server error.');
   }