feat: sse (#85)

* feat: sse

* lint: deno lint + fmt

* Update router.ts

Co-authored-by: Marco Antônio <[email protected]>

* fix: import in deps.ts

* chore: ignore weird ts error

* fix: improve how we call router function

---------

Co-authored-by: Marco Antônio <[email protected]>

deno.json

@@ -1,7 +1,7 @@
 {
-  "name": "@danet/core",
-  "version": "2.3.0",
-  "exports": "./mod.ts",
+	"name": "@danet/core",
+	"version": "2.3.0",
+	"exports": "./mod.ts",
 	"lint": {
 		"files": {
 			"include": ["src/"],

spec/sse.test.ts

@@ -0,0 +1,65 @@
+import { Controller, Module, SSE } from '../mod.ts';
+import { assertEquals } from '../src/deps_test.ts';
+import { DanetApplication } from '../src/app.ts';
+import { SSEEvent } from '../src/sse/event.ts';
+
+@Controller()
+class SSEExampleController {
+	@SSE('sse')
+	sendUpdate(): EventTarget {
+		const eventTarget = new EventTarget();
+		let id = 0;
+		const interval = setInterval(() => {
+			if (id >= 4) {
+				clearInterval(interval);
+				const event = new SSEEvent({
+					retry: 1000,
+					id: `${id}`,
+					data: 'close',
+					event: 'close',
+				});
+				eventTarget.dispatchEvent(event);
+				return;
+			}
+			const event = new SSEEvent({
+				retry: 1000,
+				id: `${id}`,
+				data: 'world',
+				event: 'hello',
+			});
+			eventTarget.dispatchEvent(event);
+			id++;
+		}, 100);
+		return eventTarget;
+	}
+}
+
+@Module({
+	controllers: [SSEExampleController],
+})
+class ExampleModule {}
+
+Deno.test('Body', async () => {
+	return new Promise<void>(async (resolve, reject) => {
+		const app = new DanetApplication();
+		await app.init(ExampleModule);
+		const listenEvent = await app.listen(0);
+		let eventReceived = 0;
+		const eventSource = new EventSource(
+			`http://localhost:${listenEvent.port}/sse`,
+		);
+
+		eventSource.addEventListener('hello', async (event) => {
+			if (event.data === 'world') {
+				eventReceived++;
+			}
+		});
+
+		eventSource.addEventListener('close', async (event) => {
+			assertEquals(eventReceived, 4);
+			await eventSource.close();
+			await app.close();
+			resolve();
+		});
+	});
+});

src/app.ts

@@ -16,7 +16,7 @@ import { PossibleMiddlewareType } from './router/middleware/decorator.ts';
 import { globalMiddlewareContainer } from './router/middleware/global-container.ts';
 import { ModuleConstructor } from './module/constructor.ts';
 import { serveStatic } from './utils/serve-static.ts';
-import { cors } from 'https://deno.land/x/hono/middleware.ts';
+import { cors } from './deps.ts';
 import { DynamicModule } from './mod.ts';
 
 type CORSOptions = {

src/deps.ts

@@ -11,12 +11,14 @@ export {
 	Hono as Application,
 	type MiddlewareHandler,
 	type Next,
-} from 'https://deno.land/x/hono/mod.ts';
-export { type HandlerInterface } from 'https://deno.land/x/hono/types.ts';
-export { HonoRequest } from 'https://deno.land/x/hono/request.ts';
-export { getPath } from 'https://deno.land/x/hono/utils/url.ts';
+} from 'https://deno.land/x/hono@v4.1.5/mod.ts';
+export { type HandlerInterface } from 'https://deno.land/x/hono@v4.1.5/types.ts';
+export { HonoRequest } from 'https://deno.land/x/hono@v4.1.5/request.ts';
+export { getPath } from 'https://deno.land/x/hono@v4.1.5/utils/url.ts';
 export {
 	RegExpRouter,
 	SmartRouter,
 	TrieRouter,
-} from 'https://deno.land/x/hono/mod.ts';
+} from 'https://deno.land/x/[email protected]/mod.ts';
+export { SSEStreamingApi, streamSSE } from 'https://deno.land/x/[email protected]/helper.ts';
+export { cors } from 'https://deno.land/x/[email protected]/middleware.ts'

src/router/controller/decorator.ts

@@ -31,3 +31,11 @@ export const Delete = createMappingDecorator('DELETE');
 export const Options = createMappingDecorator('OPTIONS');
 export const Head = createMappingDecorator('HEAD');
 export const All = createMappingDecorator();
+export const SSE = (endpoint = ''): MethodDecorator => {
+	return (_target, _propertyKey, descriptor) => {
+		MetadataHelper.setMetadata('endpoint', endpoint, descriptor.value);
+		MetadataHelper.setMetadata('method', 'GET', descriptor.value);
+		MetadataHelper.setMetadata('SSE', true, descriptor.value);
+		return descriptor;
+	};
+};

src/router/router.ts

@@ -1,4 +1,10 @@
-import { Application, Context, type HandlerInterface } from '../deps.ts';
+import {
+	Application,
+	Context,
+	type HandlerInterface,
+	SSEStreamingApi,
+	streamSSE,
+} from '../deps.ts';
 
 import { FilterExecutor } from '../exception/filter/executor.ts';
 import { HTTP_STATUS } from '../exception/http/enum.ts';
@@ -12,14 +18,12 @@ import { HandlebarRenderer } from '../renderer/handlebar.ts';
 import { Renderer } from '../renderer/interface.ts';
 import { Constructor } from '../utils/constructor.ts';
 import { ControllerConstructor } from './controller/constructor.ts';
-import {
-	argumentResolverFunctionsMetadataKey,
-	Resolver,
-} from './controller/params/decorators.ts';
 import { trimSlash } from './utils.ts';
 import { MiddlewareExecutor } from './middleware/executor.ts';
 import { NextFunction } from './middleware/decorator.ts';
 import { resolveMethodParam } from './controller/params/resolver.ts';
+import { SSEMessage } from '../sse/message.ts';
+import { SSEEvent } from '../sse/event.ts';
 
 // deno-lint-ignore no-explicit-any
 export type Callback = (...args: any[]) => unknown;
@@ -101,8 +105,7 @@ export class DanetHTTPRouter {
 				path ? path : '/'
 			}`,
 		);
-		routerFn.call(
-			this.router,
+		routerFn(
 			path,
 			async (context: HttpContext, next: NextFunction) => {
 				const _id = crypto.randomUUID();
@@ -180,6 +183,14 @@ export class DanetHTTPRouter {
 					| string = await controllerInstance[ControllerMethod.name](
 						...params,
 					);
+				const isSSE = MetadataHelper.getMetadata('SSE', ControllerMethod);
+				if (isSSE) {
+					context.res = this.handleSSE(
+						executionContext,
+						response as unknown as EventTarget,
+					);
+					return context.res;
+				}
 				return await this.sendResponse(
 					response,
 					ControllerMethod,
@@ -196,6 +207,31 @@ export class DanetHTTPRouter {
 		};
 	}
 
+	private handleSSE(executionContext: ExecutionContext, response: EventTarget) {
+		return streamSSE(executionContext, async (stream: SSEStreamingApi) => {
+			let canContinue = true;
+			response.addEventListener(
+				'message',
+				async (event) => {
+					const { detail: payload } = event as SSEEvent;
+					await stream.writeSSE({
+						data: payload.data,
+						event: payload.event,
+						id: payload.id,
+						retry: payload.retry,
+					});
+					if (payload.event === 'close') {
+						canContinue = false;
+					}
+				},
+			);
+			while (canContinue) {
+				await stream.sleep(1);
+			}
+			await stream.close();
+		});
+	}
+
 	private async handleError(
 		executionContext: ExecutionContext,
 		// deno-lint-ignore no-explicit-any

src/sse/event.ts

@@ -0,0 +1,7 @@
+import { SSEMessage } from './message.ts';
+
+export class SSEEvent extends CustomEvent<SSEMessage> {
+	constructor(message: SSEMessage) {
+		super('message', { detail: message });
+	}
+}

src/sse/message.ts

@@ -0,0 +1,6 @@
+export interface SSEMessage {
+	data: string;
+	event?: string;
+	id?: string;
+	retry?: number;
+}