Merge pull request #78 from ruiqi7/feature/matching-websocket

Set up queues per match criteria

Author: nicolelim02

backend/matching-service/config/rabbitmq.ts

@@ -2,40 +2,69 @@ import amqplib, { Connection } from "amqplib";
 import dotenv from "dotenv";
 import { matchUsers } from "../src/utils/mq_utils";
 import { MatchRequestItem } from "../src/handlers/matchHandler";
+import { Complexities, Categories, Languages } from "../src/utils/constants";
 
 dotenv.config();
 
 let mrConnection: Connection;
-const queue = "match_requests";
+const queues: string[] = [];
+const pendingQueueRequests = new Map<string, Map<string, MatchRequestItem>>();
 
-export const connectRabbitMq = async () => {
+const initQueueNames = () => {
+  for (const complexity of Object.values(Complexities)) {
+    for (const category of Object.values(Categories)) {
+      for (const language of Object.values(Languages)) {
+        queues.push(`${complexity}_${category}_${language}`);
+      }
+    }
+  }
+};
+
+const setUpQueue = async (queueName: string) => {
+  const consumerChannel = await mrConnection.createChannel();
+  await consumerChannel.assertQueue(queueName);
+
+  consumerChannel.consume(queueName, (msg) => {
+    if (msg !== null) {
+      matchUsers(queueName, msg.content.toString());
+      consumerChannel.ack(msg);
+    }
+  });
+};
+
+export const connectToRabbitMq = async () => {
   try {
+    initQueueNames();
     mrConnection = await amqplib.connect(`${process.env.RABBITMQ_ADDR}`);
-    const consumerChannel = await mrConnection.createChannel();
-    await consumerChannel.assertQueue(queue);
-
-    consumerChannel.consume(queue, (msg) => {
-      if (msg !== null) {
-        matchUsers(msg.content.toString());
-        consumerChannel.ack(msg);
-      }
-    });
+    for (const queue of queues) {
+      await setUpQueue(queue);
+      pendingQueueRequests.set(queue, new Map<string, MatchRequestItem>());
+    }
   } catch (error) {
     console.error(error);
     process.exit(1);
   }
 };
 
-export const sendRabbitMq = async (
+export const sendToQueue = async (
+  complexity: string,
+  category: string,
+  language: string,
   data: MatchRequestItem
 ): Promise<boolean> => {
   try {
+    const queueName = `${complexity}_${category}_${language}`;
     const senderChannel = await mrConnection.createChannel();
-    senderChannel.sendToQueue(queue, Buffer.from(JSON.stringify(data)));
-    console.log("Sent to queue:", JSON.stringify(data));
+    senderChannel.sendToQueue(queueName, Buffer.from(JSON.stringify(data)));
     return true;
   } catch (error) {
     console.log(error);
     return false;
   }
 };
+
+export const getPendingRequests = (
+  queueName: string
+): Map<string, MatchRequestItem> => {
+  return pendingQueueRequests.get(queueName)!;
+};

backend/matching-service/server.ts

@@ -2,7 +2,7 @@ import http from "http";
 import app, { allowedOrigins } from "./app.ts";
 import { handleWebsocketMatchEvents } from "./src/handlers/websocketHandler.ts";
 import { Server } from "socket.io";
-import { connectRabbitMq } from "./config/rabbitmq.ts";
+import { connectToRabbitMq } from "./config/rabbitmq.ts";
 
 const server = http.createServer(app);
 export const io = new Server(server, {
@@ -20,7 +20,7 @@ io.on("connection", (socket) => {
 const PORT = process.env.SERVICE_PORT || 3002;
 
 if (process.env.NODE_ENV !== "test") {
-  connectRabbitMq()
+  connectToRabbitMq()
     .then(() => {
       console.log("RabbitMq connected!");
 

backend/matching-service/src/handlers/matchHandler.ts

@@ -1,5 +1,5 @@
 import { v4 as uuidv4 } from "uuid";
-import { sendRabbitMq } from "../../config/rabbitmq";
+import { sendToQueue } from "../../config/rabbitmq";
 import { sendMatchFound } from "./websocketHandler";
 
 interface Match {
@@ -16,18 +16,15 @@ export interface MatchUser {
 
 export interface MatchRequest {
   user: MatchUser;
-  complexities: string[];
-  categories: string[];
-  languages: string[];
+  complexity: string;
+  category: string;
+  language: string;
   timeout: number;
 }
 
 export interface MatchRequestItem {
   id: string;
   user: MatchUser;
-  complexities: string[];
-  categories: string[];
-  languages: string[];
   sentTimestamp: number;
   ttlInSecs: number;
   rejectedPartnerId?: string;
@@ -40,20 +37,17 @@ export const sendMatchRequest = async (
   requestId: string,
   rejectedPartnerId?: string
 ): Promise<boolean> => {
-  const { user, complexities, categories, languages, timeout } = matchRequest;
+  const { user, complexity, category, language, timeout } = matchRequest;
 
   const matchItem: MatchRequestItem = {
     id: requestId,
     user: user,
-    complexities: complexities,
-    categories: categories,
-    languages: languages,
     sentTimestamp: Date.now(),
     ttlInSecs: timeout,
     rejectedPartnerId: rejectedPartnerId,
   };
 
-  const sent = await sendRabbitMq(matchItem);
+  const sent = await sendToQueue(complexity, category, language, matchItem);
   return sent;
 };
 

backend/matching-service/src/utils/constants.ts

@@ -0,0 +1,22 @@
+export enum Complexities {
+  EASY = "Easy",
+  MEDIUM = "Medium",
+  HARD = "Hard",
+}
+
+export enum Categories {
+  STRINGS = "Strings",
+  ALGORITHMS = "Algorithms",
+  DATA_STRUCTURES = "Data Structures",
+  BIT_MANIPULATION = "Bit Manipulation",
+  RECURSION = "Recursion",
+  DYNAMIC_PROGRAMMING = "Dynamic Programming",
+  ARRAYS = "Arrays",
+  TREE = "Tree",
+}
+
+export enum Languages {
+  PYTHON = "Python",
+  JAVA = "Java",
+  C = "C",
+}

backend/matching-service/src/utils/mq_utils.ts

@@ -1,21 +1,23 @@
+import { getPendingRequests } from "../../config/rabbitmq";
 import { createMatch, MatchRequestItem } from "../handlers/matchHandler";
 import { isActiveRequest, isUserConnected } from "../handlers/websocketHandler";
 
-const matchingRequests = new Map<string, MatchRequestItem>();
-
-export const matchUsers = (newRequest: string) => {
+export const matchUsers = (queueName: string, newRequest: string) => {
+  const pendingRequests = getPendingRequests(queueName);
   const newRequestJson = JSON.parse(newRequest) as MatchRequestItem;
   const newRequestUid = newRequestJson.user.id;
-  for (const [uid, pendingRequest] of matchingRequests) {
+
+  for (const [uid, pendingRequest] of pendingRequests) {
     if (
       isExpired(pendingRequest) ||
       !isUserConnected(uid) ||
       !isActiveRequest(uid, pendingRequest.id) ||
       uid === newRequestUid
     ) {
-      matchingRequests.delete(uid);
+      pendingRequests.delete(uid);
       continue;
     }
+
     if (
       isExpired(newRequestJson) ||
       !isUserConnected(newRequestUid) ||
@@ -31,34 +33,13 @@ export const matchUsers = (newRequest: string) => {
       continue;
     }
 
-    if (isMatch(newRequestJson, pendingRequest)) {
-      matchingRequests.delete(uid);
-      createMatch(pendingRequest, newRequestJson);
-      console.log(
-        "Matched users:",
-        JSON.stringify(newRequestJson),
-        JSON.stringify(pendingRequest)
-      );
-      return;
-    }
+    pendingRequests.delete(uid);
+    createMatch(pendingRequest, newRequestJson);
+    return;
   }
-  matchingRequests.set(newRequestUid, newRequestJson);
+  pendingRequests.set(newRequestUid, newRequestJson);
 };
 
 const isExpired = (data: MatchRequestItem): boolean => {
   return Date.now() - data.sentTimestamp >= data.ttlInSecs * 1000;
 };
-
-const isMatch = (req1: MatchRequestItem, req2: MatchRequestItem): boolean => {
-  const hasCommonComplexity = req1.complexities.some((elem) =>
-    req2.complexities.includes(elem)
-  );
-  const hasCommonCategory = req1.categories.some((elem) =>
-    req2.categories.includes(elem)
-  );
-  const hasCommonLanguage = req1.languages.some((elem) =>
-    req2.languages.includes(elem)
-  );
-
-  return hasCommonComplexity && hasCommonCategory && hasCommonLanguage;
-};

frontend/src/contexts/MatchContext.tsx

@@ -25,9 +25,9 @@ type MatchUser = {
 };
 
 type MatchCriteria = {
-  complexities: string[];
-  categories: string[];
-  languages: string[];
+  complexity: string;
+  category: string;
+  language: string;
   timeout: number;
 };
 
@@ -69,9 +69,9 @@ enum MatchPaths {
 
 type MatchContextType = {
   findMatch: (
-    complexities: string[],
-    categories: string[],
-    languages: string[],
+    complexity: string,
+    category: string,
+    language: string,
     timeout: number
   ) => void;
   stopMatch: () => void;
@@ -311,9 +311,9 @@ const MatchProvider: React.FC<{ children?: React.ReactNode }> = (props) => {
   };
 
   const findMatch = (
-    complexities: string[],
-    categories: string[],
-    languages: string[],
+    complexity: string,
+    category: string,
+    language: string,
     timeout: number
   ) => {
     if (!matchUser) {
@@ -332,19 +332,19 @@ const MatchProvider: React.FC<{ children?: React.ReactNode }> = (props) => {
       MatchEvents.MATCH_REQUEST,
       {
         user: matchUser,
-        complexities: complexities,
-        categories: categories,
-        languages: languages,
+        complexity: complexity,
+        category: category,
+        language: language,
         timeout: timeout,
       },
       (requested: boolean) => {
         clearTimeout(requestTimeout);
         setTimeout(() => setLoading(false), 500);
         if (requested) {
           setMatchCriteria({
-            complexities,
-            categories,
-            languages,
+            complexity: complexity,
+            category: category,
+            language: language,
             timeout,
           });
           appNavigate(MatchPaths.MATCHING);
@@ -402,9 +402,9 @@ const MatchProvider: React.FC<{ children?: React.ReactNode }> = (props) => {
 
     const rematchRequest = {
       user: matchUser,
-      complexities: matchCriteria.complexities,
-      categories: matchCriteria.categories,
-      languages: matchCriteria.languages,
+      complexities: matchCriteria.complexity,
+      categories: matchCriteria.category,
+      languages: matchCriteria.language,
       timeout: matchCriteria.timeout,
     };
     matchSocket.emit(
@@ -430,9 +430,9 @@ const MatchProvider: React.FC<{ children?: React.ReactNode }> = (props) => {
     }
 
     findMatch(
-      matchCriteria.complexities,
-      matchCriteria.categories,
-      matchCriteria.languages,
+      matchCriteria.complexity,
+      matchCriteria.category,
+      matchCriteria.language,
       matchCriteria.timeout
     );
   };