Merge pull request #75 from CS3219-AY2425S1/feat/fe_to_rabbitmq

Feat/fe to rabbitmq

Author: simbayippy

matching-service/src/services/matchingService.ts

@@ -56,13 +56,11 @@ export const processOldUsers = async (): Promise<void> => {
         // Send to user subscribed queues if there is a match
         await sendToQueue(match._id, {
           status: "matched",
-          user1: match,
-          user2: user,
+          match: user,
         });
         await sendToQueue(user._id, {
           status: "matched",
-          user1: match,
-          user2: user,
+          match: match,
         });
         continue;
       }
@@ -101,13 +99,11 @@ export const processNewUser = async (user: User): Promise<void> => {
   if (match) {
     await sendToQueue(match._id, {
       status: "matched",
-      user1: match,
-      user2: user,
+      match: user,
     });
     await sendToQueue(user._id, {
       status: "matched",
-      user1: match,
-      user2: user,
+      match: match,
     });
   } else {
     // Add to the topic queue if no match

matching-service/src/services/rabbitMqService.ts

@@ -105,7 +105,6 @@ export const sendToQueue = async (
 
     await channel.assertQueue(queue, {
       durable: true,
-      expires: 300000, //expire after 5 minutes of idle
     });
 
     await channel.sendToQueue(queue, Buffer.from(JSON.stringify(payload)));

peerprep-fe/src/app/(main)/match/page.tsx

@@ -1,20 +1,49 @@
 'use client';
 
 import { useState, useEffect } from 'react';
+import { useRouter } from 'next/navigation';
 import { User, Code } from 'lucide-react';
+import { consumeMessageFromQueue } from '@/lib/rabbitmq';
 
 export default function LoadingPage() {
   const [elapsedTime, setElapsedTime] = useState(0);
   const [usersWaiting, setUsersWaiting] = useState(4);
+  const router = useRouter();
+
+  const startConsumingMessages = async () => {
+    try {
+      await consumeMessageFromQueue().then((message) => {
+        // This function is called when a message is consumed
+        if (message.status == 'matched') {
+          console.log('Match found, your partner is');
+          router.push('/');
+        } else {
+          console.log('Match failed');
+          router.push('/');
+        }
+      });
+    } catch (error) {
+      console.error('Error consuming message:', error);
+    }
+  };
 
   useEffect(() => {
+    startConsumingMessages();
     const timer = setInterval(() => {
       setElapsedTime((prevTime) => prevTime + 1);
     }, 1000);
     setUsersWaiting(5);
     return () => clearInterval(timer);
   }, []);
 
+  useEffect(() => {
+    if (elapsedTime >= 60) {
+      // Execute your action here
+      console.log('Elapsed time reached 60 seconds. Going back to main page');
+      router.push('/');
+    }
+  }, [elapsedTime]);
+
   return (
     <div className="flex min-h-screen flex-col bg-[#1a1f2e] text-gray-300">
       <header className="flex items-center justify-between border-b border-gray-700 p-4">

peerprep-fe/src/lib/rabbitmq.ts

@@ -34,3 +34,39 @@ export const sendMessageToQueue = async (message: Record<string, any>) => {
     throw err;
   }
 };
+
+export const consumeMessageFromQueue = async () => {
+  return new Promise<any>((resolve, reject) => {
+    (async () => {
+      try {
+        // Connect to RabbitMQ server
+        const connection = await connect(process.env.RABBITMQ_URL);
+        const channel = await connection.createChannel();
+        const queue = process.env.MATCHING_SERVICE_QUEUE;
+
+        // Ensure the queue exists
+        await channel.assertQueue(queue, { durable: true });
+
+        // Consume messages from the queue
+        console.log(`Waiting for messages in ${queue}...`);
+        channel.consume(
+          queue,
+          (msg: any) => {
+            if (msg !== null) {
+              const messageContent = JSON.parse(msg.content.toString());
+              console.log(`Received:`, messageContent);
+              channel.ack(msg);
+              resolve(messageContent); // Resolve the Promise with the message content
+            }
+          },
+          {
+            noAck: false,
+          },
+        );
+      } catch (error) {
+        console.error('Error in consuming messages:', error);
+        reject(error); // Reject the Promise on error
+      }
+    })();
+  });
+};