Setup Producer and Subscriber logic

Include frontend call to enterMatchmaking
Add required dependencies to package.json

Author: RingoftheKing

Backend/MatchingService/app.js

@@ -1,28 +1,30 @@
 const express = require('express');
 const cors = require("cors");
 const dotenv = require("dotenv");
-//const matchmakingRouter = require("./controllers/matchmaking");
-//const { consumeQueue } = require('./rabbitmq/subscriber');
-//const { setupRabbitMQ } = require('./rabbitmq/setup');
-//const { publishToQueue } = require('./rabbitmq/publisher')
+const matchmakingRouter = require("./controllers/matchmaking");
+const { consumeQueue } = require('./rabbitmq/subscriber');
+const { setupRabbitMQ } = require('./rabbitmq/setup');
+const { publishToQueue } = require('./rabbitmq/publisher')
 
 dotenv.config();
 
 const app = express();
 app.use(cors());
 app.use(express.json());
+app.use(express.urlencoded({ extended: true }));
+app.options("*", cors());
 
 app.use('/api/match', matchmakingRouter);
 
 // TODO: Start consuming RabbitMQ queues 
-/*
+
 setupRabbitMQ().then(() => {
 
     consumeQueue().catch(console.error);
-    publishToQueue("user_234", "easy", "python")
-    publishToQueue("user_100", "easy", "java")
+    publishToQueue({userId: "user_234", difficulty: "easy", language: "python"})
+    publishToQueue({userId: "user_234", difficulty: "easy", language: "java"})
 
 })
-*/
+
 
 module.exports = app;

Backend/MatchingService/controllers/matchmaking.js

@@ -1,23 +1,37 @@
-// TODO: WRITE API FOR MATCHING USER, REMEMBER TO DEAL WITH CORS ALLOW ACCESS ORIGIN ERROR
+// WRITE API FOR MATCHING USER, REMEMBER TO DEAL WITH CORS ALLOW ACCESS ORIGIN ERROR
+// Cors settled in app.js
+
 
-/*
 const express = require('express');
 const router = express.Router();
 const { publishToQueue } = require('../rabbitmq/publisher');
 
 // Route for frontend to send user matching info
-router.post('/match', async (req, res) => {
-    const { userId, language, difficulty } = req.body;
+router.post('/enterMatchmaking', async (req, res) => {
+    const { userId, difficulty, language } = req.body;
 
     try {
         // Publish user info to RabbitMQ
-        await publishToQueue(userId, language, difficulty);
+        await publishToQueue({userId: userId, difficulty: difficulty, language: language});
         res.status(200).send('User info sent for matching.');
     } catch (error) {
         console.error('Error publishing user info:', error);
         res.status(500).send('Error in matchmaking process.');
     }
 });
 
-module.exports = router;
-*/
+// This is for the alternative where the player also listens to a queue after entering matchmaking
+/*
+router.post('/waitMatch', async (req, res) => {
+    try {
+        // Start consuming RabbitMQ queues
+        // await consumeQueue();
+        res.status(200).send('Waiting for match...');
+    } catch (error) {
+        console.error('Error consuming RabbitMQ queue:', error);
+        res.status(500).send('Error in matchmaking process.');
+    }
+})
+ */
+
+module.exports = router;

Backend/MatchingService/package-lock.json

@@ -3,6 +3,11 @@
     "amqplib": "^0.10.4",
     "cors": "^2.8.5",
     "dotenv": "^16.4.5",
-    "express": "^4.21.1"
+    "express": "^4.21.1",
+    "ws": "^8.18.0"
+  },
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1",
+    "start": "node index.js"
   }
 }

Backend/MatchingService/package.json

@@ -1,24 +1,29 @@
 const amqp = require('amqplib');
-// TODO: Write function to publish to rabbitMQ 
+// Connect -> Exchange -> Queue -> Bind -> Publish
+const { matching_exchange_name } = require('./setup.js');
 
-/*
-async function publishToQueue(userId, difficulty, language) {
+async function publishToQueue({userId, difficulty, language}) {
     try {
         const connection = await amqp.connect(process.env.RABBITMQ_URL);
         const channel = await connection.createChannel();
-        const matching_exchange_name = 'matching_exchange';
         const routingKey = `${difficulty}.${language}`;
-        const queueName = `${difficulty}.${language}`;
 
-        if (queueInfo) {
-            channel.publish(matching_exchange_name, routingKey, Buffer.from(JSON.stringify({ userId, language, difficulty })));
+        // Connect to the exchange (just in case it does not exist)
+        await channel.assertExchange(matching_exchange_name, 'topic', { durable: false });
 
-            console.log(`Published user: ${userId} with routing key: ${routingKey}`);
+        // Publish the message to the exchange
+        const messageSent = channel.publish(
+            matching_exchange_name,
+            routingKey,
+            Buffer.from(JSON.stringify({ userId, difficulty, language }))
+        );
+
+        if (messageSent) {
+            console.log(`Message sent: ${userId} -> ${routingKey}`);
         } else {
-            console.log(`Cannot publish message: Queue ${queueName} does not exist`);
+            console.error(`Message NOT sent: ${userId} -> ${routingKey}`);
         }
 
-        
         await channel.close();
         await connection.close();
     } catch (error) {
@@ -27,7 +32,6 @@ async function publishToQueue(userId, difficulty, language) {
 }
 
 module.exports = { publishToQueue };
-*/
 
 
 

Backend/MatchingService/rabbitmq/publisher.js

@@ -1,12 +1,20 @@
 const amqp = require("amqplib");
+const matching_exchange_name = "matching_exchange";
+const queueNames = [
+    'easy.python',
+    'easy.java',
+    'easy.cplusplus',
+    'medium.python',
+    'medium.java',
+    'medium.cplusplus',
+    'hard.python',
+    'hard.java',
+    'hard.cplusplus'
+]
 
 async function setupRabbitMQ() {
     try {
         const connection = await amqp.connect(process.env.RABBITMQ_URL)
-        .catch((error) => {
-            console.error("Error connecting to RabbitMQ:", error);
-            return null;
-        });
 
         if (!connection) {
             return;
@@ -15,24 +23,14 @@ async function setupRabbitMQ() {
         const channel = await connection.createChannel();
 
         // Declare matching exchange to be bind to queues 
-        const matching_exchange_name = "matching_exchange";
+
         await channel.assertExchange(matching_exchange_name, "topic", { durable: false });
 
         // Declare dead letter exchange
         const dead_letter_exchange_name = "dead_letter_exchange";
         await channel.assertExchange(dead_letter_exchange_name, "fanout", { durable: false });
 
-        const queueNames = [
-            'easy.python',
-            'easy.java',
-            'easy.cplusplus',
-            'medium.python',
-            'medium.java',
-            'medium.cplusplus',
-            'hard.python',
-            'hard.java',
-            'hard.cplusplus'
-        ]
+
 
         // Create and bind queues to exchange with the routing keys 
         for (let name of queueNames) {
@@ -43,9 +41,13 @@ async function setupRabbitMQ() {
                 console.log(`Queue ${name} does not exist or could not be deleted: ${err.message}`);
             }
             */
+            // this is required to add TTL and dead letter exchange to the queue
+            await channel.deleteQueue(name);
+
             await channel.assertQueue(name, 
                 { durable: false, // durable=false ensures queue will survive broker restarts 
                   arguments: {
+                    'x-message-ttl': 60000, // set message time to live to 60 seconds
                     'x-dead-letter-exchange': dead_letter_exchange_name // set dead letter exchange
                   }
 
@@ -73,6 +75,6 @@ async function setupRabbitMQ() {
     }
 }
 
-module.exports = { setupRabbitMQ };
+module.exports = { setupRabbitMQ, matching_exchange_name, queueNames };
 
 setupRabbitMQ()

Backend/MatchingService/rabbitmq/setup.js

@@ -1,31 +1,34 @@
 const amqp = require('amqplib');
+const { queueNames } = require('./setup.js');
+const { matchUsers } = require('../services/matchingService.js');
 
 // TODO: Subscribe and acknowledge messages with user info when timeout/user matched
 
-//const { matchUsers } = require('../services/matchingService');
+// To remember what goes in a subscriber use some Acronym
+// Connect, Assert, Process, E - for Acknowledge
 
-/*
 async function consumeQueue() {
     try {
+        // Connect
         const connection = await amqp.connect(process.env.RABBITMQ_URL);
         const channel = await connection.createChannel();
-        const exchange = 'matching_exchange';
 
-        // Consuming messages from multiple queues (already created in setup)
-        const queueNames = ['easy.python', 'easy.java', 'medium.python', 'medium.java', 'hard.python', 'hard.java'];
+        // Queues already created in setup.js
 
         console.log("Waiting for users...")
 
+        // Process + subscribe to each queue
         for (let queueName of queueNames) {
-            channel.consume(queueName, (msg) => {
+            await channel.consume(queueName, (msg) => {
                 if (msg !== null) {
                     const userData = JSON.parse(msg.content.toString());
-                    // const { userId, language, difficulty } = userData;
+                    const { userId, language, difficulty } = userData;
 
                     // Perform the matching logic
-                    // matchUsers(userId, language, difficulty);
-                    console.log(userData);
+                    console.log(`Received user ${userId} with ${language} and ${difficulty}`);
+                    matchUsers(userId, language, difficulty);
 
+                    // E- Acknowledge
                     channel.ack(msg);
                 }
             });
@@ -35,4 +38,4 @@ async function consumeQueue() {
     }
 }
 
-*/
+module.exports = { consumeQueue };

Backend/MatchingService/rabbitmq/subscriber.js

@@ -1,13 +1,14 @@
 // TODO: Matching users logic
 
 
-/*
-const { notifyUsers } = require('./websocket/websocket) 
+const { notifyUsers } = require('../websocket/websocket')
+// local dict to store waiting users
 const waitingUsers = {};
 
 function matchUsers(userId, language, difficulty) {
     const criteriaKey = `${difficulty}.${language}`;
 
+    // if the criteria key does not exist, create it
     if (!waitingUsers[criteriaKey]) {
         waitingUsers[criteriaKey] = [];
     }
@@ -29,4 +30,3 @@ function matchUsers(userId, language, difficulty) {
 }
 
 module.exports = { matchUsers };
-*/

Backend/MatchingService/services/matchingService.js

@@ -0,0 +1,7 @@
+import axios from 'axios';
+
+const baseUrl = 'http://localhost:3003';
+
+const enterMatchmaking = async (user) => {
+    return await axios.post(`${baseUrl}/api/match/enterQueue`, user);
+}