Merge pull request #121 from peiran18/collab-service

Collab service

Author: peiran18

backend/collab-service/src/index.ts

@@ -4,11 +4,11 @@ import express from 'express';
 import cors from 'cors';
 import { createServer } from 'http';
 import { Server } from 'socket.io';
-import { connectProducer, sendMessage } from './kafka/producer';
+import { connectProducer, sendCodeMessage, sendChatMessage } from './kafka/producer';
 import { connectRoomConsumer } from './kafka/roomConsumer';
 import { connectCodeConsumer } from './kafka/codeConsumer';
-import { connectChatConsumer } from './kafka/chatConsumer'; // Import the new chat consumer
-import { roomManager } from './models/room';
+import { connectChatConsumer } from './kafka/chatConsumer';
+import { roomManager } from './services/roomManager';
 import redis from './redisClient';
 
 const app = express();
@@ -49,7 +49,7 @@ io.on('connection', (socket) => {
       return;
     }
 
-    await sendMessage('collab-code', { key: username, value: code });
+    await sendCodeMessage('collab-code', { key: username, value: code });
   });
 
   socket.on('get-room-details', async (roomId: string, callback) => {
@@ -82,6 +82,10 @@ io.on('connection', (socket) => {
     redis.hdel(socketsToUsersKey, socket.id);
   });
 
+  socket.on('disconnect-collab', async () => {
+    handleDisconnect(socket.id);
+  });
+
   // New event handler for chat messages
   socket.on('chat-message', async (message: string) => {
     const username = await redis.hget(socketsToUsersKey, socket.id);
@@ -91,7 +95,7 @@ io.on('connection', (socket) => {
     }
 
     // Send the chat message to Kafka
-    await sendMessage('collab-chat', { key: username, value: message });
+    await sendChatMessage('collab-chat', { key: username, value: message });
   });
 });
 

backend/collab-service/src/kafka/chatConsumer.ts

@@ -1,5 +1,5 @@
 import { Kafka, Consumer, EachMessagePayload } from 'kafkajs';
-import { roomManager } from '../models/room';
+import { roomManager } from '../services/roomManager';
 import redis from '../redisClient';
 
 const kafka = new Kafka({

backend/collab-service/src/kafka/producer.ts

@@ -8,18 +8,26 @@ const kafka = new Kafka({
 const producer: Producer = kafka.producer();
 
 export async function connectProducer(): Promise<void> {
-    await producer.connect();    
+    await producer.connect();
     console.log('Producer connected');
 }
 
-export async function sendMessage(topic: string, message: { key: string; value: any }): Promise<void> {
+export async function sendCodeMessage(topic: string, message: { key: string; value: any }): Promise<void> {
     const result = await producer.send({
         topic,
         messages: [{ key: message.key, value: JSON.stringify(message.value) }],
     });
     console.log(`Sent message: ${JSON.stringify(result)}`);
 }
 
+export async function sendChatMessage(topic: string, message: { key: string; value: string }): Promise<void> {
+    const result = await producer.send({
+        topic,
+        messages: [{ key: message.key, value: message.value }],
+    });
+    console.log(`Sent chat message: ${JSON.stringify(result)}`);
+}
+
 export async function disconnectProducer(): Promise<void> {
     await producer.disconnect();
     console.log('Producer disconnected');

docker-compose.yml

@@ -40,7 +40,7 @@ services:
       - "8888:8888"
     environment:
       KAFKA_BROKER: kafka:9092
-      KAFKA_TOPIC: "collab-code,collab-room"
+      KAFKA_TOPIC: "collab-code,collab-room,collab-chat"
     depends_on:
       - kafka
     networks:

frontend/src/pages/CollaborationPage.js

@@ -18,6 +18,7 @@ function CollaborationPage() {
   const isRemoteChange = useRef(false); // Flag to prevent infinite loop, true if change to editor is from remote (other user)
   const timeoutRef = useRef(null); // Timeout reference for the read-only state of the editor
   const countdownRef = useRef(null); // Timeout reference for the countdown when user gets kicked out
+  const messagesEndRef = useRef(null); // Reference to the end of the chat messages
 
   const [code, setCode] = useState('');
   const [question, setQuestion] = useState('');
@@ -29,6 +30,7 @@ function CollaborationPage() {
   const [userId, setUserId] = useState('');
   const [chatMessages, setChatMessages] = useState([]);
   const [newMessage, setNewMessage] = useState('');
+  const [currentUsername, setCurrentUsername] = useState('');
 
   function formatQuestion(question) {
     return `
@@ -50,6 +52,7 @@ ${question.questionDescription}
         const data = await verifyToken(token);
         const username = data.data.username;
         setUserId(data.data.id);
+        setCurrentUsername(username);
 
         console.log("Username:", username);
 
@@ -110,11 +113,24 @@ ${question.questionDescription}
     });
 
     // Listen for chat messages
-    collabService.onChatMessage((data) => {
+    const handleChatMessage = (data) => {
       setChatMessages((prevMessages) => [...prevMessages, { sender: data.sender, message: data.message }]);
-    });
+    };
+
+    collabService.onChatMessage(handleChatMessage);
+
+    return () => {
+      collabService.offChatMessage(handleChatMessage);
+    };
   }, []);
 
+  // Auto-scroll to the bottom when new messages are added
+  useEffect(() => {
+    if (messagesEndRef.current) {
+      messagesEndRef.current.scrollIntoView({ behavior: 'smooth' });
+    }
+  }, [chatMessages]);
+
   // Kick user out if other user disconnects
   useEffect(() => {
     collabService.onOtherUserDisconnect(() => {
@@ -215,19 +231,27 @@ ${question.questionDescription}
 
       {/* Chatbox at the bottom */}
       <div className="h-1/4 w-full border-t border-gray-700 flex flex-col bg-gray-800">
-        <div className="flex-grow overflow-y-auto p-3">
+        <div className="flex-grow overflow-y-auto overflow-x-hidden p-3">
           {chatMessages.map((msg, index) => (
-            <div key={index} className={`mb-2 ${msg.sender === 'You' ? 'text-right' : 'text-left'}`}>
-              <span className="text-sm text-white font-semibold">{msg.sender}:</span>
-              <p className="text-base text-white">{msg.message}</p>
+            <div key={index} className={`flex flex-col mb-2 ${msg.sender === 'You' ? 'items-end' : 'items-start'}`}>
+              <div className={`text-sm font-semibold text-white ${msg.sender === 'You' ? 'text-right' : 'text-left'}`}>
+                {msg.sender}
+              </div>
+              <div
+                className={`max-w-md p-2 rounded-lg break-words ${msg.sender === 'You' ? 'bg-blue-600 text-white text-right' : 'bg-gray-700 text-white'
+                  }`}
+              >
+                <p>{msg.message}</p>
+              </div>
             </div>
           ))}
+          <div ref={messagesEndRef} />
         </div>
         <div className="p-3 border-t border-gray-700">
           <div className="flex">
             <input
               type="text"
-              className="flex-grow bg-gray-900 text-white p-2 rounded-l-md focus:outline-none"
+              className="input input-bordered w-full bg-white text-black rounded-l-full rounded-r-full"
               placeholder="Type a message..."
               value={newMessage}
               onChange={(e) => setNewMessage(e.target.value)}
@@ -238,7 +262,7 @@ ${question.questionDescription}
               }}
             />
             <button
-              className="bg-blue-600 text-white p-2 rounded-r-md"
+              className="btn btn-primary ml-2 rounded-l-full rounded-r-full"
               onClick={handleSendMessage}
             >
               Send

frontend/src/services/collabService.js

@@ -2,6 +2,10 @@ import io from 'socket.io-client';
 
 const socket = io('http://localhost:8888');
 
+const chatMessageCallbacks = new Set();
+
+let chatListenerAdded = false; // Flag to prevent duplicate listeners
+
 const collabService = {
   register: (username) => {
     return new Promise((resolve, reject) => {
@@ -62,16 +66,22 @@ const collabService = {
     });
   },
 
-  // New method to send chat messages
-  sendChatMessage: (message) => {
-    socket.emit('chat-message', message);
+  onChatMessage: (callback) => {
+    if (!chatListenerAdded) {
+      socket.on('chat-message', (data) => {
+        chatMessageCallbacks.forEach((cb) => cb(data));
+      });
+      chatListenerAdded = true;
+    }
+    chatMessageCallbacks.add(callback); // Add to the Set
   },
 
-  // New method to receive chat messages
-  onChatMessage: (callback) => {
-    socket.on('chat-message', (data) => {
-      callback(data);
-    });
+  offChatMessage: (callback) => {
+    chatMessageCallbacks.delete(callback); // Remove specific callback
+  },
+
+  sendChatMessage: (message) => {
+    socket.emit('chat-message', message);
   },
 };