[Spring] STOMP 설정과 사용하는 방법 쉽게 이해하기

 

왜 STOMP를 써야할까?

 

• Simple Text Oriented Messaging Protocol의 약자입니다.
• WebSocket는 단순히 클라이언트, 서버 양방향 통신을 열어줄 뿐이지만 STOMP는 메시지 구조나 목적지 개념을 가지고 있습니다.
• 그래서 메시지를 전송(Publish) 하면 브로커가 중개하여 목적지(Destination)에 구독(Subscribe)한 여러 클라이언트에게 전달합니다.
• 실시간 채팅, 알림 시스템에 적합합니다.

 

 

 

 

STOMP PUB/SUB 메시지 흐름 이해하기

 

이미지로 이해하기

[그림1] STOPM 메시지 흐름도

[그림2] 이해한 흐름도

텍스트로 이해하기

1. 웹소켓 클라이언트(사용자 및 서버)가 메시지를 전송한다.
2. 서버는 메시지를 Request Channel로 보낸다.
3. (임의로) /app/** 으로 시작하는 경로면 정의 해둔 스프링의 컨트롤러가 호출된다.
   (임의로) /topic/** 으로 시작하는 경로면 스프링 컨트롤러를 거치지 않고 브로커로 직접 전달한다.
4. ((임의로) /app/** 경로로 왔을 때만) 비즈니스로직이 다 실행되고 클라이언트들에게 전송해야할 메시지를 브로커 채널에 보낸다.
5. 브로커는 클라이언트들에게 메시지 전송한다.

 

즉, 클라이언트 및 서버가 Publish(발행) 하면 그것을 Subscribe(구독) 하고 있는 클라이언트는 메시지를 받게 되는 구조 입니다.

 

 

 

웹소켓 의존성 추가

 

각자 자신의 빌드 도구에 맞춰 의존성을 추가해주시면 됩니다.

build.gradle

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-websocket'
}

pom.xml

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

 

 

 

 

웹 소켓 설정

 

웹소켓을 설정할 수 있는 Java, Kotlin으로 설정하기 위해 Config 파일을 만드시면 됩니다.

 

• 프론트에서  http://localhost:8080/ws로 연결하기 위해 Endpoint를 /ws로 두었습니다.
• 스프링 컨트롤러가 받는 경로를 /pub/**로 받게 해두었습니다.
• 구독한 클라이언트에게 발행을 해버리기 위한 경로는 /sub/** 로 두었습니다.
• 인터셉터가 필요하다면 구현하고 빈으로 등록해서 인터셉터도 등록하면 되지 않을까요?!

WebSocketConfig.java

import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.simp.config.ChannelRegistration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

@Slf4j
@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    // STOMP 엔드포인트 설정
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 클라이언트가 이 소켓에 연결할 주소
        registry.addEndpoint("/ws")     // 경로 여러개 추가 가능
                .withSockJS();
    }

    // 메시지 브로커 설정
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 클라이언트가 웹소켓을 받을 Prefix 엔트포인트 = 구독 주소
        registry.enableSimpleBroker("/sub");

        // 클라이언트가 웹소켓 보낼 때 사용하는 경로 = 메시지 전송 주소
        registry.setApplicationDestinationPrefixes("/pub");
    }


    // 클라이언트 -> 서버 메시지 인터셉터 추가
    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                log.info("클라이언트가 보낸 메시지 : {}", message);
                return message; // 메시지 로깅
            }
        });
    }

    // 서버 -> 클라이언트 메시지 인터셉터 추가
    @Override
    public void configureClientOutboundChannel(ChannelRegistration registration) {
        registration.interceptors(new ChannelInterceptor() {
            @Override
            public Message<?> preSend(Message<?> message, MessageChannel channel) {
                log.info("서버가 보내는 메시지 {}", message);
                return message; // 메시지 로깅
            }
        });
    }

    // 메시지 크기 제한, 버퍼 크기, 타임아웃 등의 설정
//    @Override
//    public void configureWebSocketTransport(WebSocketTransportRegistration registry) {
//        registry.setMessageSizeLimit(8192)  // 메시지 최대 크기 (8KB) (기본 값 : 64KB (65536 bytes))
//                .setSendBufferSizeLimit(8192) // 전송 버퍼 크기 제한 (기본 값 512KB (524288 bytes))
//                .setTimeToFirstMessage(5000); // 최초 메시지 수신 대기 시간 (5초) (제한 없음, 기본값 없음)
//    }
}

• registerStompEndpoints
  : 클라이언트가 WebSocket 연결하려는 STOMP 엔드포인트 설정
• configureMessageBroker
  : 메시지 브로커 설정
• configureClientInboundChannel
  : 클라이언트 → 서버 메시지 인터셉터
• configureClientOutboundChannel
  : 서버 → 클라이언트 메시지 인터셉터
• configureWebSocketTransport
  : 메시지 크기, 버퍼, 타임 아웃 여러가지 설정

https://docs.spring.io/spring-framework/reference/web/websocket/stomp/message-flow.html

Flow of Messages :: Spring Framework

 

 

 

 

이벤트 리스너 등록

 

이벤트 리스너를 등록하면 이 이벤트가 일어났을 때의 디테일한 컨트롤이 가능해지기 때문에 아래에 있는 5개의 이벤트를 꼭 알아두시기 바랍니다.

 

• 연결 시도
• 연결 완료
• 구독
• 구독 취소
• 연결 끊기

이것을 왜? 등록을 하냐면! 아래의 2가지 이유가 있습니다.

 

1. 원하는 시점에 비즈니스 로직을 녹일 수 가 있습니다.
2. 현재 소켓에 연결된 사람들을 관리 할 수 있습니다.

 

연결 시도 (SessionConnectEvent)

• 클라이언트가 연결을 시도하는 초기 단계이며, 세션ID, 헤더 등을 포함하고 있습니다.

// 소켓 연결 신청
@EventListener
public void handleConnect(SessionConnectEvent sessionConnectEvent) {
    log.info("sessionConnectEvent : {}", sessionConnectEvent);
}

 

연결 완료 (SessionConnectedEvent)

• 연결이 실제로 성립한 상태이며, STOMP 세션이 완전히 열렸습니다.

// 소켓 연결 완료
@EventListener
public void handleConnected(SessionConnectedEvent sessionConnectedEvent) {
    log.info("sessionConnectedEvent : {}", sessionConnectedEvent);
    String sessionId = sessionConnectedEvent.getMessage().getHeaders().get("simpSessionId").toString();
    sessionMap.put(sessionId, sessionId);
}

 

구독 (SessionSubscribeEvent)

• 특정 경로(채널) 구독 요청 했다는 것입니다.

// 사용자가 구독
@EventListener
public void handleSubscribe(SessionSubscribeEvent sessionSubscribeEvent) {
    log.info("sessionSubscribeEvent : {}", sessionSubscribeEvent);
}

 

구독 취소 (SessionUnsubscribeEvent)

• 특정 경로(채널) 구독 해지 요청을 했다는 것입니다.

// 사용자가 구독 취소
@EventListener
public void handleUnsubscribe(SessionUnsubscribeEvent sessionUnsubscribeEvent) {
    log.info("sessionUnsubscribeEvent : {}", sessionUnsubscribeEvent);
}

 

연결 끊기 (SessionDisconnectEvent)

• STOMP 세션이 종료 된 것입니다.

// 소켓 연결 끊기
@EventListener
public void handleDisconnect(SessionDisconnectEvent sessionDisconnectEvent) {
    log.info("sessionDisconnectEvent : {}", sessionDisconnectEvent);
    String sessionId = sessionDisconnectEvent.getMessage().getHeaders().get("simpSessionId").toString();
    sessionMap.remove(sessionId);
}

 

이벤트 리스너 등록한 전체 코드 보기

@Slf4j
@Component
public class StompEventListener {

    // 실시간 연결된 사용자 세션 ID들
    private final ConcurrentHashMap<String, String> sessionMap = new ConcurrentHashMap<>();

    // 모든 세션 ID
    public Set<String> getUsersInfo() {
        return sessionMap.keySet();
    }

    // 소켓 연결 신청
    @EventListener
    public void handleConnect(SessionConnectEvent sessionConnectEvent) {
        log.info("sessionConnectEvent : {}", sessionConnectEvent);
    }

    // 소켓 연결 완료
    @EventListener
    public void handleConnected(SessionConnectedEvent sessionConnectedEvent) {
        log.info("sessionConnectedEvent : {}", sessionConnectedEvent);
        String sessionId = sessionConnectedEvent.getMessage().getHeaders().get("simpSessionId").toString();
        sessionMap.put(sessionId, sessionId);
    }

    // 사용자가 구독
    @EventListener
    public void handleSubscribe(SessionSubscribeEvent sessionSubscribeEvent) {
        log.info("sessionSubscribeEvent : {}", sessionSubscribeEvent);
    }

    // 사용자가 구독 취소
    @EventListener
    public void handleUnsubscribe(SessionUnsubscribeEvent sessionUnsubscribeEvent) {
        log.info("sessionUnsubscribeEvent : {}", sessionUnsubscribeEvent);
    }

    // 소켓 연결 끊기
    @EventListener
    public void handleDisconnect(SessionDisconnectEvent sessionDisconnectEvent) {
        log.info("sessionDisconnectEvent : {}", sessionDisconnectEvent);
        String sessionId = sessionDisconnectEvent.getMessage().getHeaders().get("simpSessionId").toString();
        sessionMap.remove(sessionId);
    }


}

 

https://docs.spring.io/spring-framework/reference/web/websocket/stomp/application-context-events.html

Events :: Spring Framework

 

 

 

 

 

발행하여 메시지를 받는 컨트롤러 등록

 

• Spring MVC의 @RequestMapping과 되게 유사한 느낌을 받으실 겁니다.
• 클라이언트가 메시지를 발행 했는데 스프링 어플리케이션의 어떤 메서드와 메시지를 매핑할지 결정하는 어노테이션이 @MessageMapping()이며, N개의 경로를 등록할 수 있습니다.
• 즉, 여러 곳에서 발행되는 메시지를 이 메서드 한개가 처리할 수 있게 됩니다.

 

사용 예시

@Controller
public class SocketController {

    /****************************
     * @MessageMapping : 클라이언트가 보낸 주소 (config에 설정한 값)/(매핑 경로) (pub/{roomId})
     * @SendTo : 서버가 클라이언트에게 보내는 주소
     ***************************/

    @MessageMapping("/{roomId}")
    @SendTo("/sub/room/{roomId}")
    public String sendMessage(@DestinationVariable(value = "roomId") String roomId,
                                              @Payload String message,
                                              Message<?> message,
                                              MessageHeaders messageHeaders) {
		// 로직 생략

        return chatMessage.getMessage();
    }
    
}

@DestinationVariable

• 스프링 컨트롤러의 @PathVariable 과 같은 성격을 가집니다.
• 즉, 메시지의 경로의 변수에 접근하기 위한 어노테이션 입니다.

Message

• 소켓으로 전달 받은 메시지의 모든 정보입니다.

MessageHeader

• 소켓으로 전달 받은 메시지의 header값 입니다.

@Payload

• 클라이언트가 보낸 실제 메시지입니다.
• String, Object, 여러분이 만든 ChatMessageRequest 객체든 다 Converter가 일해주니 클라이언트단에서 던진걸로 받으시면 됩니다!!

https://docs.spring.io/spring-framework/reference/web/websocket/stomp/handle-annotations.html

Annotated Controllers :: Spring Framework

 

 

 

 

구독자에게 메시지를 전송하는 방법

 

방법1) SimpMessagingTemplate을 이용해 프로그래밍 방식으로 메시지 발송

• 구독자에게 메시지를 전달하기 전에 비즈니스 로직이 실행되거나 분기처리로 인하여 동적으로 메시지를 보내야 한다고 생각하시면 됩니다.
• 이제 1회성 메시지 전송 뿐만 아니라 특정 로직에 따라 재전송을 할 수 있게 되는 것입니다.
• 예를 들어, 특정 사용자에게만 메시지를 보내거나, 여러 채널로 메시지를 동시에 전송하는 등의 유연한 처리가 가능합니다.

@Controller
@RequiredArgsConstructor
public class SocketController {

    private final SimpMessagingTemplate template;

    /****************************
     * @MessageMapping : 클라이언트가 보낸 주소 (config에 설정한 값)/(매핑 경로) (pub/{roomId})
     * SendTo : 서버가 클라이언트에게 보내는 주소
     ***************************/

    @MessageMapping("/{roomId}")
    public void sendMessage(@DestinationVariable(value = "roomId") String roomId,
                                                  @Payload MessageRequest chatMessage,
                                                  Message<?> message,
                                                  MessageHeaders messageHeaders) {

        template.convertAndSend("/sub/room/"+roomId, chatMessage.getMessage());
    }
}

방법2) @SendTo로 간편하게 메시지 발송

• 어노테이션으로 편하게 작성할 수 있습니다.
• 응답 값을 리턴만 하면 자동으로 목적지까지(구독 한 경로)로 메시지가 전달됩니다.
• 재사용하는 것이 아니고 1회성 전송에 유용합니다.
• 즉, 단순한 간단한 메시지를 전송할 때와 목적지가 고정되어 있을 때 매우 유용합니다.

@Controller
public class SocketController {

    /****************************
     * @MessageMapping : 클라이언트가 보낸 주소 (config에 설정한 값)/(매핑 경로) (pub/{roomId})
     * @SendTo : 서버가 클라이언트에게 보내는 주소
     ***************************/

    @MessageMapping("/{roomId}")
    @SendTo("/sub/room/{roomId}")
    public String sendMessage(@DestinationVariable(value = "roomId") String roomId,
                                              @Payload MessageRequest chatMessage,
                                              Message<?> message,
                                              MessageHeaders messageHeaders) {
		// 로직 생략

        return chatMessage.getMessage();
    }
    
}

 

https://docs.spring.io/spring-framework/reference/web/websocket/stomp/handle-annotations.html#return-values

Annotated Controllers :: Spring Framework

 

 

 

 

 

소켓 전송 예외처리

 

Spring MVC에서 예외처리는 했을 때 @ExceptionHandler를 사용하여 처리하였습니다.

Spring WebSocket도 비슷하게 @Controller 내부에 있는 @MessageMapping에서 발생한 예외를 @MessageExceptionHandler를 이용하여 처리합니다.

 

• 클래스를 분리하여 예외 처리만 하는 클래스로 분리하셔도 됩니다.
• 채팅하는 것도 예외로 내려주는 것보다 서버에서 잡아서 커스텀 예외 또는 그에 맞는 다른 채팅으로 변환하여 내려주면 좋을 것 같습니다.

사용 예시

@ControllerAdvice
public class StompExceptionHandler {

    @MessageExceptionHandler(ChatBusinessException.class)
    @SendToUser("/queue/errors")
    public ErrorResponse handleBusinessException(ChatBusinessException ex) {
        return new ErrorResponse("CHAT_ERROR", ex.getMessage());
    }

    @MessageExceptionHandler(Exception.class)
    @SendToUser("/queue/errors")
    public ErrorResponse handleGeneric(Exception ex) {
        return new ErrorResponse("SERVER_ERROR", "알 수 없는 오류가 발생했습니다.");
    }
}

 

https://docs.spring.io/spring-framework/reference/web/websocket/stomp/handle-annotations.html#websocket-stomp-exception-handler

Annotated Controllers :: Spring Framework

 

 

 

감사합니다.