Resolve streaming data for flight requests (vercel#32010)

## Bug

- [ ] Related issues linked using `fixes #number`
- [x] Integration tests added
- [ ] Errors have helpful link attached, see `contributing.md`

Author: huozhi

packages/next/server/render.tsx

@@ -1008,12 +1008,13 @@ export async function renderToHTML(
       serverComponentManifest
     )
     const reader = stream.getReader()
-    return new RenderResult((innerRes, next) => {
+    const piper: NodeWritablePiper = (innerRes, next) => {
       bufferedReadFromReadableStream(reader, (val) => innerRes.write(val)).then(
         () => next(),
         (innerErr) => next(innerErr)
       )
-    })
+    }
+    return new RenderResult(chainPipers([piper]))
   }
 
   // we preload the buildManifest for auto-export dynamic pages

test/integration/react-streaming-and-server-components/app/pages/index.server.js

@@ -19,7 +19,7 @@ export default function Index({ header, router }) {
 
 export function getServerSideProps({ req }) {
   const { headers } = req
-  const header = headers[headerKey]
+  const header = headers[headerKey] || ''
 
   return {
     props: {

test/integration/react-streaming-and-server-components/test/index.test.js

@@ -86,6 +86,20 @@ async function nextDev(dir, port) {
   })
 }
 
+async function resolveStreamResponse(response, onData) {
+  let result = ''
+  onData = onData || (() => {})
+  await new Promise((resolve) => {
+    response.body.on('data', (chunk) => {
+      result += chunk.toString()
+      onData(chunk.toString(), result)
+    })
+
+    response.body.on('end', resolve)
+  })
+  return result
+}
+
 describe('concurrentFeatures - basic', () => {
   it('should warn user for experimental risk with server components', async () => {
     const edgeRuntimeWarning =
@@ -296,24 +310,17 @@ async function runBasicTests(context, env) {
   it('should support streaming', async () => {
     await fetchViaHTTP(context.appPort, '/streaming', null, {}).then(
       async (response) => {
-        let result = ''
         let gotFallback = false
         let gotData = false
 
-        await new Promise((resolve) => {
-          response.body.on('data', (chunk) => {
-            result += chunk.toString()
-
-            gotData = result.includes('next_streaming_data')
-            if (!gotFallback) {
-              gotFallback = result.includes('next_streaming_fallback')
-              if (gotFallback) {
-                expect(gotData).toBe(false)
-              }
+        await resolveStreamResponse(response, (_, result) => {
+          gotData = result.includes('next_streaming_data')
+          if (!gotFallback) {
+            gotFallback = result.includes('next_streaming_fallback')
+            if (gotFallback) {
+              expect(gotData).toBe(false)
             }
-          })
-
-          response.body.on('end', () => resolve())
+          }
         })
 
         expect(gotFallback).toBe(true)
@@ -327,6 +334,15 @@ async function runBasicTests(context, env) {
     expect(content).toMatchInlineSnapshot('"next_streaming_data"')
   })
 
+  it('should support streaming flight request', async () => {
+    await fetchViaHTTP(context.appPort, '/?__flight__=1').then(
+      async (response) => {
+        const result = await resolveStreamResponse(response)
+        expect(result).toContain('component:index.server')
+      }
+    )
+  })
+
   it('should support api routes', async () => {
     const res = await renderViaHTTP(context.appPort, '/api/ping')
     expect(res).toContain('pong')