Set up queue per match criteria

Author: ruiqi7

backend/matching-service/config/rabbitmq.ts

@@ -1,13 +1,95 @@
 import amqplib, { Connection } from "amqplib";
 import dotenv from "dotenv";
-import { matchUsers } from "../src/utils/mq_utils";
+import { matchUsers, matchUsersInQueue } from "../src/utils/mq_utils";
 import { MatchRequestItem } from "../src/handlers/matchHandler";
 
 dotenv.config();
 
 let mrConnection: Connection;
 const queue = "match_requests";
 
+let mrConnectionNew: Connection;
+const queues: string[] = [];
+
+enum Complexities {
+  EASY = "Easy",
+  MEDIUM = "Medium",
+  HARD = "Hard",
+}
+
+enum Categories {
+  STRINGS = "Strings",
+  ALGORITHMS = "Algorithms",
+  DATA_STRUCTURES = "Data Structures",
+  BIT_MANIPULATION = "Bit Manipulation",
+  RECURSION = "Recursion",
+  DYNAMIC_PROGRAMMING = "Dynamic Programming",
+  ARRAYS = "Arrays",
+  TREE = "Tree",
+}
+
+enum LANGUAGES {
+  PYTHON = "Python",
+  JAVA = "Java",
+  C = "C",
+}
+
+export const pendingRequestsPerQueue = new Map<
+  string,
+  Map<string, MatchRequestItem>
+>();
+
+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}`);
+      }
+    }
+  }
+};
+
+export const connectToRabbitMq = async () => {
+  try {
+    initQueueNames();
+    mrConnectionNew = await amqplib.connect(`${process.env.RABBITMQ_ADDR}`);
+    for (const queue of queues) {
+      await setUpQueue(queue);
+      pendingRequestsPerQueue.set(queue, new Map<string, MatchRequestItem>());
+    }
+  } catch (error) {
+    console.error(error);
+    process.exit(1);
+  }
+};
+
+const setUpQueue = async (queueName: string) => {
+  const consumerChannel = await mrConnectionNew.createChannel();
+  await consumerChannel.assertQueue(queueName);
+
+  consumerChannel.consume(queueName, (msg) => {
+    console.log(`consume from queue: ${queueName}`);
+    if (msg !== null) {
+      matchUsersInQueue(queueName, msg.content.toString());
+      consumerChannel.ack(msg);
+    }
+  });
+};
+
+export const sendToQueue = async (data: MatchRequestItem) => {
+  try {
+    const queueName = `${data.complexities[0]}_${data.categories[0]}_${data.languages[0]}`;
+    const senderChannel = await mrConnectionNew.createChannel();
+    senderChannel.sendToQueue(queueName, Buffer.from(JSON.stringify(data)));
+    return true;
+  } catch (error) {
+    console.log(error);
+    return false;
+  }
+};
+
+// ----------------
+
 export const connectRabbitMq = async () => {
   try {
     mrConnection = await amqplib.connect(`${process.env.RABBITMQ_ADDR}`);

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 { connectRabbitMq, connectToRabbitMq } from "./config/rabbitmq.ts";
 
 const server = http.createServer(app);
 export const io = new Server(server, {
@@ -20,7 +20,8 @@ io.on("connection", (socket) => {
 const PORT = process.env.PORT || 3002;
 
 if (process.env.NODE_ENV !== "test") {
-  connectRabbitMq()
+  connectToRabbitMq()
+    // connectRabbitMq()
     .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 {
@@ -53,7 +53,7 @@ export const sendMatchRequest = async (
     rejectedPartnerId: rejectedPartnerId,
   };
 
-  const sent = await sendRabbitMq(matchItem);
+  const sent = await sendToQueue(matchItem);
   return sent;
 };
 

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

@@ -1,6 +1,47 @@
+import { pendingRequestsPerQueue } from "../../config/rabbitmq";
 import { createMatch, MatchRequestItem } from "../handlers/matchHandler";
 import { isActiveRequest, isUserConnected } from "../handlers/websocketHandler";
 
+export const matchUsersInQueue = (queueName: string, newRequest: string) => {
+  const pendingRequests = pendingRequestsPerQueue.get(queueName)!;
+  const newRequestJson = JSON.parse(newRequest) as MatchRequestItem;
+  const newRequestUid = newRequestJson.user.id;
+
+  for (const [uid, pendingRequest] of pendingRequests) {
+    if (
+      isExpired(pendingRequest) ||
+      !isUserConnected(uid) ||
+      !isActiveRequest(uid, pendingRequest.id) ||
+      uid === newRequestUid
+    ) {
+      pendingRequests.delete(uid);
+      continue;
+    }
+
+    if (
+      isExpired(newRequestJson) ||
+      !isUserConnected(newRequestUid) ||
+      !isActiveRequest(newRequestUid, newRequestJson.id)
+    ) {
+      return;
+    }
+
+    if (
+      uid === newRequestJson.rejectedPartnerId ||
+      newRequestUid === pendingRequest.rejectedPartnerId
+    ) {
+      continue;
+    }
+
+    pendingRequests.delete(uid);
+    createMatch(pendingRequest, newRequestJson);
+    return;
+  }
+  pendingRequests.set(newRequestUid, newRequestJson);
+};
+
+// ----------------
+
 const matchingRequests = new Map<string, MatchRequestItem>();
 
 export const matchUsers = (newRequest: string) => {