#89 SSE 기능 개선 및 Kafka 활용

Author: SeJonJ

nodejs-frontend/static/js/roomlist/roomlist.js

@@ -65,10 +65,9 @@ const roomList = {
     });
 
     eventSource.addEventListener('roomDeleted', function(event) {
-      const deletedRoom = JSON.parse(event.data);
-      const deletedRoomId = deletedRoom.roomId;
+      const { roomId } = JSON.parse(event.data);
       $('#roomTableBody')
-          .find(`[data-id='${deletedRoomId}'], [data-roomid='${deletedRoomId}']`)
+          .find(`[data-id='${roomId}'], [data-roomid='${roomId}']`)
           .closest('tr')
           .remove();
     });
@@ -78,16 +77,24 @@ const roomList = {
     });
 
     eventSource.addEventListener("changeUserCnt", function (event) {
-      const chatRoom = JSON.parse(event.data);
-      const chatRoomId = chatRoom.roomId;
-      const userCnt = chatRoom.userCount;
-      const {maxUserCnt} = chatRoom;
+      const { roomId, userCount, maxUserCnt } = JSON.parse(event.data);
 
       const $row = $('#roomTableBody')
-          .find(`[data-id='${chatRoomId}'], [data-roomid='${chatRoomId}']`)
+          .find(`[data-id='${roomId}'], [data-roomid='${roomId}']`)
           .closest('tr');
 
-      $row.find('span.room-user-count').text(`${userCnt}/${maxUserCnt}`);
+      $row.find('span.room-user-count').text(`${userCount}/${maxUserCnt}`);
+    });
+
+    eventSource.addEventListener("changeRoomSetting", function (event) {
+      const { roomId, roomName, userCount, maxUserCnt } = JSON.parse(event.data);
+
+      const $row = $('#roomTableBody')
+          .find(`[data-id='${roomId}'], [data-roomid='${roomId}']`)
+          .closest('tr');
+
+      $row.find('span.room-user-count').text(`${userCount}/${maxUserCnt}`);
+      $row.find('.enterRoomBtn, .directEnterBtn').text(roomName);
     });
 
     window.addEventListener("beforeunload", () => {

springboot-backend/src/main/java/webChat/model/kafka/RoomEvent.java

@@ -5,7 +5,7 @@
 @Getter
 public enum RoomEvent {
     ROOM_CREATE,
-//    ROOM_UPDATE,
+    ROOM_UPDATE,
     ROOM_DELETE,
     ROOM_USER_CNT
 }

springboot-backend/src/main/java/webChat/service/chatroom/ChatRoomService.java

@@ -201,6 +201,10 @@ public boolean delChatRoom(String roomId) throws BadRequestException {
     // 채팅방 수정
     public ChatRoom updateRoom(String roomId, String roomName, String roomPwd, int maxUserCnt) throws BadRequestException {
         ChatRoom chatRoom = redisService.getRedisDataByDataType(roomId, DataType.CHATROOM, KurentoRoom.class);
+        // 방 이름 혹은 최대 인원 수 변경 시에 kafka -> sse를 통해 실시간 방 업데이트
+        if (!chatRoom.getRoomName().equals(roomName) || chatRoom.getMaxUserCnt() != maxUserCnt) {
+            chatKafkaProducer.sendChangedRoomSettingEvent(chatRoom);
+        }
         chatRoom.setRoomName(roomName);
         chatRoom.setRoomPwd(roomPwd);
         chatRoom.setMaxUserCnt(maxUserCnt);

springboot-backend/src/main/java/webChat/service/chatroom/SseService.java

@@ -6,6 +6,7 @@
 import webChat.model.room.ChatRoom;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -31,7 +32,7 @@ public SseEmitter createEmitter() {
 
     // 방 생성 시, 연결된 모든 클라이언트에게 이벤트 전송
     public void sendRoomCreatedEvent(ChatRoom room) {
-        List<SseEmitter> deadEmitters = new CopyOnWriteArrayList<>();
+        List<SseEmitter> deadEmitters = new ArrayList<>();
 
         for (SseEmitter emitter : emitters) {
             try {
@@ -48,7 +49,7 @@ public void sendRoomCreatedEvent(ChatRoom room) {
 
     // 방 삭제 시 이벤트 전송
     public void sendRoomDeletedEvent(ChatRoom room) {
-        List<SseEmitter> deadEmitters = new CopyOnWriteArrayList<>();
+        List<SseEmitter> deadEmitters = new ArrayList<>();
 
         for (SseEmitter emitter : emitters) {
             try {
@@ -67,7 +68,7 @@ public void sendRoomDeletedEvent(ChatRoom room) {
     // 핑 관련 이벤트 전송
     @Scheduled(fixedDelay = 10_000L * 7) // 7분
     public void sendPingToClients() {
-        List<SseEmitter> deadEmitters = new CopyOnWriteArrayList<>();
+        List<SseEmitter> deadEmitters = new ArrayList<>();
         for (SseEmitter emitter : emitters) {
             try {
                 emitter.send(SseEmitter.event()
@@ -82,7 +83,7 @@ public void sendPingToClients() {
 
     // 방 인원수 변경 시 이벤트 전송
     public void sendRoomUserCntEvent(ChatRoom room) {
-        List<SseEmitter> deadEmitters = new CopyOnWriteArrayList<>();
+        List<SseEmitter> deadEmitters = new ArrayList<>();
 
         for (SseEmitter emitter : emitters) {
             try {
@@ -93,7 +94,21 @@ public void sendRoomUserCntEvent(ChatRoom room) {
                 deadEmitters.add(emitter);
             }
         }
+        emitters.removeAll(deadEmitters);
+    }
+
+    public void sendChangeRoomSettingEvent(ChatRoom room) {
+        List<SseEmitter> deadEmitters = new ArrayList<>();
 
+        for (SseEmitter emitter : emitters) {
+            try {
+                emitter.send(SseEmitter.event()
+                        .name("changeRoomSetting")
+                        .data(room));
+            } catch (IOException e) {
+                deadEmitters.add(emitter);
+            }
+        }
         emitters.removeAll(deadEmitters);
     }
 }

springboot-backend/src/main/java/webChat/service/kafka/ChatKafkaProducer.java

@@ -28,4 +28,9 @@ public void sendRoomUserCntEvent(ChatRoom chatRoom) {
         KafkaRoomEvent kafkaRoomEvent = KafkaRoomEvent.of(chatRoom.getRoomId(), chatRoom.getInstanceId(), RoomEvent.ROOM_USER_CNT);
         kafkaTemplate.send("room-events", chatRoom.getRoomId(), kafkaRoomEvent);
     }
+
+    public void sendChangedRoomSettingEvent(ChatRoom chatRoom) {
+        KafkaRoomEvent kafkaRoomEvent = KafkaRoomEvent.of(chatRoom.getRoomId(), chatRoom.getInstanceId(), RoomEvent.ROOM_UPDATE);
+        kafkaTemplate.send("room-events", chatRoom.getRoomId(), kafkaRoomEvent);
+    }
 }

springboot-backend/src/main/java/webChat/service/kafka/ChatRoomEventConsumer.java

@@ -48,6 +48,11 @@ public void listen(ConsumerRecord<String, KafkaEvent> record) throws BadRequestE
                 log.info("===========roomUserCnt==========");
                 sseService.sendRoomUserCntEvent(chatRoom);
             }
+            // 방 설정 변경 (방 이름, 최대 인원 수)
+            case ROOM_UPDATE -> {
+                log.info("===========roomSetting==========");
+                sseService.sendChangeRoomSettingEvent(chatRoom);
+            }
         }
     }
 }