Try to replace the underlying stream of IncomingMessage after we read it

Author: Schniz

packages/next/server/base-http/node.ts

@@ -7,6 +7,11 @@ import { NEXT_REQUEST_META, RequestMeta } from '../request-meta'
 
 import { BaseNextRequest, BaseNextResponse } from './index'
 
+type Req = IncomingMessage & {
+  [NEXT_REQUEST_META]?: RequestMeta
+  cookies?: NextApiRequestCookies
+}
+
 export class NodeNextRequest extends BaseNextRequest<Readable> {
   public headers = this._req.headers;
 
@@ -21,12 +26,11 @@ export class NodeNextRequest extends BaseNextRequest<Readable> {
     return this._req
   }
 
-  constructor(
-    private _req: IncomingMessage & {
-      [NEXT_REQUEST_META]?: RequestMeta
-      cookies?: NextApiRequestCookies
-    }
-  ) {
+  set originalRequest(value: Req) {
+    this._req = value
+  }
+
+  constructor(private _req: Req) {
     super(_req.method!.toUpperCase(), _req.url!, _req)
   }
 

packages/next/server/next-server.ts

@@ -8,8 +8,8 @@ import type { ParsedNextUrl } from '../shared/lib/router/utils/parse-next-url'
 import type { PrerenderManifest } from '../build'
 import type { Rewrite } from '../lib/load-custom-routes'
 import type { BaseNextRequest, BaseNextResponse } from './base-http'
-import type { ReadableStream as ReadableStreamPolyfill } from 'next/dist/compiled/web-streams-polyfill/ponyfill'
 import { TransformStream } from 'next/dist/compiled/web-streams-polyfill/ponyfill'
+import NodeStreams from 'stream'
 
 import { execOnce } from '../shared/lib/utils'
 import {
@@ -40,7 +40,7 @@ import { PagesManifest } from '../build/webpack/plugins/pages-manifest-plugin'
 import { recursiveReadDirSync } from './lib/recursive-readdir-sync'
 import { format as formatUrl, UrlWithParsedQuery } from 'url'
 import compression from 'next/dist/compiled/compression'
-import Proxy from 'next/dist/compiled/http-proxy'
+import HttpProxy from 'next/dist/compiled/http-proxy'
 import { route } from './router'
 import { run } from './web/sandbox'
 
@@ -92,6 +92,8 @@ export interface NodeRequestHandler {
   ): Promise<void>
 }
 
+type BodyStream = ReadableStream<Uint8Array>
+
 export default class NextNodeServer extends BaseServer {
   private imageResponseCache?: ResponseCache
 
@@ -490,7 +492,7 @@ export default class NextNodeServer extends BaseServer {
     parsedUrl.search = stringifyQuery(req, query)
 
     const target = formatUrl(parsedUrl)
-    const proxy = new Proxy({
+    const proxy = new HttpProxy({
       target,
       changeOrigin: true,
       ignorePath: true,
@@ -1310,6 +1312,14 @@ export default class NextNodeServer extends BaseServer {
       }
     }
 
+    if (originalBody) {
+      const noderequest = params.request as NodeNextRequest
+      noderequest.originalRequest = enhanceIncomingMessage(
+        noderequest.originalRequest,
+        originalBody.original()
+      )
+    }
+
     return result
   }
 
@@ -1350,9 +1360,7 @@ export default class NextNodeServer extends BaseServer {
 /**
  * Creates a ReadableStream from a Node.js HTTP request
  */
-function requestToBodyStream(
-  request: IncomingMessage
-): ReadableStreamPolyfill<Uint8Array> {
+function requestToBodyStream(request: IncomingMessage): BodyStream {
   const transform = new TransformStream<Uint8Array, Uint8Array>({
     start(controller) {
       request.on('data', (chunk) => controller.enqueue(chunk))
@@ -1361,21 +1369,61 @@ function requestToBodyStream(
     },
   })
 
-  return transform.readable
+  return transform.readable as unknown as ReadableStream<Uint8Array>
 }
 
 /**
  * A simple utility to take an original stream and have
  * an API to duplicate it without closing it or mutate any variables
  */
-function teeableStream<T>(originalStream: ReadableStreamPolyfill<T>): {
-  duplicate(): ReadableStreamPolyfill<T>
+function teeableStream<T>(originalStream: ReadableStream<T>): {
+  duplicate(): ReadableStream<T>
+  original(): ReadableStream<T>
 } {
   return {
     duplicate() {
       const [stream1, stream2] = originalStream.tee()
       originalStream = stream1
       return stream2
     },
+    original() {
+      return originalStream
+    },
   }
 }
+
+function bodyStreamToNodeStream(bodyStream: BodyStream): NodeStreams.Readable {
+  const reader = bodyStream.getReader()
+  return NodeStreams.Readable.from(
+    (async function* () {
+      while (true) {
+        const { done, value } = await reader.read()
+        if (done) {
+          return
+        }
+        yield value
+      }
+    })()
+  )
+}
+
+function enhanceIncomingMessage<T extends IncomingMessage>(
+  base: T,
+  body: BodyStream
+): T {
+  const stream = bodyStreamToNodeStream(body)
+  return new Proxy<T>(base, {
+    get(target, name) {
+      if (name in stream) {
+        const v = stream[name]
+        if (typeof v === 'function') {
+          return v.bind(stream)
+        } else {
+          return v
+        }
+      }
+
+      return target[name]
+    },
+  })
+}

test/production/reading-request-body-in-middleware/index.test.ts

@@ -19,7 +19,9 @@ describe('reading request body in middleware', () => {
             const json = await request.json();
 
             if (request.nextUrl.searchParams.has("next")) {
-              return NextResponse.next();
+              const res = NextResponse.next();
+              res.headers.set('x-from-root-middleware', '1');
+              return res;
             }
 
             return new Response(JSON.stringify({
@@ -55,6 +57,15 @@ describe('reading request body in middleware', () => {
             })
           }
         `,
+
+        'pages/api/hi.js': `
+          export default function hi(req, res) {
+            res.json({
+              ...req.body,
+              api: true,
+            })
+          }
+        `,
       },
       dependencies: {},
     })
@@ -105,4 +116,29 @@ describe('reading request body in middleware', () => {
       root: false,
     })
   })
+
+  it('passes the body to the api endpoint', async () => {
+    const response = await fetchViaHTTP(
+      next.url,
+      '/api/hi',
+      {
+        next: '1',
+      },
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+        },
+        body: JSON.stringify({
+          foo: 'bar',
+        }),
+      }
+    )
+    expect(response.status).toEqual(200)
+    expect(await response.json()).toEqual({
+      foo: 'bar',
+      api: true,
+    })
+    expect(response.headers.get('x-from-root-middleware')).toEqual('1')
+  })
 })