Feat: impl STOMP Websocket Redis pub/sub message relay with DTO and constants

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();
+  }
+}

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;
+  }
+}

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() {
+
+  }
+}

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);
+  }
+}

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
+) {
+
+}

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()
+    );
+  }
+}

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;
+}

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);
+    }
+  }
+
+}

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);
+    }
+  }
+
+}