feat(multipart-parser): add ReadableStream support for multipart parts

packages/multipart-parser/src/index.ts

@@ -6,7 +6,8 @@ export {
   parseMultipart,
   parseMultipartStream,
   MultipartParser,
-  MultipartPart,
+  BufferedMultipartPart,
+  StreamedMultipartPart
 } from './lib/multipart.ts'
 
 export {

packages/multipart-parser/src/lib/multipart-request.test.ts

@@ -3,7 +3,6 @@ import { describe, it } from 'node:test'
 
 import { createMultipartMessage, getRandomBytes } from '../../test/utils.ts'
 
-import type { MultipartPart } from './multipart.ts'
 import {
   MultipartParseError,
   MaxHeaderSizeExceededError,
@@ -82,7 +81,6 @@ describe('parseMultipartRequest', async () => {
     for await (let part of parseMultipartRequest(request)) {
       parts.push(part)
     }
-
     assert.equal(parts.length, 0)
   })
 
@@ -97,10 +95,11 @@ describe('parseMultipartRequest', async () => {
       }),
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
+    let parts = await Promise.all(buffering_parts)
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].name, 'field1')
@@ -119,11 +118,11 @@ describe('parseMultipartRequest', async () => {
       }),
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
-
+    let parts = await Promise.all(buffering_parts)
     assert.equal(parts.length, 2)
     assert.equal(parts[0].name, 'field1')
     assert.equal(parts[0].text, 'value1')
@@ -142,10 +141,11 @@ describe('parseMultipartRequest', async () => {
       }),
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
+    let parts = await Promise.all(buffering_parts)
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].name, 'empty')
@@ -167,10 +167,11 @@ describe('parseMultipartRequest', async () => {
       }),
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
+    let parts = await Promise.all(buffering_parts)
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].name, 'file1')
@@ -196,10 +197,11 @@ describe('parseMultipartRequest', async () => {
       }),
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
+    let parts = await Promise.all(buffering_parts)
 
     assert.equal(parts.length, 3)
     assert.equal(parts[0].name, 'field1')
@@ -229,21 +231,21 @@ describe('parseMultipartRequest', async () => {
       }),
     })
 
-    let parts: { name?: string; filename?: string; mediaType?: string; content: Uint8Array }[] = []
+    let parts: { name?: string; filename?: string; mediaType?: string; content: Promise<Uint8Array> }[] = []
     for await (let part of parseMultipartRequest(request, { maxFileSize })) {
       parts.push({
         name: part.name,
         filename: part.filename,
         mediaType: part.mediaType,
-        content: part.bytes,
+        content: part.toBuffered().then(b => b.bytes),
       })
     }
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].name, 'file1')
     assert.equal(parts[0].filename, 'random.dat')
     assert.equal(parts[0].mediaType, 'application/octet-stream')
-    assert.deepEqual(parts[0].content, content)
+    assert.deepEqual(await parts[0].content, content)
   })
 
   it('throws when Content-Type is not multipart/form-data', async () => {
@@ -331,10 +333,11 @@ describe('parseMultipartRequest', async () => {
       body: [`--${boundary}`, 'Invalid-Header', '', 'Some content', `--${boundary}--`].join(CRLF),
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
+    let parts = await Promise.all(buffering_parts)
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].headers.get('Invalid-Header'), null)

packages/multipart-parser/src/lib/multipart-request.ts

@@ -1,4 +1,4 @@
-import type { MultipartParserOptions, MultipartPart } from './multipart.ts'
+import type { MultipartParserOptions, StreamedMultipartPart } from './multipart.ts'
 import { MultipartParseError, parseMultipartStream } from './multipart.ts'
 
 /**
@@ -34,7 +34,7 @@ export function isMultipartRequest(request: Request): boolean {
 export async function* parseMultipartRequest(
   request: Request,
   options?: MultipartParserOptions,
-): AsyncGenerator<MultipartPart, void, unknown> {
+): AsyncGenerator<StreamedMultipartPart, void, unknown> {
   if (!isMultipartRequest(request)) {
     throw new MultipartParseError('Request is not a multipart request')
   }

packages/multipart-parser/src/lib/multipart.node.test.ts

@@ -1,11 +1,10 @@
 import * as assert from 'node:assert/strict'
 import { describe, it } from 'node:test'
 
-import { getRandomBytes } from '../../test/utils.ts'
+import { createMultipartMessage, getRandomBytes } from '../../test/utils.ts'
 import { createMultipartRequest } from '../../test/utils.node.ts'
 
-import type { MultipartPart } from './multipart.ts'
-import { parseMultipartRequest } from './multipart.node.ts'
+import { parseMultipartRequest, parseMultipart } from './multipart.node.ts'
 
 describe('parseMultipartRequest (node)', () => {
   let boundary = '----WebKitFormBoundaryzv5f5B2cY6tjQ0Rn'
@@ -26,10 +25,11 @@ describe('parseMultipartRequest (node)', () => {
       field1: 'value1',
     })
 
-    let parts: MultipartPart[] = []
+    let buffering_parts = []
     for await (let part of parseMultipartRequest(request)) {
-      parts.push(part)
+      buffering_parts.push(part.toBuffered())
     }
+    let parts = await Promise.all(buffering_parts)
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].name, 'field1')
@@ -47,20 +47,38 @@ describe('parseMultipartRequest (node)', () => {
       },
     })
 
-    let parts: { name?: string; filename?: string; mediaType?: string; content: Uint8Array }[] = []
+    let parts: { name?: string; filename?: string; mediaType?: string; content: Promise<Uint8Array> }[] = []
     for await (let part of parseMultipartRequest(request, { maxFileSize })) {
       parts.push({
         name: part.name,
         filename: part.filename,
         mediaType: part.mediaType,
-        content: part.bytes,
+        content: part.toBuffered().then((b) => b.bytes),
       })
     }
 
     assert.equal(parts.length, 1)
     assert.equal(parts[0].name, 'file1')
     assert.equal(parts[0].filename, 'tesla.jpg')
     assert.equal(parts[0].mediaType, 'image/jpeg')
-    assert.deepEqual(parts[0].content, content)
+    assert.deepEqual(await parts[0].content, content)
+  })
+
+  it('parses multiple parts correctly', async () => {
+    let message = Buffer.from(createMultipartMessage(boundary, {
+        field1: 'value1',
+        field2: 'value2',
+      }))
+
+    let parts = []
+    for await (let part of parseMultipart(message, {boundary})) {
+      parts.push(part)
+    }
+
+    assert.equal(parts.length, 2)
+    assert.equal(parts[0].name, 'field1')
+    assert.equal(parts[0].text, 'value1')
+    assert.equal(parts[1].name, 'field2')
+    assert.equal(parts[1].text, 'value2')
   })
 })

packages/multipart-parser/src/lib/multipart.node.ts

@@ -1,46 +1,89 @@
 import type * as http from 'node:http'
 import { Readable } from 'node:stream'
+import type { ReadableStream as NodeReadableStream } from 'node:stream/web'
 
-import type { ParseMultipartOptions, MultipartParserOptions, MultipartPart } from './multipart.ts'
+import type {
+  ParseMultipartOptions,
+  MultipartParserOptions,
+  BufferedMultipartPart,
+  StreamedMultipartPart as StreamedMultipartPartWeb
+} from './multipart.ts'
 import {
   MultipartParseError,
   parseMultipart as parseMultipartWeb,
   parseMultipartStream as parseMultipartStreamWeb,
+  MultipartPart
 } from './multipart.ts'
 import { getMultipartBoundary } from './multipart-request.ts'
+/**
+ * A part of a `multipart/*` HTTP message with content as Readable.
+ */
+export class StreamedMultipartPart extends MultipartPart {
+  #webMultipartPart: StreamedMultipartPartWeb
+  /**
+   * Readable of raw content of this part.
+   */
+  readonly contentReadable: Readable
 
+  constructor(webMultipartPart: StreamedMultipartPartWeb) {
+    super(webMultipartPart.rawHeader)
+    this.contentReadable = Readable.fromWeb(webMultipartPart.content as NodeReadableStream<Uint8Array>)
+    this.#webMultipartPart = webMultipartPart
+  }
+  /**
+   * Consumes stream of content into buffered content,
+   * that could be used to create Blob
+   * 
+   * Note: This will throw if stream is started thus buffered can't be complete
+   * check if content is consumed
+   */
+  async toBuffered(): Promise<BufferedMultipartPart> {
+    return this.#webMultipartPart.toBufferedFromIterator(this.contentReadable)
+  }
+  /**
+   * Signal end-of-stream
+   */
+  close() {
+    this.#webMultipartPart.close()
+  }
+}
 /**
- * Parse a `multipart/*` Node.js `Buffer` and yield each part as a `MultipartPart` object.
+ * Parse a `multipart/*` Node.js `Buffer` and yield each part as a `BufferedMultipartPart` object.
  *
  * Note: This is a low-level API that requires manual handling of the content and boundary. If you're
  * building a web server, consider using `parseMultipartRequest(request)` instead.
  *
  * @param message The multipart message as a `Buffer` or an iterable of `Buffer` chunks
  * @param options Options for the parser
- * @return A generator yielding `MultipartPart` objects
+ * @return A generator yielding `BufferedMultipartPart` objects
  */
-export function* parseMultipart(
+export async function* parseMultipart(
   message: Buffer | Iterable<Buffer>,
   options: ParseMultipartOptions,
-): Generator<MultipartPart, void, unknown> {
+): AsyncGenerator<BufferedMultipartPart, void, unknown> {
   yield* parseMultipartWeb(message as Uint8Array | Iterable<Uint8Array>, options)
 }
 
 /**
- * Parse a `multipart/*` Node.js `Readable` stream and yield each part as a `MultipartPart` object.
+ * Parse a `multipart/*` Node.js `Readable` stream and yield each part as a `StreamedMultipartPart` object.
  *
  * Note: This is a low-level API that requires manual handling of the stream and boundary. If you're
  * building a web server, consider using `parseMultipartRequest(request)` instead.
  *
  * @param stream A Node.js `Readable` stream containing multipart data
  * @param options Options for the parser
- * @return An async generator yielding `MultipartPart` objects
+ * @return An async generator yielding `StreamedMultipartPart` objects
  */
 export async function* parseMultipartStream(
   stream: Readable,
   options: ParseMultipartOptions,
-): AsyncGenerator<MultipartPart, void, unknown> {
-  yield* parseMultipartStreamWeb(Readable.toWeb(stream) as ReadableStream, options)
+): AsyncGenerator<StreamedMultipartPart, void, unknown> {
+  let asyncParser = parseMultipartStreamWeb(Readable.toWeb(stream) as ReadableStream, options)
+  while (true) {
+    let {value, done} = await asyncParser.next()
+    if (done) break
+    if (value) yield new StreamedMultipartPart(value)
+  }
 }
 
 /**
@@ -55,16 +98,16 @@ export function isMultipartRequest(req: http.IncomingMessage): boolean {
 }
 
 /**
- * Parse a multipart Node.js request and yield each part as a `MultipartPart` object.
+ * Parse a multipart Node.js request and yield each part as a `StreamedMultipartPart` object.
  *
  * @param req The Node.js `http.IncomingMessage` object containing multipart data
  * @param options Options for the parser
- * @return An async generator yielding `MultipartPart` objects
+ * @return An async generator yielding `StreamedMultipartPart` objects
  */
 export async function* parseMultipartRequest(
   req: http.IncomingMessage,
   options?: MultipartParserOptions,
-): AsyncGenerator<MultipartPart, void, unknown> {
+): AsyncGenerator<StreamedMultipartPart, void, unknown> {
   if (!isMultipartRequest(req)) {
     throw new MultipartParseError('Request is not a multipart request')
   }