Use ReadableStream in RenderResult (vercel#34005)

Since we're always using `ReadableStream`, we should just get rid of `ResultPiper`.

This also lets us replace things like `bufferedReadFromReadableStream` with a `TransformStream` that does the same thing, so that it's `TransformStream`s all the way down.

Finally, we can get rid of the one-off call to `renderToReadableStream` and just use `renderToStream` whenever we're rendering a concurrent tree.

packages/next/server/node-polyfill-web-streams.js

@@ -0,0 +1,7 @@
+import { ReadableStream, TransformStream } from './web/sandbox/readable-stream'
+
+// Polyfill Web Streams in the Node.js environment
+if (!global.ReadableStream) {
+  global.ReadableStream = ReadableStream
+  global.TransformStream = TransformStream
+}

packages/next/server/render-result.ts

@@ -1,14 +1,9 @@
 import type { ServerResponse } from 'http'
 
-export type ResultPiper = (
-  push: (chunks: Uint8Array[]) => void,
-  next: (err?: Error) => void
-) => void
-
 export default class RenderResult {
-  _result: string | ResultPiper
+  _result: string | ReadableStream
 
-  constructor(response: string | ResultPiper) {
+  constructor(response: string | ReadableStream) {
     this._result = response
   }
 
@@ -33,31 +28,30 @@ export default class RenderResult {
         ? () => (res as any).flush()
         : () => {}
 
-    return new Promise((resolve, reject) => {
+    return (async () => {
+      const reader = response.getReader()
       let fatalError = false
-      response(
-        (chunks) => {
-          // The state of the stream is non-deterministic after
-          // writing, so any error becomes fatal.
-          fatalError = true
-          res.cork()
-          chunks.forEach((chunk) => res.write(chunk))
-          res.uncork()
-          flush()
-        },
-        (err) => {
-          if (err) {
-            if (fatalError) {
-              res.destroy(err)
-            }
-            reject(err)
-          } else {
+
+      try {
+        while (true) {
+          const { done, value } = await reader.read()
+
+          if (done) {
             res.end()
-            resolve()
+            return
           }
+
+          fatalError = true
+          res.write(value)
+          flush()
         }
-      )
-    })
+      } catch (err) {
+        if (fatalError) {
+          res.destroy(err as any)
+        }
+        throw err
+      }
+    })()
   }
 
   isDynamic(): boolean {