Merge pull request #33 from CS3219-AY2425S1/websocket-implementation

Websocket implementation and Matching UI

Author: qiaoen17

Backend/MatchingService/Dockerfile

@@ -0,0 +1,11 @@
+FROM node:20
+
+WORKDIR /app
+
+COPY package*.json ./
+RUN npm install
+
+COPY . .
+
+EXPOSE 3003
+CMD ["npm", "start"]

Backend/MatchingService/app.js

@@ -4,7 +4,6 @@ const dotenv = require("dotenv");
 const matchmakingRouter = require("./controllers/matchmaking");
 const { consumeQueue, consumeDLQ } = require('./rabbitmq/subscriber');
 const { setupRabbitMQ } = require('./rabbitmq/setup');
-const { publishToQueue } = require('./rabbitmq/publisher')
 
 dotenv.config();
 
@@ -23,9 +22,9 @@ setupRabbitMQ().then(() => {
     consumeQueue().catch(console.error);
     consumeDLQ().catch(console.error);
 
-    publishToQueue({userId: "user_1", difficulty: "easy", language: "java"})
-    publishToQueue({userId: "user_2", difficulty: "easy", language: "python"})
-    publishToQueue({userId: "user_3", difficulty: "easy", language: "java"})
+    // publishToQueue({userId: "user_1", difficulty: "easy", language: "java"})
+    // publishToQueue({userId: "user_2", difficulty: "easy", language: "python"})
+    // publishToQueue({userId: "user_3", difficulty: "easy", language: "java"})
 
 })
 

Backend/MatchingService/rabbitmq/publisher.js

@@ -1,11 +1,24 @@
 const amqp = require('amqplib');
-// Connect -> Exchange -> Queue -> Bind -> Publish
 const { matching_exchange_name } = require('./setup.js');
 
+let channel = null;  // Store a persistent channel connection
+
+async function connectToRabbitMQ() {
+    if (!channel) {
+        try {
+            const connection = await amqp.connect(process.env.RABBITMQ_URL);
+            channel = await connection.createChannel();
+            console.log("RabbitMQ channel created");
+        } catch (error) {
+            console.error('Error creating RabbitMQ channel:', error);
+        }
+    }
+    return channel;
+}
+
 async function publishToQueue({userId, difficulty, language}) {
     try {
-        const connection = await amqp.connect(process.env.RABBITMQ_URL);
-        const channel = await connection.createChannel();
+        const channel = await connectToRabbitMQ();  // Reuse persistent connection
         const routingKey = `${difficulty}.${language}`;
 
         // Publish the message to the exchange
@@ -20,18 +33,31 @@ async function publishToQueue({userId, difficulty, language}) {
         } else {
             console.error(`Message NOT sent: ${userId} -> ${routingKey}`);
         }
-        
-        await channel.close();
-        await connection.close();
     } catch (error) {
         console.error('Error publishing to RabbitMQ:', error);
     }
 }
 
-module.exports = { publishToQueue };
-
-
-
+async function publishCancelRequest({ userId }) {
+    try {
+        const channel = await connectToRabbitMQ(); // Reuse persistent connection
+        const routingKey = 'cancel'; // Define a routing key for cancellation
 
+        // Publish the cancel message to the exchange
+        const messageSent = channel.publish(
+            matching_exchange_name,
+            routingKey,
+            Buffer.from(JSON.stringify({ userId }))
+        );
 
+        if (messageSent) {
+            console.log(`Cancel request sent: ${userId}`);
+        } else {
+            console.error(`Cancel request NOT sent: ${userId}`);
+        }
+    } catch (error) {
+        console.error('Error publishing cancel request to RabbitMQ:', error);
+    }
+}
 
+module.exports = { publishToQueue, publishCancelRequest }; 

Backend/MatchingService/rabbitmq/setup.js

@@ -3,6 +3,7 @@ const amqp = require("amqplib");
 const matching_exchange_name = "matching_exchange";
 const dead_letter_exchange_name = "dead_letter_exchange";
 const dead_letter_queue_name = "dead_letter_queue";
+const cancel_queue_name = "cancel_queue";
 const queueNames = [
     'easy.python',
     'easy.java',
@@ -12,7 +13,7 @@ const queueNames = [
     'medium.cplusplus',
     'hard.python',
     'hard.java',
-    'hard.cplusplus'
+    'hard.cplusplus',
 ];
 
 async function setupRabbitMQ() {
@@ -26,17 +27,17 @@ async function setupRabbitMQ() {
         const channel = await connection.createChannel();
 
         // Declare the matching exchange (topic)
-        await channel.assertExchange(matching_exchange_name, "topic", { durable: false });
+        await channel.assertExchange(matching_exchange_name, "topic", { durable: true });
 
         // Declare the dead-letter exchange (fanout)
-        await channel.assertExchange(dead_letter_exchange_name, "fanout", { durable: false });
+        await channel.assertExchange(dead_letter_exchange_name, "fanout", { durable: true });
 
         // Declare and bind all main queues with TTL and DLQ bindings
         for (let queueName of queueNames) {
             await channel.deleteQueue(queueName);  // Ensure we start fresh for each setup
 
             await channel.assertQueue(queueName, {
-                durable: false, 
+                durable: true, 
                 arguments: {
                     'x-message-ttl': 10000,  // 60 seconds TTL
                     'x-dead-letter-exchange': dead_letter_exchange_name  // Bind to dead-letter exchange
@@ -50,9 +51,14 @@ async function setupRabbitMQ() {
         await channel.deleteQueue(dead_letter_queue_name);
 
         // Declare the dead-letter queue and bind it to the dead-letter exchange
-        await channel.assertQueue(dead_letter_queue_name, { durable: false });
+        await channel.assertQueue(dead_letter_queue_name, { durable: true });
         await channel.bindQueue(dead_letter_queue_name, dead_letter_exchange_name, ''); // Bind with no routing key
 
+        // Declare and bind the cancel queue
+        await channel.deleteQueue(cancel_queue_name);  // Delete any existing cancel queue
+        await channel.assertQueue(cancel_queue_name, { durable: true }); // Declare the cancel queue
+        await channel.bindQueue(cancel_queue_name, matching_exchange_name, 'cancel'); // Bind with the "cancel" routing key
+
         console.log("RabbitMQ setup complete with queues, DLQ, and bindings.");
 
         await channel.close();
@@ -62,4 +68,4 @@ async function setupRabbitMQ() {
     }
 }
 
-module.exports = { setupRabbitMQ, matching_exchange_name, queueNames, dead_letter_queue_name };
+module.exports = { setupRabbitMQ, matching_exchange_name, queueNames, dead_letter_queue_name , cancel_queue_name};

Backend/MatchingService/rabbitmq/subscriber.js

@@ -23,20 +23,20 @@ function matchUsers(channel, msg, userId, language, difficulty) {
         waitingUsers[criteriaKey] = [];
     }
 
-    waitingUsers[criteriaKey].push({ userId, msg }); // Store both userId and the message
+    // Store both the userId, message, and the channel in waitingUsers
+    waitingUsers[criteriaKey].push({ userId, msg, channel });
     console.log(`User ${userId} added to ${criteriaKey}. Waiting list: ${waitingUsers[criteriaKey].length}`);
 
     // Check if there are 2 or more users waiting for this criteria
     if (waitingUsers[criteriaKey].length >= 2) {
         const matchedUsers = waitingUsers[criteriaKey].splice(0, 2); // Match the first two users
         console.log(`Matched users: ${matchedUsers.map(user => user.userId)}`);
 
-        // Send match success (this could trigger WebSocket communication)
-        // notifyUsers(matchedUsers.map(user => user.userId));
+        // Notify users of the match
         notifyUsers(matchedUsers.map(user => user.userId), 'Match found!', 'match');
 
         // Acknowledge the messages for both matched users
-        matchedUsers.forEach(({ msg }) => {
+        matchedUsers.forEach(({ msg, channel }) => {
             acknowledgeMessage(channel, msg);
         });
 
@@ -46,6 +46,7 @@ function matchUsers(channel, msg, userId, language, difficulty) {
     return false;
 }
 
+
 async function acknowledgeMessage(channel, msg) {
     return new Promise((resolve, reject) => {
         try {
@@ -64,19 +65,46 @@ async function acknowledgeMessage(channel, msg) {
 async function rejectMessage(channel, msg, userId) {
     return new Promise((resolve, reject) => {
         try {
+            // Get user data from the message to find the correct key in waitingUsers
+            const userData = JSON.parse(msg.content.toString());
+            const { language, difficulty } = userData;
+
+            // Correctly creating the criteriaKey using template literals
+            const criteriaKey = `${difficulty}.${language}`;
+
+            
+            // Find the user in the waitingUsers list and remove them
+            if (waitingUsers[criteriaKey]) {
+                // Find the index of the user in the waiting list
+                const userIndex = waitingUsers[criteriaKey].findIndex(user => user.userId === userId);
+
+                if (userIndex !== -1) {
+                    // Remove the user from the waiting list
+                    waitingUsers[criteriaKey].splice(userIndex, 1);
+                    console.log(`Removed user ${userId} from waiting list for ${criteriaKey}`);
+                }
+            }
+
+            // Reject the message without requeuing
             channel.reject(msg, false); // Reject without requeuing
             console.log(`Rejected message for user: ${userId}`);
+
+            // Clean up the timeoutMap
+            if (timeoutMap[userId]) {
+                clearTimeout(timeoutMap[userId]);
+                delete timeoutMap[userId];
+            }
+
             resolve();
         } catch (error) {
-            console.error(`Failed to reject message for user ${userId}:`, error);
+            console.error(`Failed to reject message for user ${userId}:, error`);
             reject(error);
         }
     });
 }
 
 async function consumeQueue() {
     try {
-        // Connect
         const connection = await amqp.connect(process.env.RABBITMQ_URL);
         const channel = await connection.createChannel();
 
@@ -95,23 +123,23 @@ async function consumeQueue() {
                     // Call matchUsers with channel, message, and user details
                     const matched = matchUsers(channel, msg, userId, language, difficulty);
 
-                    // If not matched, set a timeout for rejection
                     if (!matched) {
                         console.log(`No match for ${userId}, waiting for rejection timeout.`);
 
-                        // Set a timeout for rejection after 10 seconds
                         const timeoutId = setTimeout(async () => {
                             await rejectMessage(channel, msg, userId);
                         }, 10000); // 10 seconds delay
 
-                        // Store the timeout ID
                         timeoutMap[userId] = timeoutId;
                     }
                 }
-            });
+            }, { noAck: false }); // Ensure manual acknowledgment
         }
 
         console.log("Listening to matchmaking queues");
+
+        await consumeCancelQueue();
+        console.log("Listening to Cancel Queue");
     } catch (error) {
         console.error('Error consuming RabbitMQ queue:', error);
     }
@@ -144,4 +172,71 @@ async function consumeDLQ() {
   }
 }
 
+async function consumeCancelQueue() {
+    try {
+        const connection = await amqp.connect(process.env.RABBITMQ_URL);
+        const channel = await connection.createChannel();
+
+        // Subscribe to the cancel queue
+        await channel.consume('cancel_queue', async (msg) => {
+            if (msg !== null) {
+                const { userId } = JSON.parse(msg.content.toString());
+                console.log(`Received cancel request for user: ${userId}`);
+
+                // Process the cancel request
+                await cancelMatching(channel, msg, userId);
+            }
+        }, { noAck: false }); // Ensure manual acknowledgment
+
+        console.log("Listening for cancel requests");
+    } catch (error) {
+        console.error('Error consuming cancel queue:', error);
+    }
+}
+
+async function cancelMatching(cancelChannel, cancelMsg, userId) {
+    try {
+        let foundOriginalMsg = false;
+
+        // Loop through waitingUsers to find the original message for the user
+        Object.keys(waitingUsers).forEach(criteriaKey => {
+            const userIndex = waitingUsers[criteriaKey].findIndex(user => user.userId === userId);
+
+            if (userIndex !== -1) {
+                const { msg, channel } = waitingUsers[criteriaKey][userIndex]; // Get original msg and its channel
+
+                // Acknowledge the original matchmaking message from the queue (e.g., easy.python)
+                if (msg && channel) {
+                    console.log(`Acknowledging original message for user ${userId} in queue ${criteriaKey}`);
+                    channel.ack(msg); // Use the same channel that consumed the message to acknowledge it
+                    foundOriginalMsg = true;
+                }
+
+                // Remove the user from the waiting list
+                waitingUsers[criteriaKey].splice(userIndex, 1);
+                console.log(`User ${userId} removed from waiting list for ${criteriaKey}`);
+            }
+        });
+
+        // If original message not found, log a warning
+        if (!foundOriginalMsg) {
+            console.warn(`Original message for user ${userId} not found in matchmaking queues.`);
+        }
+
+        // Clear any timeouts for the user
+        if (timeoutMap[userId]) {
+            clearTimeout(timeoutMap[userId]);
+            delete timeoutMap[userId];
+        }
+
+        // Acknowledge the cancel message from the cancel queue
+        cancelChannel.ack(cancelMsg);
+        console.log(`Cancel processed for user ${userId}`);
+
+    } catch (error) {
+        console.error(`Failed to process cancel for user ${userId}:`, error);
+    }
+}
+
+
 module.exports = { consumeQueue, consumeDLQ };