diff --git a/momogo-core/src/main/java/com/momogo/core/common/exception/RealtimeErrorCode.java b/momogo-core/src/main/java/com/momogo/core/common/exception/RealtimeErrorCode.java new file mode 100644 index 0000000..698581d --- /dev/null +++ b/momogo-core/src/main/java/com/momogo/core/common/exception/RealtimeErrorCode.java @@ -0,0 +1,28 @@ +package com.momogo.core.common.exception; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; +import org.springframework.http.HttpStatus; + +@Getter +@RequiredArgsConstructor +public enum RealtimeErrorCode implements ErrorCode { + + REDIS_PUBLISH_FAILED(8001, "REDIS_PUBLISH_FAILED", HttpStatus.SERVICE_UNAVAILABLE, "실시간 메시지 발행 중 오류가 발생했습니다."), + JSON_SERIALIZATION_FAILED(8002, "JSON_SERIALIZATION_FAILED", HttpStatus.INTERNAL_SERVER_ERROR, "실시간 메시지 직렬화 중 오류가 발생했습니다."); + + private final int numeric; + private final String errorKey; + private final HttpStatus httpStatus; + private final String message; + + @Override + public String getDomain() { + return "REALTIME"; + } + + @Override + public String getCode() { + return getDomain() + "-" + getErrorKey(); + } +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/config/RedisPubSubConfig.java b/momogo-realtime/src/main/java/com/momogo/realtime/config/RedisPubSubConfig.java new file mode 100644 index 0000000..37a6212 --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/config/RedisPubSubConfig.java @@ -0,0 +1,84 @@ +package com.momogo.realtime.config; + +import com.momogo.realtime.websocket.redis.RedisMessageSubscriber; +import java.util.concurrent.Executor; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.data.redis.listener.RedisMessageListenerContainer; +import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +@Slf4j +@Configuration +@RequiredArgsConstructor +public class RedisPubSubConfig { + + // 실시간 시험방 통신에 사용할 Redis 채널 토픽 정의 + @Bean + public ChannelTopic roomTopic() { + return new ChannelTopic("room-realtime-channel"); + } + + // Redis Message를 수신할 어댑터 설정 (subscriber의 handleMessage 메소드를 호출) + @Bean + public MessageListenerAdapter listenerAdapter(RedisMessageSubscriber subscriber) { + return new MessageListenerAdapter(subscriber, "handleMessage"); + } + + /** + * 1. 개별 Pub/Sub 메시지 비동기 처리 전용 스레드 풀 (Spring Bean으로 등록하여 Graceful Shutdown 보장) + */ + @Bean(name = "redisTaskExecutor") + public ThreadPoolTaskExecutor redisTaskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(10); + executor.setMaxPoolSize(50); + executor.setQueueCapacity(500); + executor.setThreadNamePrefix("redis-task-"); + executor.initialize(); + return executor; + } + + /** + * 2. Redis SUBSCRIBE 커넥션 롱폴링/유지 전용 독립 스레드 풀 (Spring Bean 등록) + */ + @Bean(name = "redisSubscriptionExecutor") + public ThreadPoolTaskExecutor redisSubscriptionExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(2); + executor.setMaxPoolSize(5); + executor.setQueueCapacity(10); + executor.setThreadNamePrefix("redis-sub-"); + executor.initialize(); + return executor; + } + + @Bean + public RedisMessageListenerContainer redisMessageListenerContainer( + RedisConnectionFactory connectionFactory, + MessageListenerAdapter listenerAdapter, + ChannelTopic roomTopic, + @Qualifier("redisTaskExecutor") Executor redisTaskExecutor, + @Qualifier("redisSubscriptionExecutor") Executor redisSubscriptionExecutor + ) { + RedisMessageListenerContainer container = new RedisMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + container.addMessageListener(listenerAdapter, roomTopic); + + // Spring Bean으로 생명주기가 안전하게 관리되는 두 스레드 풀 주입 + container.setTaskExecutor(redisTaskExecutor); + container.setSubscriptionExecutor(redisSubscriptionExecutor); + + // 비동기 메시지 처리 중 예외 발생 시 에러 로깅 ErrorHandler + container.setErrorHandler(e -> + log.error("[Redis Pub/Sub Error] 비동기 메시지 수신/처리 중 오류 발생", e) + ); + + return container; + } +} \ No newline at end of file diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/constant/WebSocketConstants.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/constant/WebSocketConstants.java new file mode 100644 index 0000000..fa22331 --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/constant/WebSocketConstants.java @@ -0,0 +1,10 @@ +package com.momogo.realtime.websocket.constant; + +public final class WebSocketConstants { + + public static final String SUB_ROOM_PREFIX = "/sub/rooms/"; + + private WebSocketConstants() { + + } +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/controller/RoomRealtimeController.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/controller/RoomRealtimeController.java new file mode 100644 index 0000000..aa2e98b --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/controller/RoomRealtimeController.java @@ -0,0 +1,38 @@ +package com.momogo.realtime.websocket.controller; + +import com.momogo.realtime.websocket.dto.request.RealtimeMessageRequest; +import com.momogo.realtime.websocket.dto.response.RealtimeMessageResponse; +import com.momogo.realtime.websocket.redis.RedisMessagePublisher; +import jakarta.validation.Valid; +import java.security.Principal; +import java.util.UUID; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.messaging.handler.annotation.DestinationVariable; +import org.springframework.messaging.handler.annotation.MessageMapping; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.stereotype.Controller; + +@Slf4j +@Controller +@RequiredArgsConstructor +public class RoomRealtimeController { + + private final RedisMessagePublisher redisMessagePublisher; + + @MessageMapping("/room/{roomId}/send") + public void sendMessage( + @DestinationVariable("roomId")UUID roomId, + Principal principal, + @Payload @Valid RealtimeMessageRequest request + ) { + + UUID authenticatedUserId = UUID.fromString(principal.getName()); + + log.info("[RoomRealtimeController] 시험방({}) 상태 변경 수신 - authenticatedUserId: {}, status: {}", + roomId, authenticatedUserId, request.status()); + + RealtimeMessageResponse response = RealtimeMessageResponse.of(roomId, authenticatedUserId, request); + redisMessagePublisher.publish(response); + } +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/request/RealtimeMessageRequest.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/request/RealtimeMessageRequest.java new file mode 100644 index 0000000..388111f --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/request/RealtimeMessageRequest.java @@ -0,0 +1,21 @@ +package com.momogo.realtime.websocket.dto.request; + +import com.momogo.realtime.websocket.dto.type.RoomRealtimeStatus; +import jakarta.validation.constraints.Min; +import jakarta.validation.constraints.NotNull; + +/** + * 웹소켓 실시간 응시 상태 변경 요청 DTO + * @param status 응시 상태 + * @param solvedCount 푼 문제 개수 + * @param details 기타 실시간 상세 정보 + */ +public record RealtimeMessageRequest( + @NotNull(message = "응시 상태(status)는 필수 입력값입니다.") + RoomRealtimeStatus status, + @Min(value = 0, message = "푼 문제 개수(solvedCount)는 0 이상이어야 합니다.") + int solvedCount, + Object details +) { + +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/response/RealtimeMessageResponse.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/response/RealtimeMessageResponse.java new file mode 100644 index 0000000..5499031 --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/response/RealtimeMessageResponse.java @@ -0,0 +1,42 @@ +package com.momogo.realtime.websocket.dto.response; + +import com.momogo.realtime.websocket.dto.request.RealtimeMessageRequest; +import com.momogo.realtime.websocket.dto.type.RoomRealtimeStatus; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.util.UUID; + +/** + * 웹소켓 실시간 응시 상태 브로드캐스팅 응답 DTO + * @param roomId 시험방 ID + * @param userId 수험생 ID + * @param status 응시 상태 + * @param solvedCount 푼 문제 개수 + * @param details 기타 실시간 상세 정보 + * @param timestamp 응답 시간 + */ +public record RealtimeMessageResponse( + UUID roomId, + UUID userId, + RoomRealtimeStatus status, + int solvedCount, + Object details, + OffsetDateTime timestamp +) { + + /** + * Request DTO로부터 Response DTO를 정적 팩토리 메서드로 생성 + * @param request + * @return + */ + public static RealtimeMessageResponse of(UUID roomId, UUID userId, RealtimeMessageRequest request) { + return new RealtimeMessageResponse( + roomId, + userId, + request.status(), + request.solvedCount(), + request.details(), + OffsetDateTime.now() + ); + } +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/type/RoomRealtimeStatus.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/type/RoomRealtimeStatus.java new file mode 100644 index 0000000..0c14584 --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/dto/type/RoomRealtimeStatus.java @@ -0,0 +1,19 @@ +package com.momogo.realtime.websocket.dto.type; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * 실시간 시험방 응시 상태 구분용 ENUM + */ +@Getter +@RequiredArgsConstructor +public enum RoomRealtimeStatus { + + ENTER("시험방 입장"), + PROGRESS("문제 풀이 진행 중"), + SUBMIT("답안 제출 완료"), + LEAVE("시험방 퇴장"); + + private final String description; +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/redis/RedisMessagePublisher.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/redis/RedisMessagePublisher.java new file mode 100644 index 0000000..e584ce1 --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/redis/RedisMessagePublisher.java @@ -0,0 +1,54 @@ +package com.momogo.realtime.websocket.redis; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.momogo.core.common.exception.BusinessException; +import com.momogo.core.common.exception.RealtimeErrorCode; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.listener.ChannelTopic; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class RedisMessagePublisher { + + private final StringRedisTemplate redisTemplate; + private final ChannelTopic roomTopic; + private final ObjectMapper objectMapper; + + /** + * 메시지를 JSON으로 직렬화하여 Redis 채널로 Publish 합니다. + * @param messagePayload + */ + public void publish(Object messagePayload) { + + String jsonMessage; + + try { + jsonMessage = objectMapper.writeValueAsString(messagePayload); + } catch (JsonProcessingException e) { + log.error("[Redis Publisher] JSON 직렬화 실패", e); + throw new BusinessException(RealtimeErrorCode.JSON_SERIALIZATION_FAILED); + } + + // 2. Redis 메시지 발행 수행 (실패 시 503 REDIS_PUBLISH_FAILED) + try { + log.debug("[Redis Publisher] Redis 채널({})로 메시지 발행: {}", roomTopic.getTopic(), jsonMessage); + + Long receiversCount = redisTemplate.convertAndSend(roomTopic.getTopic(), jsonMessage); + + if (receiversCount == null || receiversCount == 0) { + log.warn("[Redis Publisher] 수신자 0명 - 채널({})로 발행되었으나 메시지를 수신한 구독자/인스턴스가 없습니다.", roomTopic.getTopic()); + } else { + log.info("[Redis Publisher] Redis 채널({}) 메시지 발행 성공 (수신 인스턴스: {}개)", roomTopic.getTopic(), receiversCount); + } + } catch (Exception e) { + log.error("[Redis Publisher] Redis 메시지 전송 실패 - topic: {}", roomTopic.getTopic(), e); + throw new BusinessException(RealtimeErrorCode.REDIS_PUBLISH_FAILED); + } + } + +} diff --git a/momogo-realtime/src/main/java/com/momogo/realtime/websocket/redis/RedisMessageSubscriber.java b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/redis/RedisMessageSubscriber.java new file mode 100644 index 0000000..54d4c0f --- /dev/null +++ b/momogo-realtime/src/main/java/com/momogo/realtime/websocket/redis/RedisMessageSubscriber.java @@ -0,0 +1,39 @@ +package com.momogo.realtime.websocket.redis; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.momogo.realtime.websocket.constant.WebSocketConstants; +import com.momogo.realtime.websocket.dto.response.RealtimeMessageResponse; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.messaging.simp.SimpMessagingTemplate; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class RedisMessageSubscriber { + + private final ObjectMapper objectMapper; + private final SimpMessagingTemplate messagingTemplate; + + /** + * RedisPubSubConfig의 MessageListenerAdapter에 의해 호출되는 메소드 + * Redis에서 발행된 메시지를 받아 웹소켓 구독자(/sub/...)들에게 전파합니다. + * @param messageJson + */ + public void handleMessage(String messageJson) { + try { + log.info("[Redis Subscriber] 수신된 Pub/Sub 메시지: {}", messageJson); + + RealtimeMessageResponse response = objectMapper.readValue(messageJson, RealtimeMessageResponse.class); + String destination = WebSocketConstants.SUB_ROOM_PREFIX + response.roomId(); + + messagingTemplate.convertAndSend(destination, response); + log.info("[WebSocket Broadcast] 목적지({})로 상태 메시지 전파 완료", destination); + + } catch (Exception e) { + log.error("[Redis Subscriber] 메시지 처리 중 오류 발생", e); + } + } + +}