PEER-222: Add IPC for ws message sending

Signed-off-by: SeeuSim <[email protected]>

Author: SeeuSim

backend/matching/src/index.ts

@@ -12,7 +12,9 @@ const port = Number.parseInt(EXPRESS_PORT || '8001');
 const listenMessage = `App listening on port: ${port}`;
 server.listen(port, () => {
   logger.info(listenMessage);
-  ['Cleaner', 'Matcher'].map(initWorker).forEach((process) => workers.push(process));
+  ['Cleaner', 'Matcher']
+    .map((name) => initWorker(name, io))
+    .forEach((process) => workers.push(process));
 });
 
 const shutdown = () => {

backend/matching/src/workers/cleaner.ts

@@ -1,7 +1,8 @@
 import { client } from '@/lib/db';
 import { STREAM_CLEANER, STREAM_GROUP, STREAM_NAME } from '@/lib/db/constants';
 import { decodePoolTicket, getPoolKey } from '@/lib/utils';
-import { io } from '@/server';
+import { MATCH_SVC_EVENT } from '@/ws';
+import { sendNotif } from './common';
 
 const logger = {
   info: (message: unknown) => process.send && process.send(message),
@@ -60,8 +61,8 @@ async function clean() {
 
     if (socketRoom) {
       // Notify client
-      io.sockets.in(socketRoom).emit('FAILED');
-      io.sockets.in(socketRoom).disconnectSockets();
+      sendNotif([socketRoom], MATCH_SVC_EVENT.FAILED);
+      sendNotif([socketRoom], MATCH_SVC_EVENT.DISCONNECT);
     }
   }
 }

backend/matching/src/workers/common.ts

@@ -0,0 +1,12 @@
+import type { IChildProcessMessage, IMatchEvent } from '@/ws';
+
+export const sendNotif = (roomIds: Array<string>, event: IMatchEvent, message?: unknown) => {
+  if (process.send) {
+    const payload: IChildProcessMessage = {
+      rooms: roomIds,
+      event,
+      message,
+    };
+    process.send(payload);
+  }
+};

backend/matching/src/workers/index.ts

@@ -1,13 +1,24 @@
 import { fork } from 'child_process';
 import path from 'path';
+
+import type { Server } from 'socket.io';
+
 import { logger } from '@/lib/utils';
+import { MATCH_SVC_EVENT, type IChildProcessMessage } from '@/ws';
 
-export const initWorker = (name: string) => {
+export const initWorker = (name: string, io: Server) => {
   const lCaseName = name.toLowerCase();
   const worker = fork(path.join(__dirname, `${lCaseName}.js`));
   const upperCaseName = name.replace(/^[A-Za-z]/, (c) => c.toUpperCase());
   worker.on('message', (message) => {
-    logger.info(`[${upperCaseName}]: ${message}`);
+    logger.info(`Received message from '${upperCaseName}': ${JSON.stringify(message)}`);
+
+    const { rooms, event, message: payload } = message as IChildProcessMessage;
+    if (event === MATCH_SVC_EVENT.DISCONNECT) {
+      io.sockets.in(rooms).disconnectSockets();
+      return;
+    }
+    io.sockets.in(rooms).emit(event, payload);
   });
   worker.on('exit', (code) => {
     logger.error(`${upperCaseName} exited with code ${code}.`);

backend/matching/src/workers/matcher.ts

@@ -1,8 +1,9 @@
 import { client } from '@/lib/db';
 import { POOL_INDEX, STREAM_GROUP, STREAM_NAME, STREAM_WORKER } from '@/lib/db/constants';
 import { decodePoolTicket, getPoolKey, getStreamId } from '@/lib/utils';
-import { io } from '@/server';
 import { getMatchItems } from '@/services';
+import { MATCH_SVC_EVENT } from '@/ws';
+import { sendNotif } from './common';
 
 const logger = {
   info: (message: unknown) => process.send && process.send(message),
@@ -48,7 +49,7 @@ async function processMatch(
     } = decodePoolTicket(matched);
 
     // To block cancellation
-    io.sockets.in([matchedSocketPort]).emit('Matching');
+    sendNotif([matchedSocketPort], MATCH_SVC_EVENT.MATCHING);
 
     const matchedStreamId = getStreamId(timestamp);
 
@@ -63,7 +64,8 @@ async function processMatch(
 
     // Notify both sockets
     const { ...matchItems } = getMatchItems();
-    io.sockets.in([requestorSocketPort, matchedSocketPort]).emit(JSON.stringify(matchItems));
+    sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.SUCCESS, matchItems);
+    sendNotif([requestorSocketPort, matchedSocketPort], MATCH_SVC_EVENT.DISCONNECT);
     return true;
   }
 
@@ -105,7 +107,7 @@ async function match() {
       } = decodePoolTicket(matchRequest);
 
       // To Block Cancellation
-      io.sockets.in([requestorSocketPort]).emit('Matching');
+      sendNotif([requestorSocketPort], MATCH_SVC_EVENT.MATCHING);
 
       const clause = [`-@userId:(${requestorUserId})`];
       if (difficulty) {
@@ -164,7 +166,7 @@ async function match() {
 
       if (!hasDifficultyMatch) {
         // To allow cancellation
-        io.sockets.in(requestorSocketPort).emit('Waiting');
+        sendNotif([requestorSocketPort], MATCH_SVC_EVENT.PENDING);
       }
     }
   }

backend/matching/src/ws.ts

@@ -1,15 +1,37 @@
 import http from 'http';
 import { Server } from 'socket.io';
+import { logger } from './lib/utils';
 
 export const createWs = (server: ReturnType<(typeof http)['createServer']>) => {
   const io = new Server(server);
-  io.sockets.on('connection', (socket) => {
-    socket.on('create', (room) => {
+  io.on('connection', (socket) => {
+    logger.info(`${socket.id} connected`);
+
+    socket.on('joinRoom', (room) => {
       socket.join(room);
+      logger.info(`Client joined room: ${room}`);
+      socket.to(room).emit('message', `A new user has joined room: ${room}`);
     });
+    // socket.on('create', (room) => {
+    //   socket.join(room);
+    // });
     socket.on('leave', (room) => {
       socket.leave(room);
     });
   });
   return io;
 };
+
+export const MATCH_SVC_EVENT = {
+  SUCCESS: 'SUCCESS', // When match successful
+  FAILED: 'FAILED', // When match failed
+  PENDING: 'PENDING', // When waiting for match
+  MATCHING: 'MATCHING', // When matching in progress
+  DISCONNECT: 'DISCONNECT', // To disconnect all sockets in room
+} as const;
+export type IMatchEvent = (typeof MATCH_SVC_EVENT)[keyof typeof MATCH_SVC_EVENT];
+export type IChildProcessMessage = {
+  rooms: Array<string>;
+  event: IMatchEvent;
+  message?: unknown;
+};