Add Match handling

Author: tyouwei

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,11 +1,14 @@
 'use client';
 
 import { useState, useEffect } from 'react';
+import { useRouter } from 'next/router';
 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();
 
   useEffect(() => {
     const timer = setInterval(() => {
@@ -15,6 +18,36 @@ export default function LoadingPage() {
     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]);
+
+  useEffect(() => {
+    // Start consuming messages from the queue when the component mounts
+    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);
+      }
+    };
+
+    startConsumingMessages();
+  }, []);
+
   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
+      }
+    })();
+  });
+};