Refactor server/render for SSR streaming (#31231)

Initial step to refactor the rendering logic by decoupling the handler and renderer:
1. Delegate Flight rendering to server/render
2. Reuse the piper glue code for both Fizz and Flight streams
3. Add buffering for ReadableStream

In 1), this PR also makes sure that gSSP/gSP are correctly executed before the Flight stream and `pageProps` and `router` are correctly delivered to the component.

Related to #30994.

## Bug

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

## Feature

- [ ] Implements an existing feature request or RFC. Make sure the feature request has been accepted for implementation before opening a PR.
- [ ] Related issues linked using `fixes #number`
- [ ] Integration tests added
- [ ] Documentation added
- [ ] Telemetry added. In case of a feature if it's used or not.
- [ ] Errors have helpful link attached, see `contributing.md`

## Documentation / Examples

- [ ] Make sure the linting passes by running `yarn lint`

packages/next/build/webpack/loaders/next-middleware-ssr-loader/index.ts

@@ -28,16 +28,6 @@ export default async function middlewareRSCLoader(this: any) {
         import { RouterContext } from 'next/dist/shared/lib/router-context'
         import { renderToHTML } from 'next/dist/server/web/render'
 
-        import React, { createElement } from 'react'
-
-        ${
-          isServerComponent
-            ? `
-        import { renderToReadableStream } from 'next/dist/compiled/react-server-dom-webpack/writer.browser.server'
-        import { createFromReadableStream } from 'next/dist/compiled/react-server-dom-webpack'`
-            : ''
-        }
-
         ${appDefinition}
         ${documentDefinition}
         
@@ -57,47 +47,7 @@ export default async function middlewareRSCLoader(this: any) {
           throw new Error('Your page must export a \`default\` component')
         }
 
-        function wrapReadable(readable) {
-          const encoder = new TextEncoder()
-          const transformStream = new TransformStream()
-          const writer = transformStream.writable.getWriter()
-          const reader = readable.getReader()
-          const process = () => {
-            reader.read().then(({ done, value }) => {
-              if (!done) {
-                writer.write(typeof value === 'string' ? encoder.encode(value) : value)
-                process()
-              } else {
-                writer.close()
-              }
-            })
-          }
-          process()
-          return transformStream.readable
-        }
-        
-        ${
-          isServerComponent
-            ? `
-        const renderFlight = props => renderToReadableStream(createElement(Page, props), rscManifest)
-
-        let responseCache
-        const FlightWrapper = props => {
-          let response = responseCache
-          if (!response) {
-            responseCache = response = createFromReadableStream(renderFlight(props))
-          }
-          return response.readRoot()
-        }
-        const Component = props => {
-          return createElement(
-            React.Suspense,
-            { fallback: null },
-            createElement(FlightWrapper, props)
-          )
-        }`
-            : `const Component = Page`
-        }
+        const Component = Page
 
         async function render(request) {
           const url = request.nextUrl
@@ -111,28 +61,10 @@ export default async function middlewareRSCLoader(this: any) {
             })
           }
 
-          ${
-            isServerComponent
-              ? `
-          // Flight data request
-          const isFlightDataRequest = query.__flight__ !== undefined
-          if (isFlightDataRequest) {
-            delete query.__flight__
-            return new Response(
-              wrapReadable(
-                renderFlight({
-                  router: {
-                    route: pathname,
-                    asPath: pathname,
-                    pathname: pathname,
-                    query,
-                  }
-                })
-              )
-            )
-          }`
-              : ''
+          const renderServerComponentData = ${
+            isServerComponent ? `query.__flight__ !== undefined` : 'false'
           }
+          delete query.__flight__
 
           const renderOpts = {
             Component,
@@ -156,7 +88,10 @@ export default async function middlewareRSCLoader(this: any) {
             basePath: ${JSON.stringify(basePath || '')},
             supportsDynamicHTML: true,
             concurrentFeatures: true,
-            renderServerComponent: ${isServerComponent ? 'true' : 'false'},
+            renderServerComponentData,
+            serverComponentManifest: ${
+              isServerComponent ? 'rscManifest' : 'null'
+            },
           }
 
           const transformStream = new TransformStream()
@@ -173,7 +108,8 @@ export default async function middlewareRSCLoader(this: any) {
             )
             result.pipe({
               write: str => writer.write(encoder.encode(str)),
-              end: () => writer.close()
+              end: () => writer.close(),
+              // Not implemented: cork/uncork/on/removeListener
             })
           } catch (err) {
             return new Response(