11const amqp = require ( 'amqplib' ) ;
22const { queueNames } = require ( './setup.js' ) ;
3- const { matchUsers } = require ( '../services/matchingService.js' ) ;
3+ // const { matchUsers } = require('../services/matchingService.js');
4+ const { notifyUsers } = require ( '../websocket/websocket' ) ;
45
56// TODO: Subscribe and acknowledge messages with user info when timeout/user matched
67
78// To remember what goes in a subscriber use some Acronym
89// Connect, Assert, Process, E - for Acknowledge
910
11+ const dead_letter_queue_name = "dead_letter_queue" ;
12+ const timeoutMap = { } ;
13+
14+ // Local dictionary to store waiting users
15+ const waitingUsers = { } ;
16+
17+ // using promises to handle errors and ensure clearing of timer.
18+ function matchUsers ( channel , msg , userId , language , difficulty ) {
19+ const criteriaKey = `${ difficulty } ${ language } ;
20+
21+ // If the criteria key does not exist, create it
22+ if ( ! waitingUsers [ criteriaKey ] ) {
23+ waitingUsers [ criteriaKey ] = [ ] ;
24+ }
25+
26+ waitingUsers [ criteriaKey ] . push ( { userId, msg } ) ; // Store both userId and the message
27+ console . log ( `User ${ userId } ${ criteriaKey } ${ waitingUsers [ criteriaKey ] . length } ) ;
28+
29+ // Check if there are 2 or more users waiting for this criteria
30+ if ( waitingUsers [ criteriaKey ] . length >= 2 ) {
31+ const matchedUsers = waitingUsers [ criteriaKey ] . splice ( 0 , 2 ) ; // Match the first two users
32+ console . log ( `Matched users: ${ matchedUsers . map ( user => user . userId ) } ) ;
33+
34+ // Send match success (this could trigger WebSocket communication)
35+ // notifyUsers(matchedUsers.map(user => user.userId));
36+ notifyUsers ( matchedUsers . map ( user => user . userId ) , 'Match found!' , 'match' ) ;
37+
38+ // Acknowledge the messages for both matched users
39+ matchedUsers . forEach ( ( { msg } ) => {
40+ acknowledgeMessage ( channel , msg ) ;
41+ } ) ;
42+
43+ return true ;
44+ }
45+
46+ return false ;
47+ }
48+
49+ async function acknowledgeMessage ( channel , msg ) {
50+ return new Promise ( ( resolve , reject ) => {
51+ try {
52+ channel . ack ( msg ) ;
53+ console . log ( `Acknowledged message for user: ${ JSON . parse ( msg . content ) . userId } ) ;
54+ clearTimeout ( timeoutMap [ JSON . parse ( msg . content ) . userId ] ) ; // Clear any pending timeout
55+ delete timeoutMap [ JSON . parse ( msg . content ) . userId ] ; // Clean up
56+ resolve ( ) ;
57+ } catch ( error ) {
58+ console . error ( `Failed to acknowledge message:` , error ) ;
59+ reject ( error ) ;
60+ }
61+ } ) ;
62+ }
63+
64+ async function rejectMessage ( channel , msg , userId ) {
65+ return new Promise ( ( resolve , reject ) => {
66+ try {
67+ channel . reject ( msg , false ) ; // Reject without requeuing
68+ console . log ( `Rejected message for user: ${ userId } ) ;
69+ resolve ( ) ;
70+ } catch ( error ) {
71+ console . error ( `Failed to reject message for user ${ userId } , error ) ;
72+ reject ( error ) ;
73+ }
74+ } ) ;
75+ }
76+
1077async function consumeQueue ( ) {
1178 try {
1279 // Connect
1380 const connection = await amqp . connect ( process . env . RABBITMQ_URL ) ;
1481 const channel = await connection . createChannel ( ) ;
1582
16- // Queues already created in setup.js
83+ console . log ( "Waiting for users..." ) ;
1784
18- console . log ( "Waiting for users..." )
19-
20- // Process + subscribe to each queue
85+ // Process + subscribe to each matchmaking queue
2186 for ( let queueName of queueNames ) {
22- await channel . consume ( queueName , ( msg ) => {
87+ await channel . consume ( queueName , async ( msg ) => {
2388 if ( msg !== null ) {
2489 const userData = JSON . parse ( msg . content . toString ( ) ) ;
2590 const { userId, language, difficulty } = userData ;
2691
2792 // Perform the matching logic
2893 console . log ( `Received user ${ userId } ${ language } ${ difficulty } ) ;
29- matchUsers ( userId , language , difficulty ) ;
94+
95+ // Call matchUsers with channel, message, and user details
96+ const matched = matchUsers ( channel , msg , userId , language , difficulty ) ;
97+
98+ // If not matched, set a timeout for rejection
99+ if ( ! matched ) {
100+ console . log ( `No match for ${ userId } ) ;
30101
31- // E- Acknowledge
32- channel . ack ( msg ) ;
102+ // Set a timeout for rejection after 10 seconds
103+ const timeoutId = setTimeout ( async ( ) => {
104+ await rejectMessage ( channel , msg , userId ) ;
105+ } , 10000 ) ; // 10 seconds delay
106+
107+ // Store the timeout ID
108+ timeoutMap [ userId ] = timeoutId ;
109+ }
33110 }
34111 } ) ;
35112 }
113+
114+ console . log ( "Listening to matchmaking queues" ) ;
36115 } catch ( error ) {
37116 console . error ( 'Error consuming RabbitMQ queue:' , error ) ;
38117 }
39118}
40119
41- module . exports = { consumeQueue } ;
120+ async function consumeDLQ ( ) {
121+ try {
122+ const connection = await amqp . connect ( process . env . RABBITMQ_URL ) ;
123+ const channel = await connection . createChannel ( ) ;
124+
125+ // Consume messages from the DLQ
126+ await channel . consume ( dead_letter_queue_name , ( msg ) => {
127+ if ( msg !== null ) {
128+ const messageContent = JSON . parse ( msg . content . toString ( ) ) ;
129+ const { userId, difficulty, language } = messageContent ;
130+
131+ console . log ( `Received message from DLQ for user: ${ userId } ) ;
132+
133+ // Notify the user via WebSocket
134+ notifyUsers ( userId , `Match not found for ${ difficulty } ${ language } , 'rejection' ) ;
135+
136+ // Acknowledge the message (so it's removed from the DLQ)
137+ channel . ack ( msg ) ;
138+ }
139+ } ) ;
140+
141+ console . log ( `Listening to Dead Letter Queue for unmatched users...` ) ;
142+ } catch ( error ) {
143+ console . error ( 'Error consuming from DLQ:' , error ) ;
144+ }
145+ }
146+
147+ module . exports = { consumeQueue, consumeDLQ } ;
0 commit comments