🎉 feat: refactor sse

Author: SaltyAom

CHANGELOG.md

@@ -1,3 +1,13 @@
+# 1.3.16 - 22 Aug 2025
+Improvement:
+- `sse` now infer type
+- `sse` now accepts `ReadableStream` to return stream as `text/event-stream`
+- refactor SSE handler
+- support returning `ReadbleStream` from generator or async generator
+
+Bug fix:
+- static response now use callback clone instead of bind
+
 # 1.3.15 - 21 Aug 2025
 Bug fix:
 - ValidationError.detail only handle custom error

example/a.ts

@@ -1,14 +1,23 @@
-import { Elysia, t, validationDetail } from '../src'
+import { Elysia, sse, t } from '../src'
+import { streamResponse } from '../src/adapter/utils'
+import { req } from '../test/utils'
 
-new Elysia()
-	.onError(({ error, code }) => {
-		if (code === 'VALIDATION') return error.detail(error.message)
-	})
-	.post('/', () => 'Hello World!', {
-		body: t.Object({
-			x: t.Number({
-				error: 'x must be a number'
-			}),
+const app = new Elysia()
+	.get('/', function () {
+		return new ReadableStream({
+			async start(controller) {
+				controller.enqueue('a')
+				await Bun.sleep(100)
+				controller.enqueue('b')
+				await Bun.sleep(100)
+				controller.close()
+			}
 		})
 	})
 	.listen(3000)
+
+const response = await app.handle(req('/'))
+
+for await (const a of streamResponse(response)) {
+	console.log(a)
+}

src/adapter/bun/handler-native.ts

@@ -39,6 +39,6 @@ export const createNativeStaticHandler = (
 		if (!response.headers.has('content-type'))
 			response.headers.append('content-type', 'text/plain')
 
-		return () => response.clone() as any
+		return () => response.clone() as Response
 	}
 }

src/adapter/bun/handler.ts

@@ -664,7 +664,7 @@ export const createStaticHandler = (
 		!hooks.beforeHandle?.length &&
 		!hooks.afterHandle?.length
 	)
-		return response.clone.bind(response) as any
+		return () => response.clone() as Response
 }
 
 const handleResponse = createResponseHandler({

src/adapter/utils.ts

@@ -147,28 +147,48 @@ type CreateHandlerParameter = {
 	mapCompactResponse(response: unknown, request?: Request): Response
 }
 
+const asyncIterator = Symbol.asyncIterator
+
 export const createStreamHandler =
 	({ mapResponse, mapCompactResponse }: CreateHandlerParameter) =>
 	async (
-		generator: Generator | AsyncGenerator,
+		generator: Generator | AsyncGenerator | ReadableStream,
 		set?: Context['set'],
 		request?: Request
 	) => {
-		let init = generator.next()
+		// Since ReadableStream doesn't have next, init might be undefined
+		let init = (generator as Generator).next?.() as
+			| IteratorResult<unknown>
+			| undefined
+
 		if (init instanceof Promise) init = await init
 
-		if (typeof init?.done === 'undefined' || init?.done) {
+		// Generator or ReadableStream is returned from a generator function
+		if (init?.value instanceof ReadableStream) {
+			// @ts-ignore
+			generator = init.value
+		} else if (init && (typeof init?.done === 'undefined' || init?.done)) {
 			if (set) return mapResponse(init.value, set, request)
 			return mapCompactResponse(init.value, request)
 		}
 
-		const contentType =
-			// @ts-ignore
-			init.value && typeof init.value?.stream
-				? 'text/event-stream'
-				: init.value && typeof init.value === 'object'
-					? 'application/json'
-					: 'text/plain'
+		const isSSE =
+			// @ts-ignore First SSE result is wrapped with sse()
+			init?.value?.sse ??
+			// @ts-ignore ReadableStream is wrapped with sse()
+			generator?.sse ??
+			// User explicitly set content-type to SSE
+			set?.headers['content-type']?.startsWith('text/event-stream')
+
+		const format = isSSE
+			? (data: string) => `data: ${data}\n\n`
+			: (data: string) => data
+
+		const contentType = isSSE
+			? 'text/event-stream'
+			: init?.value && typeof init?.value === 'object'
+				? 'application/json'
+				: 'text/plain'
 
 		if (set?.headers) {
 			if (!set.headers['transfer-encoding'])
@@ -177,7 +197,7 @@ export const createStreamHandler =
 				set.headers['content-type'] = contentType
 			if (!set.headers['cache-control'])
 				set.headers['cache-control'] = 'no-cache'
-		} else {
+		} else
 			set = {
 				status: 200,
 				headers: {
@@ -187,7 +207,6 @@ export const createStreamHandler =
 					connection: 'keep-alive'
 				}
 			}
-		}
 
 		return new Response(
 			new ReadableStream({
@@ -199,57 +218,58 @@ export const createStreamHandler =
 
 						try {
 							controller.close()
-						} catch {
-							// nothing
-						}
+						} catch {}
 					})
 
-					if (init.value !== undefined && init.value !== null) {
+					if (!init || init.value instanceof ReadableStream) {
+					} else if (
+						init.value !== undefined &&
+						init.value !== null
+					) {
 						// @ts-ignore
-						if (init.value.toStream)
+						if (init.value.toSSE)
 							// @ts-ignore
-							controller.enqueue(init.value.toStream())
+							controller.enqueue(init.value.toSSE())
 						else if (typeof init.value === 'object')
 							try {
 								controller.enqueue(
-									Buffer.from(JSON.stringify(init.value))
+									format(JSON.stringify(init.value))
 								)
 							} catch {
 								controller.enqueue(
-									Buffer.from(init.value.toString())
+									format(init.value.toString())
 								)
 							}
-						else
-							controller.enqueue(
-								Buffer.from(init.value.toString())
-							)
+						else controller.enqueue(format(init.value.toString()))
 					}
 
-					for await (const chunk of generator) {
-						if (end) break
-						if (chunk === undefined || chunk === null) continue
+					try {
+						for await (const chunk of generator) {
+							if (end) break
+							if (chunk === undefined || chunk === null) continue
 
-						// @ts-ignore
-						if (chunk.toStream)
 							// @ts-ignore
-							controller.enqueue(chunk.toStream())
-						else if (typeof chunk === 'object')
-							try {
-								controller.enqueue(
-									Buffer.from(JSON.stringify(chunk))
-								)
-							} catch {
-								controller.enqueue(
-									Buffer.from(chunk.toString())
-								)
-							}
-						else controller.enqueue(Buffer.from(chunk.toString()))
-
-						// Wait for the next event loop
-						// Otherwise the data will be mixed up
-						await new Promise<void>((resolve) =>
-							setTimeout(() => resolve(), 0)
-						)
+							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 controller.enqueue(format(chunk.toString()))
+
+							// Wait for the next event loop
+							// Otherwise the data will be mixed up
+							await new Promise<void>((resolve) =>
+								setTimeout(() => resolve(), 0)
+							)
+						}
+					} catch (error) {
+						console.warn(error)
 					}
 
 					try {
@@ -276,7 +296,8 @@ export async function* streamResponse(response: Response) {
 			const { done, value } = await reader.read()
 			if (done) break
 
-			yield decoder.decode(value)
+			if (typeof value === 'string') yield value
+			else yield decoder.decode(value)
 		}
 	} finally {
 		reader.releaseLock()

src/adapter/web-standard/handler.ts

@@ -699,7 +699,7 @@ export const createStaticHandler = (
 		!hooks.beforeHandle?.length &&
 		!hooks.afterHandle?.length
 	)
-		return response.clone.bind(response) as any
+		return () => response.clone() as Response
 }
 
 const handleResponse = createResponseHandler({

src/types.ts

@@ -1612,14 +1612,14 @@ type _CreateEden<
 				[x in Path]: Property
 			}
 
-type RemoveStartinSlash<T> = T extends `/${infer Rest}` ? Rest : T
+type RemoveStartingSlash<T> = T extends `/${infer Rest}` ? Rest : T
 
 export type CreateEden<
 	Path extends string,
 	Property extends Record<string, unknown> = {}
 > = Path extends '' | '/'
 	? Property
-	: _CreateEden<RemoveStartinSlash<Path>, Property>
+	: _CreateEden<RemoveStartingSlash<Path>, Property>
 
 export type ComposeElysiaResponse<
 	Schema extends RouteSchema,
@@ -1652,7 +1652,9 @@ type _ComposeElysiaResponse<Schema extends RouteSchema, Handle> = Prettify<
 							  >['response']
 					: Handle extends Generator<infer A, infer B, infer C>
 						? AsyncGenerator<A, B, C>
-						: Replace<Handle, ElysiaFile, File>
+						: Handle extends ReadableStream<infer A>
+							? AsyncGenerator<A, void, unknown>
+							: Replace<Handle, ElysiaFile, File>
 			}) &
 		ExtractErrorFromHandle<Handle> &
 		({} extends Omit<Schema['response'], 200>
@@ -1843,8 +1845,9 @@ type SetContentType =
 	| 'application/zip'
 	| 'text/css'
 	| 'text/csv'
-	| 'text/html'
 	| 'text/calendar'
+	| 'text/event-stream'
+	| 'text/html'
 	| 'text/javascript'
 	| 'text/plain'
 	| 'text/xml'
@@ -2036,13 +2039,16 @@ export type MergeTypeModule<
 	B extends TModule<any, any>
 > = TModule<Prettify<UnwrapTypeModule<A> & UnwrapTypeModule<B>>>
 
-export type SSEPayload = {
+export type SSEPayload<
+	Data extends unknown = unknown,
+	Event extends string | undefined = string | undefined
+> = {
 	/** id of the event */
 	id?: string | number | null
 	/** event name */
-	event?: string
+	event?: Event
 	/** retry in millisecond */
 	retry?: number
 	/** data to send */
-	data?: unknown
+	data?: Data
 }

src/utils.ts

@@ -19,7 +19,8 @@ import type {
 	AfterResponseHandler,
 	SchemaValidator,
 	AnyLocalHook,
-	SSEPayload
+	SSEPayload,
+	Prettify
 } from './types'
 import { ElysiaFile } from './universal/file'
 
@@ -1129,6 +1130,10 @@ export const supportPerMethodInlineHandler = (() => {
 	return true
 })()
 
+type FormatSSEPayload<T = unknown> = T extends string
+	? { readonly data: T }
+	: Prettify<SSEPayload<T>>
+
 /**
  * Return a Server Sent Events (SSE) payload
  *
@@ -1145,18 +1150,44 @@ export const supportPerMethodInlineHandler = (() => {
  *     })
  *   }
  */
-export const sse = (
-	payload: string | SSEPayload
-): SSEPayload & { toStream(): string } => {
-	if (typeof payload === 'string')
-		payload = {
-			data: payload
-		}
+export const sse = <
+	const T extends
+		| string
+		| SSEPayload
+		| Generator
+		| AsyncGenerator
+		| ReadableStream
+>(
+	_payload: T
+): T extends string
+	? { readonly data: T }
+	: T extends SSEPayload
+		? T
+		: T extends ReadableStream<infer A>
+			? ReadableStream<FormatSSEPayload<A>>
+			: T extends Generator<infer A, infer B, infer C>
+				? Generator<FormatSSEPayload<A>, B, C>
+				: T extends AsyncGenerator<infer A, infer B, infer C>
+					? AsyncGenerator<FormatSSEPayload<A>, B, C>
+					: T => {
+	if (_payload instanceof ReadableStream) {
+		// @ts-expect-error
+		_payload.sse = true
+		return _payload as any
+	}
+
+	const payload: SSEPayload =
+		typeof _payload === 'string'
+			? { data: _payload }
+			: (_payload as SSEPayload)
 
-	if (payload.id === undefined) payload.id = randomId()
+	// if (payload.id === undefined) payload.id = randomId()
+
+	// @ts-ignore
+	payload.sse = true
 
 	// @ts-ignore
-	payload.toStream = () => {
+	payload.toSSE = () => {
 		let payloadString = ''
 
 		if (payload.id !== undefined && payload.id !== null)