diff --git a/.eslintrc.json b/.eslintrc.json index 62667f63..3c009c11 100644 --- a/.eslintrc.json +++ b/.eslintrc.json @@ -24,7 +24,8 @@ "@typescript-eslint/no-namespace": "off", "no-case-declarations": "off", "no-extra-semi": "off", - "sonarjs/cognitive-complexity": "off" + "sonarjs/cognitive-complexity": "off", + "sonarjs/no-all-duplicated-branches": "off" }, "ignorePatterns": ["example/*", "test/**/*"] } diff --git a/bun.lockb b/bun.lockb index 36f18b45..dcb0e4a4 100755 Binary files a/bun.lockb and b/bun.lockb differ diff --git a/example/a.ts b/example/a.ts index 6264b132..d42aaddd 100644 --- a/example/a.ts +++ b/example/a.ts @@ -1,35 +1,46 @@ import { Elysia } from '../src' - -const app = new Elysia() - .trace(async ({ onBeforeHandle, onAfterHandle }) => { - await onBeforeHandle(async ({ name, start, stop, children }) => { - for (const onStart of children) - onStart(async ({ name, start, stop }) => { - console.log(name, start, '->', await stop, 'ms') - }) - - console.log('beforeHandle took', (await stop) - start, 'ms') - }) - - await onAfterHandle(async ({ name, start, stop, children }) => { - for (const onStart of children) - onStart(async ({ name, start, stop }) => { - console.log(name, start, '->', await stop, 'ms') - }) - - console.log('afterHandle took', (await stop) - start, 'ms') - }) - }) - .get('/', 'okkkk XD', { - beforeHandle: [ - async function a() { - await Bun.sleep(100) - }, - async function b() { - await Bun.sleep(150) - } - ] - }) - import { req } from '../test/utils' -app.handle(req('/')).then((x) => x.text()) + +const app = new Elysia().get('/', async function* () { + yield 'a' + await Bun.sleep(10) + + yield 'b' + await Bun.sleep(10) + + yield 'c' +}) + +// console.log(app.routes[0].composed?.toString()) + +const controller = new AbortController() + +setTimeout(() => { + controller.abort() +}, 15) + +app.handle( + new Request('http://e.ly', { + signal: controller.signal + }) +) + .then((x) => x.body) + .then((x) => { + if (!x) return + + const reader = x?.getReader() + + let acc = '' + const { promise, resolve } = Promise.withResolvers() + + reader.read().then(function pump({ done, value }): unknown { + if (done) return resolve(acc) + + acc += value.toString() + console.log(value.toString()) + return reader.read().then(pump) + }) + + return promise + }) + .then((x) => console.log('Result:', x)) diff --git a/example/b.ts b/example/b.ts index 972065b0..6264b132 100644 --- a/example/b.ts +++ b/example/b.ts @@ -1,43 +1,35 @@ -import { Elysia, t } from '../src' -import { req, post } from '../test/utils' +import { Elysia } from '../src' -const app = new Elysia({ precompile: true }) - .trace(async ({ parse, transform, beforeHandle, context }) => { - { - const { end, time } = await parse - console.log('Parse', (await end) - time) - } +const app = new Elysia() + .trace(async ({ onBeforeHandle, onAfterHandle }) => { + await onBeforeHandle(async ({ name, start, stop, children }) => { + for (const onStart of children) + onStart(async ({ name, start, stop }) => { + console.log(name, start, '->', await stop, 'ms') + }) - { - const { end, time } = await transform - console.log('Transform', (await end) - time) - } + console.log('beforeHandle took', (await stop) - start, 'ms') + }) - { - const { end, time } = await beforeHandle - console.log('Before Handle', (await end) - time) - } + await onAfterHandle(async ({ name, start, stop, children }) => { + for (const onStart of children) + onStart(async ({ name, start, stop }) => { + console.log(name, start, '->', await stop, 'ms') + }) + + console.log('afterHandle took', (await stop) - start, 'ms') + }) }) - .get( - '/', - (context) => { - console.log("A") - - return 'a' - }, - { - beforeHandle() { - // await new Promise((r) => setTimeout(r, 10)) + .get('/', 'okkkk XD', { + beforeHandle: [ + async function a() { + await Bun.sleep(100) }, - error({ error }) { - console.log(error) + async function b() { + await Bun.sleep(150) } - } - ) + ] + }) -// console.log(app.routes[0].composed.toString()) - -await app.handle(req('/')).then((x) => x.text()) -// .then(console.log) - -// console.log(app.event.trace[0].fn) +import { req } from '../test/utils' +app.handle(req('/')).then((x) => x.text()) diff --git a/package.json b/package.json index 01f64ed3..4214b6e4 100644 --- a/package.json +++ b/package.json @@ -103,6 +103,7 @@ "openapi-types": "^12.1.3" }, "devDependencies": { + "@elysiajs/stream": "^1.0.2", "@types/benchmark": "^2.1.5", "@types/bun": "^1.1.2", "@types/cookie": "^0.6.0", diff --git a/src/handler.ts b/src/handler.ts index 0fe9c3f0..c4066cd3 100644 --- a/src/handler.ts +++ b/src/handler.ts @@ -111,6 +111,50 @@ export const serializeCookie = (cookies: Context['set']['cookie']) => { return set } +// const concatUint8Array = (a: Uint8Array, b: Uint8Array) => { +// const arr = new Uint8Array(a.length + b.length) +// arr.set(a, 0) +// arr.set(b, a.length) + +// return arr +// } + +const handleStream = ( + generator: Generator | AsyncGenerator, + request?: Request +) => { + let end = false + + return new Response( + new ReadableStream({ + async start(controller) { + request?.signal.addEventListener('abort', () => { + end = true + + try { + controller.close() + } catch { + // nothing + } + }) + + for await (const chunk of generator) { + if (end) break + if (!chunk) continue + + controller.enqueue(Buffer.from(chunk.toString())) + } + + try { + controller.close() + } catch { + // nothing + } + } + }) + ) +} + export const mapResponse = ( response: unknown, set: Context['set'], @@ -324,6 +368,10 @@ export const mapResponse = ( if (response instanceof Error) return errorToResponse(response as Error, set) + // @ts-expect-error + if (typeof response?.next === 'function') + return handleStream(response as any, request) + if ('charCodeAt' in (response as any)) { const code = (response as any).charCodeAt(0) @@ -408,7 +456,7 @@ export const mapResponse = ( case 'Promise': // @ts-ignore return (response as any as Promise).then((x) => { - const r = mapCompactResponse(x) + const r = mapCompactResponse(x, request) if (r !== undefined) return r @@ -417,7 +465,7 @@ export const mapResponse = ( // ? Maybe response or Blob case 'Function': - return mapCompactResponse((response as Function)()) + return mapCompactResponse((response as Function)(), request) case 'Number': case 'Boolean': @@ -446,6 +494,10 @@ export const mapResponse = ( if (response instanceof Error) return errorToResponse(response as Error, set) + // @ts-expect-error + if (typeof response?.next === 'function') + return handleStream(response as any, request) + if ('charCodeAt' in (response as any)) { const code = (response as any).charCodeAt(0) @@ -677,6 +729,10 @@ export const mapEarlyResponse = ( if (response instanceof Error) return errorToResponse(response as Error, set) + // @ts-expect-error + if (typeof response?.next === 'function') + return handleStream(response as any, request) + if ('charCodeAt' in (response as any)) { const code = (response as any).charCodeAt(0) @@ -766,7 +822,7 @@ export const mapEarlyResponse = ( return errorToResponse(response as Error, set) case 'Function': - return mapCompactResponse((response as Function)()) + return mapCompactResponse((response as Function)(), request) case 'Number': case 'Boolean': @@ -795,6 +851,10 @@ export const mapEarlyResponse = ( if (response instanceof Error) return errorToResponse(response as Error, set) + // @ts-expect-error + if (typeof response?.next === 'function') + return handleStream(response as any, request) + if ('charCodeAt' in (response as any)) { const code = (response as any).charCodeAt(0) @@ -896,13 +956,13 @@ export const mapCompactResponse = ( case 'Promise': // @ts-ignore - return (response as any as Promise).then( - mapCompactResponse + return (response as any as Promise).then((x) => + mapCompactResponse(x, request) ) // ? Maybe response or Blob case 'Function': - return mapCompactResponse((response as Function)()) + return mapCompactResponse((response as Function)(), request) case 'Number': case 'Boolean': @@ -920,11 +980,17 @@ export const mapCompactResponse = ( }) if (response instanceof Promise) - return response.then(mapCompactResponse) as any + return response.then((x) => + mapCompactResponse(x, request) + ) as any if (response instanceof Error) return errorToResponse(response as Error) + // @ts-expect-error + if (typeof response?.next === 'function') + return handleStream(response as any, request) + if ('charCodeAt' in (response as any)) { const code = (response as any).charCodeAt(0) diff --git a/test/response/stream.test.ts b/test/response/stream.test.ts new file mode 100644 index 00000000..08002327 --- /dev/null +++ b/test/response/stream.test.ts @@ -0,0 +1,96 @@ +import { describe, it, expect } from 'bun:test' +import { req } from '../utils' + +import { Elysia } from '../../src' + +describe('Stream', () => { + it('handle stream', async () => { + const expected = ['a', 'b', 'c'] + + const app = new Elysia().get('/', async function* () { + yield 'a' + await Bun.sleep(10) + + yield 'b' + await Bun.sleep(10) + + yield 'c' + }) + + const response = await app + .handle(req('/')) + .then((x) => x.body) + .then((x) => { + if (!x) return + + const reader = x?.getReader() + + let acc = '' + const { promise, resolve } = Promise.withResolvers() + + reader.read().then(function pump({ done, value }): unknown { + if (done) return resolve(acc) + + expect(value.toString()).toBe(expected.shift()!) + + acc += value.toString() + return reader.read().then(pump) + }) + + return promise + }) + + expect(expected).toHaveLength(0) + expect(response).toBe('abc') + }) + + it('stop stream on canceled request', async () => { + const expected = ['a', 'b'] + + const app = new Elysia().get('/', async function* () { + yield 'a' + await Bun.sleep(10) + + yield 'b' + await Bun.sleep(10) + + yield 'c' + }) + + const controller = new AbortController() + + setTimeout(() => { + controller.abort() + }, 15) + + const response = await app + .handle( + new Request('http://e.ly', { + signal: controller.signal + }) + ) + .then((x) => x.body) + .then((x) => { + if (!x) return + + const reader = x?.getReader() + + let acc = '' + const { promise, resolve } = Promise.withResolvers() + + reader.read().then(function pump({ done, value }): unknown { + if (done) return resolve(acc) + + expect(value.toString()).toBe(expected.shift()!) + + acc += value.toString() + return reader.read().then(pump) + }) + + return promise + }) + + expect(expected).toHaveLength(0) + expect(response).toBe('ab') + }) +})