Set up notifications subscribe/unsubscribe

apps/api/src/app/gateway/gateway.ts

@@ -1,16 +1,61 @@
-import { ConnectedSocket, MessageBody, SubscribeMessage, WebSocketGateway, WebSocketServer } from '@nestjs/websockets';
+import {
+	ConnectedSocket,
+	MessageBody,
+	OnGatewayConnection,
+	OnGatewayDisconnect,
+	SubscribeMessage,
+	WebSocketGateway,
+	WebSocketServer,
+	WsResponse,
+} from '@nestjs/websockets';
 import { Socket } from 'dgram';
+import { Observable, delay, from, map } from 'rxjs';
+import { GenericWsMessage, SubscriptionEvent, SubscriptionMessage } from '@rxjs-ws-demo/api-interfaces';
 
 @WebSocketGateway()
-export class WsGateway {
-	@WebSocketServer() server;
+export class WsGateway implements OnGatewayConnection, OnGatewayDisconnect {
+	@WebSocketServer()
+	server;
 
-	@SubscribeMessage('newMessage')
-	handleMessage(@ConnectedSocket() client: Socket, @MessageBody() payload: any): void {
-		console.log('handleMessage', client, payload);
+	wsClients: Socket[] = [];
+
+	handleDisconnect(client: Socket) {
+		//console.log('handleDisconnect', client);
+
+		this.wsClients = this.wsClients.filter((c) => c !== client);
+
+		console.log('Client disconnected ' + this.wsClients.length);
 	}
 
-	sendToAll(msg: string) {
-		this.server.emit('message', msg);
+	handleConnection(client: Socket, ...args: any[]) {
+		//console.log('handleConnection', args, client['sec-websocket-key']);
+
+		this.wsClients.push(client);
+
+		console.log('Client connected ' + this.wsClients.length);
+	}
+
+	// @SubscribeMessage('newMessage')
+	// handleMessage(@ConnectedSocket() client: Socket, @MessageBody() payload: any): void {
+	// 	console.log('handleMessage', client, payload);
+	// }
+
+	// sendToAll(msg: string) {
+	// 	this.server.emit('message', msg);
+	// }
+
+	@SubscribeMessage('subscriptions')
+	onEvent(client: Socket, data: SubscriptionMessage) {
+		//: Observable<WsResponse<number>> {
+		console.log('onEvent', data);
+
+		this.wsClients.forEach((c) => {
+			c.send(JSON.stringify(data));
+		});
+
+		// return from([1, 2, 3]).pipe(
+		// 	delay(1000),
+		// 	map((item) => ({ event: 'events', data: item * 2 })),
+		// );
 	}
 }

apps/api/src/main.ts

@@ -7,13 +7,17 @@ import { Logger } from '@nestjs/common';
 import { NestFactory } from '@nestjs/core';
 
 import { AppModule } from './app/app.module';
+import { WsAdapter } from '@nestjs/platform-ws';
 
 async function bootstrap() {
 	const app = await NestFactory.create(AppModule);
 	const globalPrefix = 'api';
 	app.setGlobalPrefix(globalPrefix);
+	app.useWebSocketAdapter(new WsAdapter(app));
+
 	const port = process.env.PORT || 3100;
 	await app.listen(port);
+
 	Logger.log(`🚀 Application is running on: http://localhost:${port}/${globalPrefix}`);
 }
 

libs/api-interfaces/src/lib/types.ts

@@ -2,24 +2,40 @@ export interface Message {
 	message: string;
 }
 
-// eslint-disable-next-line @typescript-eslint/no-empty-interface
-export interface WebSocketMessage {}
+// // eslint-disable-next-line @typescript-eslint/no-empty-interface
+// export interface WebSocketMessage {}
 
-export interface GenericMessage extends WebSocketMessage {
-	message: string;
-}
+// export interface GenericMessage extends WebSocketMessage {
+// 	message: string;
+// }
 
-export interface WebSocketSubscriptionMsg extends WebSocketMessage {
-	beginSub: boolean;
-	className: string;
-	eventType: number;
-	session: boolean;
-}
+// export interface WebSocketSubscriptionMsg extends WebSocketMessage {
+// 	beginSub: boolean;
+// 	className: string;
+// 	eventType: number;
+// 	session: boolean;
+// }
 
-export interface NotificationJsonModel {
-	$type: string;
-	note_causer_sessionId: string;
-	readonly note_causer_staffCode: string;
-	note_eventType: number;
-	note_typeName: string;
-}
+// export interface NotificationJsonModel {
+// 	$type: string;
+// 	note_causer_sessionId: string;
+// 	readonly note_causer_staffCode: string;
+// 	note_eventType: number;
+// 	note_typeName: string;
+// }
+
+export type SubscriptionMessage = {
+	eventType: string;
+	isSubscribe: boolean;
+	clientId?: string;
+};
+
+export type SubscriptionEvent = {
+	eventType: string;
+	body: unknown;
+};
+
+export type GenericWsMessage = {
+	event: string;
+	data: SubscriptionMessage;
+};

libs/main-ui/src/lib/demo.component.ts

@@ -9,7 +9,7 @@ import { Message } from '@rxjs-ws-demo/api-interfaces';
 	standalone: true,
 	imports: [CommonModule, MatToolbarModule],
 	template: `
-		<mat-toolbar color="primary">Rxjs Web Sockets Demo </mat-toolbar>
+		<mat-toolbar color="primary">RxJs Web Sockets Demo </mat-toolbar>
 		<div class="wrapper">
 			<h1>Powered by Angular and NestJS</h1>