📘 fix: #1772 resolve merge conflict

Author: SaltyAom

CHANGELOG.md

@@ -2,6 +2,7 @@
 Feature:
 - [#1803](https://github.com/elysiajs/elysia/pull/1803) stream response with pull based backpressure
 - [#1802](https://github.com/elysiajs/elysia/pull/1802) handle range header for file/blob response
+- [#1722](https://github.com/elysiajs/elysia/pull/1772), [#1741](https://github.com/elysiajs/elysia/issues/1741) direct ReadableStream perf blow-up
 
 Bug fix:
 - [#1805](https://github.com/elysiajs/elysia/pull/1805) dynamic imports inside .guard not registering routes
@@ -10,6 +11,8 @@ Bug fix:
 - [#1794](https://github.com/elysiajs/elysia/pull/1794) merge app cookie config into route cookie validator config
 - [#1796](https://github.com/elysiajs/elysia/pull/1796) check custom parser by full name
 - [#1795](https://github.com/elysiajs/elysia/pull/1795)  write transformed cookie value to cookie entry directly
+- [#1793](https://github.com/elysiajs/elysia/pull/1793) use cookie schema for cookie noValidate check
+- [#1792](https://github.com/elysiajs/elysia/pull/1792) throw ValidationError instead of boolean in response encode path
 - detect HTML bundle when inline response is Promise
 
 Change:

example/a.ts

@@ -1,28 +1,34 @@
 import { Elysia, t } from '../src'
-import html from '../example/index.html'
+import { req } from '../test/utils'
 
 const app = new Elysia()
-	.get(
-		'/',
-		({ cookie: { thing } }) => {
-			console.log(typeof thing.value)
+	.get('/', async () => {
+		const file = Bun.file('test/kyuukurarin.mp4')
 
-			return thing.value
-		},
-		{
-			cookie: t.Object({
-				thing: t.Number()
-			})
-		}
-	)
-	.listen(3000)
+		// Wrap the stream in another ReadableStream
+		// perhaps we are concatenating streams or whatever
+		const body = new ReadableStream({
+			async start(controller) {
+				const reader = file.stream().getReader()
+				try {
+					while (true) {
+						const { done, value } = await reader.read()
+						if (done) break
+						controller.enqueue(value)
+					}
+					controller.close()
+				} catch (err) {
+					controller.error(err)
+				} finally {
+					reader.releaseLock()
+				}
+			}
+		})
+
+		// Returning the stream uses 100% for several minutes
+		return body
 
-app.handle(
-	new Request('http://localhost:3000/', {
-		headers: {
-			cookie: 'thing=9'
-		}
+		// Returning the same stream wrapped in a Response servers the stream in a fraction of a second
+		// return new Response(body);
 	})
-)
-	.then((response) => response.json())
-	.then(console.log)
+	.listen(3000)

src/adapter/utils.ts

@@ -4,6 +4,7 @@ import { hasHeaderShorthand, isNotEmpty, StatusMap } from '../utils'
 import type { Context } from '../context'
 import { env } from '../universal'
 import { isBun } from '../universal/utils'
+import { MaybePromise } from '../types'
 
 export const handleFile = (
 	response: File | Blob,
@@ -65,7 +66,11 @@ export const handleFile = (
 			return new Response(
 				(
 					response as unknown as {
-						slice(start: number, end: number, contentType?: string): Blob
+						slice(
+							start: number,
+							end: number,
+							contentType?: string
+						): Blob
 					}
 				).slice(start, end + 1, response.type),
 				{
@@ -208,7 +213,35 @@ interface CreateHandlerParameter {
 	mapCompactResponse(response: unknown, request?: Request): Response
 }
 
-const allowRapidStream = env.ELYSIA_RAPID_STREAM === 'true'
+const enqueueBinaryChunk = (
+	controller: ReadableStreamDefaultController,
+	chunk: unknown
+): MaybePromise<boolean> => {
+	if (chunk instanceof Blob)
+		return chunk.arrayBuffer().then((buffer) => {
+			controller.enqueue(new Uint8Array(buffer))
+			return true as const
+		})
+
+	if (chunk instanceof Uint8Array) {
+		controller.enqueue(chunk)
+		return true
+	}
+
+	if (chunk instanceof ArrayBuffer) {
+		controller.enqueue(new Uint8Array(chunk))
+		return true
+	}
+
+	if (ArrayBuffer.isView(chunk)) {
+		controller.enqueue(
+			new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength)
+		)
+		return true
+	}
+
+	return false
+}
 
 export const createStreamHandler =
 	({ mapResponse, mapCompactResponse }: CreateHandlerParameter) =>
@@ -301,27 +334,28 @@ export const createStreamHandler =
 					// Enqueue the already-extracted init value (first generator
 					// result, used above for SSE detection). Subsequent values
 					// are produced on-demand by pull().
-					if (!init || init.value instanceof ReadableStream) {
-					} else if (
-						init.value !== undefined &&
-						init.value !== null
-					) {
+					if (
+						!init ||
+						init.value instanceof ReadableStream ||
+						init.value === undefined ||
+						init.value === null
+					)
+						return
+
+					// @ts-ignore
+					if (init.value.toSSE)
 						// @ts-ignore
-						if (init.value.toSSE)
-							// @ts-ignore
-							controller.enqueue(init.value.toSSE())
-						else if (typeof init.value === 'object')
-							try {
-								controller.enqueue(
-									format(JSON.stringify(init.value))
-								)
-							} catch {
-								controller.enqueue(
-									format(init.value.toString())
-								)
-							}
-						else controller.enqueue(format(init.value.toString()))
-					}
+						controller.enqueue(init.value.toSSE())
+					else if (enqueueBinaryChunk(controller, init.value)) return
+					else if (typeof init.value === 'object')
+						try {
+							controller.enqueue(
+								format(JSON.stringify(init.value))
+							)
+						} catch {
+							controller.enqueue(format(init.value.toString()))
+						}
+					else controller.enqueue(format(init.value.toString()))
 				},
 
 				async pull(controller) {
@@ -348,23 +382,19 @@ export const createStreamHandler =
 						if (chunk === undefined || chunk === null) return
 
 						// @ts-ignore
-						if (chunk.toSSE) {
+						if (chunk.toSSE)
 							// @ts-ignore
 							controller.enqueue(chunk.toSSE())
-						} else {
-							if (typeof chunk === 'object')
-								try {
-									controller.enqueue(
-										format(JSON.stringify(chunk))
-									)
-								} catch {
-									controller.enqueue(
-										format(chunk.toString())
-									)
-								}
-							else
+						else if (enqueueBinaryChunk(controller, chunk)) return
+						else if (typeof chunk === 'object')
+							try {
+								controller.enqueue(
+									format(JSON.stringify(chunk))
+								)
+							} catch {
 								controller.enqueue(format(chunk.toString()))
-						}
+							}
+						else controller.enqueue(format(chunk.toString()))
 					} catch (error) {
 						console.warn(error)
 

test/response/stream.test.ts

@@ -707,4 +707,70 @@ describe('Stream', () => {
 
 		reader.cancel()
 	})
+
+	it('stream ReadableStream binary chunks', async () => {
+		const payload = new Uint8Array(128 * 1024)
+
+		for (let i = 0; i < payload.length; i++) payload[i] = i % 251
+
+		const app = new Elysia().get(
+			'/',
+			() =>
+				new ReadableStream({
+					start(controller) {
+						controller.enqueue(payload.subarray(0, 32768))
+						controller.enqueue(payload.subarray(32768, 65536))
+						controller.enqueue(payload.subarray(65536, 98304))
+						controller.enqueue(payload.subarray(98304))
+						controller.close()
+					}
+				})
+		)
+
+		const response = await app.handle(req('/'))
+		const result = new Uint8Array(await response.arrayBuffer())
+
+		expect(result.byteLength).toBe(payload.byteLength)
+		expect(result).toEqual(payload)
+	})
+
+	it('stream ReadableStream binary views and blob chunks', async () => {
+		const source = new Uint8Array([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
+		const expected = new Uint8Array([6, 7, 8, 9, 2, 3, 4, 5, 10, 11, 12])
+
+		const app = new Elysia().get(
+			'/',
+			() =>
+				new ReadableStream({
+					start(controller) {
+						controller.enqueue(source.subarray(6, 10))
+						controller.enqueue(new DataView(source.buffer, 2, 4))
+						controller.enqueue(
+							new Blob([new Uint8Array([10, 11, 12])])
+						)
+						controller.close()
+					}
+				})
+		)
+
+		const response = await app.handle(req('/'))
+		const result = new Uint8Array(await response.arrayBuffer())
+
+		expect(result).toEqual(expected)
+	})
+
+	it('stream generator Uint8Array chunks as binary', async () => {
+		const app = new Elysia().get('/', async function* () {
+			yield new Uint8Array([1, 2])
+			await Bun.sleep(1)
+			yield new Uint8Array([3, 4])
+		})
+
+		const response = await app.handle(req('/'))
+		const result = new Uint8Array(await response.arrayBuffer())
+
+		// expect(result).toEqual(result.toBase64())
+
+		expect(result).toEqual(new Uint8Array([1, 2, 3, 4]))
+	})
 })