[BE] refactor: stomp로 변경 (#98)

Signed-off-by: EunJiJung <bianbbc87@gmail.com>
sgdevcamp2025 · Mar 5, 2025 · f1bc61a · f1bc61a
1 parent aae3cfa
commit f1bc61a
Show file tree

Hide file tree

Showing 10 changed files with 431 additions and 280 deletions.
diff --git a/...d/signaling-server/src/main/java/com/asyncgate/signaling_server/config/KurentoConfig.java b/...d/signaling-server/src/main/java/com/asyncgate/signaling_server/config/KurentoConfig.java
@@ -2,6 +2,7 @@
 
 import com.asyncgate.signaling_server.infrastructure.client.MemberServiceClient;
 import com.asyncgate.signaling_server.infrastructure.utility.WebClientUtil;
+import com.asyncgate.signaling_server.security.filter.FilterChannelInterceptor;
 import com.asyncgate.signaling_server.security.filter.WebSocketHandshakeInterceptor;
 import com.asyncgate.signaling_server.security.utility.JsonWebTokenUtil;
 import com.asyncgate.signaling_server.signaling.KurentoManager;
@@ -10,26 +11,41 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.web.socket.config.annotation.EnableWebSocket;
 import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
 import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
 import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;
 
+import org.kurento.client.KurentoClient;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.config.ChannelRegistration;
+import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+
 @Configuration
-@EnableWebSocket
-public class KurentoConfig implements WebSocketConfigurer {
+@EnableWebSocketMessageBroker
+public class KurentoConfig implements WebSocketMessageBrokerConfigurer {
 
     @Value("${kms.url}")
     private String kmsUrl;
 
+    private final FilterChannelInterceptor filterChannelInterceptor;
     private final WebSocketHandshakeInterceptor webSocketHandshakeInterceptor;
-    private final JsonWebTokenUtil jsonWebTokenUtil;  // 추가
-
+    private final SimpMessagingTemplate messagingTemplate;
     private final WebClientUtil webClientUtil;
 
-    public KurentoConfig(WebSocketHandshakeInterceptor webSocketHandshakeInterceptor, JsonWebTokenUtil jsonWebTokenUtil, WebClientUtil webClientUtil) {
+    public KurentoConfig(FilterChannelInterceptor filterChannelInterceptor,
+                         WebSocketHandshakeInterceptor webSocketHandshakeInterceptor,
+                         SimpMessagingTemplate messagingTemplate,
+                         WebClientUtil webClientUtil) {
+        this.filterChannelInterceptor = filterChannelInterceptor;
         this.webSocketHandshakeInterceptor = webSocketHandshakeInterceptor;
-        this.jsonWebTokenUtil = jsonWebTokenUtil;  // 추가
+        this.messagingTemplate = messagingTemplate;
         this.webClientUtil = webClientUtil;
     }
 
@@ -45,22 +61,34 @@ public MemberServiceClient memberServiceClient() {
 
     @Bean
     public KurentoManager kurentoManager(KurentoClient kurentoClient, MemberServiceClient memberServiceClient) {
-        return new KurentoManager(kurentoClient, memberServiceClient);
+        return new KurentoManager(kurentoClient, memberServiceClient, messagingTemplate);
     }
 
     @Bean
     public KurentoHandler kurentoHandler(KurentoManager kurentoManager) {
-        return new KurentoHandler(kurentoManager, jsonWebTokenUtil); // JsonWebTokenUtil 주입
+        return new KurentoHandler(kurentoManager);
     }
 
     @Override
-    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
-        System.out.println("🚀 WebSocketHandlerRegistry 등록");
-        registry.addHandler(kurentoHandler(kurentoManager(kurentoClient(), memberServiceClient())), "/signal")
+    public void registerStompEndpoints(StompEndpointRegistry registry) {
+        registry.addEndpoint("/signal")
                 .setAllowedOriginPatterns("*")
                 .addInterceptors(webSocketHandshakeInterceptor);
     }
 
+    /*
+    @Override
+    public void configureMessageBroker(MessageBrokerRegistry registry) {
+        registry.setApplicationDestinationPrefixes("/kafka");
+        registry.enableSimpleBroker("/topic/");
+    }
+     */
+
+    @Override
+    public void configureClientInboundChannel(ChannelRegistration registration) {
+        registration.interceptors(filterChannelInterceptor);
+    }
+
     @Bean
     public ServletServerContainerFactoryBean createWebSocketContainer() {
         ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();

diff --git a/...er/src/main/java/com/asyncgate/signaling_server/controller/ChatRoomCommandController.java b/...er/src/main/java/com/asyncgate/signaling_server/controller/ChatRoomCommandController.java
@@ -15,7 +15,7 @@
 public class ChatRoomCommandController {
 
     private final JoinRoomUseCase joinRoomUseCase;
-    private final GetUsersInRoomUseCase getUsersInRoomUseCase;
+    // private final GetUsersInRoomUseCase getUsersInRoomUseCase;
 
     /**
      * 채널 참여
@@ -26,13 +26,12 @@ public SuccessResponse<String> joinRoom(@PathVariable("room_id") final String ro
         return SuccessResponse.ok("room: " + roomId + "에 user: " + memberId + "가 참여하였습니다.");
     }
 
-    /**
-     * 채널에 참여 중인 유저 조회
-     */
+    /*
     @GetMapping("/{room_id}/users")
     public SuccessResponse<GetUsersInChannelResponse> getUsersInRoom(@PathVariable("room_id") String roomId) {
         return SuccessResponse.ok(getUsersInRoomUseCase.execute(roomId));
     }
+    */
 
     /**
      *

diff --git a/...server/src/main/java/com/asyncgate/signaling_server/controller/StompWebRtcController.java b/...server/src/main/java/com/asyncgate/signaling_server/controller/StompWebRtcController.java
@@ -0,0 +1,66 @@
+package com.asyncgate.signaling_server.controller;
+
+import com.asyncgate.signaling_server.dto.request.KurentoOfferRequest;
+import com.asyncgate.signaling_server.signaling.KurentoManager;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.messaging.handler.annotation.MessageMapping;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+import org.springframework.stereotype.Controller;
+
+@Controller
+@RequiredArgsConstructor
+public class StompWebRtcController {
+
+    private static final Logger log = LoggerFactory.getLogger(StompWebRtcController.class);
+    private final SimpMessagingTemplate messagingTemplate;
+    private final KurentoManager kurentoManager;
+
+    /**
+     * WebRTC Offer 메시지 처리
+     */
+    @MessageMapping("/offer")
+    public void handleOffer(KurentoOfferRequest message, StompHeaderAccessor accessor) {
+        // 메시지 처리 및 Kurento에서 사용자 연결
+        kurentoManager.processSdpOffer(message, accessor);
+
+        kurentoManager.startIceCandidateListener(message, accessor);
+    }
+
+    /**
+     * ICE Candidate 메시지 처리
+     */
+    @MessageMapping("/candidate")
+    public void handleIceCandidate(KurentoOfferRequest message, StompHeaderAccessor accessor) {
+
+        // Kurento에 ICE Candidate 전달
+        kurentoManager.addIceCandidates(message, accessor);
+    }
+
+    /**
+     * 미디어 토글 (AUDIO, MEDIA, DATA)
+     */
+    @MessageMapping("/toggle")
+    public void toggleMediaState(KurentoOfferRequest message, StompHeaderAccessor accessor) {
+        kurentoManager.updateUserMediaState(message, accessor);
+    }
+
+    /**
+     * 유저 정보 반환
+     */
+    @MessageMapping("/users")
+    public void getUsersInRoom(KurentoOfferRequest message, StompHeaderAccessor accessor) {
+        kurentoManager.getUsersInChannel(message);
+    }
+
+    /**
+     * WebRTC 종료 처리
+     */
+    @MessageMapping("/exit")
+    public void handleExit(KurentoOfferRequest message, StompHeaderAccessor accessor) {
+        // Kurento에서 사용자 제거
+        kurentoManager.removeUserFromChannel(message, accessor);
+    }
+}
diff --git a/...-server/src/main/java/com/asyncgate/signaling_server/dto/request/KurentoOfferRequest.java b/...-server/src/main/java/com/asyncgate/signaling_server/dto/request/KurentoOfferRequest.java
@@ -0,0 +1,32 @@
+package com.asyncgate.signaling_server.dto.request;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.kurento.client.IceCandidate;
+import org.springframework.lang.Nullable;
+
+
+public record KurentoOfferRequest(
+
+        @JsonProperty("type")
+        String type,
+
+        @JsonProperty("data")
+        KurentoOfferData data
+) {
+    public record KurentoOfferData(
+            @JsonProperty("room_id")
+            String roomId,
+
+            @Nullable
+            @JsonProperty("sdp_offer")
+            String sdpOffer,
+
+            @Nullable
+            @JsonProperty("candidate")
+            IceCandidate candidate,
+
+            @Nullable
+            @JsonProperty("enabled")
+            boolean enabled
+    ) {}
+}
diff --git a/...rver/src/main/java/com/asyncgate/signaling_server/dto/response/KurentoAnswerResponse.java b/...rver/src/main/java/com/asyncgate/signaling_server/dto/response/KurentoAnswerResponse.java
@@ -0,0 +1,21 @@
+package com.asyncgate.signaling_server.dto.response;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Builder;
+import lombok.Getter;
+
+@Getter
+public class KurentoAnswerResponse {
+
+    @JsonProperty("type")
+    private String type;
+
+    @JsonProperty("message")
+    private String message;
+
+    @Builder
+    public KurentoAnswerResponse(String type, String message) {
+        this.type = type;
+        this.message = message;
+    }
+}
diff --git a/...rc/main/java/com/asyncgate/signaling_server/security/filter/FilterChannelInterceptor.java b/...rc/main/java/com/asyncgate/signaling_server/security/filter/FilterChannelInterceptor.java
@@ -0,0 +1,101 @@
+package com.asyncgate.signaling_server.security.filter;
+
+import com.asyncgate.signaling_server.security.constant.Constants;
+import com.asyncgate.signaling_server.security.utility.JsonWebTokenUtil;
+import io.jsonwebtoken.Claims;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.HttpStatus;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.simp.stomp.StompCommand;
+import org.springframework.messaging.simp.stomp.StompHeaderAccessor;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.stereotype.Component;
+import org.springframework.web.server.ResponseStatusException;
+
+@Component
+public class FilterChannelInterceptor implements ChannelInterceptor {
+
+    private static final Logger log = LoggerFactory.getLogger(FilterChannelInterceptor.class);
+    private static final String AUTHORIZATION_HEADER = "Authorization";
+    private static final String BEARER_PREFIX = "Bearer ";
+    private final JsonWebTokenUtil jsonWebTokenUtil;
+
+    public FilterChannelInterceptor(JsonWebTokenUtil jsonWebTokenUtil) {
+        this.jsonWebTokenUtil = jsonWebTokenUtil;
+    }
+
+    private String extractToken(String headerValue) {
+        if (headerValue == null || headerValue.trim().isEmpty()) {
+            return null;
+        }
+        String token = headerValue.trim();
+        return token.toLowerCase().startsWith(BEARER_PREFIX.toLowerCase()) ? token.substring(BEARER_PREFIX.length()) : token;
+    }
+
+    @Override
+    public Message<?> preSend(Message<?> message, MessageChannel channel) {
+        StompHeaderAccessor headerAccessor = StompHeaderAccessor.wrap(message);
+        log.info("📥 [STOMP] Command: {}, sessionId: {}", headerAccessor.getCommand(), headerAccessor.getSessionId());
+
+        if (StompCommand.CONNECT.equals(headerAccessor.getCommand())) {
+            String rawAuth = headerAccessor.getFirstNativeHeader(AUTHORIZATION_HEADER);
+            log.info("🔑 [STOMP] Raw Authorization Header: {}", rawAuth);
+            String jwtToken = extractToken(rawAuth);
+            if (jwtToken == null || jwtToken.isEmpty()) {
+                log.error("🚨 [STOMP] Access Token is missing or improperly formatted!");
+                throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "Access token is missing");
+            }
+
+            if(headerAccessor.getSessionAttributes().get("userId") == null) {
+                throw new ResponseStatusException(HttpStatus.UNAUTHORIZED, "userId is missing");
+            }
+            log.info("✅ [STOMP] CONNECT 요청 처리 완료");
+        }
+        return message;
+    }
+
+    @Override
+    public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+        StompHeaderAccessor accessor = StompHeaderAccessor.wrap(message);
+        log.info("📡 [STOMP] Command: {}, sessionId: {}, sent: {}", accessor.getCommand(), accessor.getSessionId(), sent);
+
+        if (StompCommand.CONNECT.equals(accessor.getCommand())) {
+            log.info("✅ [STOMP] CONNECT 성공 - sessionId: {}", accessor.getSessionId());
+            handleConnect(accessor);
+            log.info("🔎 [STOMP] CONNECTED 프레임 헤더: {}", accessor.getMessageHeaders());
+            accessor.getMessageHeaders().forEach((key, value) -> log.info("messageHeader = {}: {}", key, value));
+        } else if (StompCommand.DISCONNECT.equals(accessor.getCommand())) {
+            log.info("🔌 [STOMP] DISCONNECT 요청 - sessionId: {}", accessor.getSessionId());
+            handleDisconnect(accessor);
+        }
+    }
+
+    private void handleDisconnect(StompHeaderAccessor accessor) {
+        log.info("🔌 [STOMP] WebSocket 연결 해제 - sessionId: {}", accessor.getSessionId());
+    }
+
+    private void handleConnect(StompHeaderAccessor accessor) {
+        String currentSessionId = accessor.getSessionId();
+        if (currentSessionId == null) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Not session now");
+        }
+        String rawAuth = accessor.getFirstNativeHeader(AUTHORIZATION_HEADER);
+        String jwtToken = extractToken(rawAuth);
+        if (jwtToken == null) {
+            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "JWT token is missing");
+        }
+
+        Claims claims = jsonWebTokenUtil.validate(jwtToken);
+
+        String memberId = claims.get(Constants.MEMBER_ID_CLAIM_NAME, String.class);
+
+        // LoginSessionRequest loginSessionRequest = new LoginSessionRequest(LoginType.LOGIN, currentSessionId, currentUserId);
+        // StateRequest stateRequest = new StateRequest(StatusType.CONNECT, currentUserId);
+        // ToDo: 상태관리 서버에 로그인 전달 (주석 유지)
+        // val guildIds = guildClient.getGuildIds(jwtToken)
+        // 시그널링 서버에 전달 (주석 유지)
+        // messageSender.signaling(stateTopic, stateRequest)
+    }
+}