Spring WebSocket & STOMP

웹소켓

---

 웹소켓은 텍스트와 바이너리 타입의 메시지를 양방향으로 주고받을 수 있는 프로토콜이다. 게임, 채팅, 실시간 주식 거래 사이트 등 실시간성을 보장하는 서비스에서 사용된다. HTTP에서도 실시간성을 보장하는 기법이 존재한다. (Polling, Long Polling, Streaming) 웹소켓을 지원하지 않는 브라우저에서는 SockJS, Socket.io를 사용하여 HTTP의 다른 기법을 대신 사용하여 웹소켓을 사용하는 거처럼 보이게 할 수 있다.

 

HTTP vs 웹소켓

HTTP	웹소켓
비 연결성 매번 연결 맺고 끊는 과정의 비용 (매 요청마다 수많은 헤더들을 만들어서 보내야 한다) 요청 - 응답 구조	연결 지향 한번 연결 맺은 뒤 유지 양방향 통신 첫 연결은 HTTP를 사용한다

 

설정

@Configuration
@EnableWebSocket
class WebSocketConfig : WebSocketConfigurer {
    override fun registerWebSocketHandlers(registry: WebSocketHandlerRegistry) {

        registry.addHandler(
            SocketTextHandler(),     // 클라이언트가 보내오는 통신을 처리할 핸들러
            "/user"                  // 웹소켓 연결 주소
        )
            .setAllowedOrigins("*") // cors 설정
            .withSockJS()           // sockJS를 사용하도록 설정

    }
}


class SocketTextHandler : TextWebSocketHandler() {
    // 연결 정보를 담고 있는 Set
    private val sessions: MutableSet<WebSocketSession> = ConcurrentHashMap.newKeySet()

    override fun afterConnectionEstablished(session: WebSocketSession) {
        sessions.add(session)
    }

    override fun handleTextMessage(session: WebSocketSession, message: TextMessage) {
        val payload = message.payload
        for (s in sessions) {
            s.sendMessage(TextMessage("Receive Message: $payload"))
        }
    }

    override fun afterConnectionClosed(session: WebSocketSession, status: CloseStatus) {
        sessions.remove(session)
    }
}

 

 

STOMP

---

 Simple Text Oriented Messaging Protocol의 약자로 메시지 브로커를 활용하여 발행-구독 방법으로 쉽게 메시지를 주고받을 수 있는 프로토콜이다. 웹소켓 위에 얹어 함께 사용할 수 있는 하위(서브) 프로토콜이다. 

 웹소켓은 텍스트와 바이너리를 어떤 형식으로 주고 받을지에 대해 따로 정해진 것이 없다. STOMP는 프레임이라고 해서 커맨드, 헤더, 바디로 이미 그 형식을 정의해 두었다.

COMMAND
header1:value1
header2:value2

Body^@

 

 때문에 하위 프로토콜 혹은 컨벤션을 새롭게 정의할 필요가 없고, 연결 주소마다 핸들러를 구현하고 설정해 줄 필요도 없다. 또한 RabbitMQ, Kafka 등 외부 메세지큐를 사용할 수 있으며 spring security를 적용할 수도 있다.

 

설정

@Configuration
@EnableWebSocketMessageBroker // 스프링의 내장 브로커를 사용
class WebSocketBrokerConfig(
    private val stompHandler: StompHandler,
) : WebSocketMessageBrokerConfigurer {

    override fun configureMessageBroker(registry: MessageBrokerRegistry) {
        // prefix가 붙은 경로로 메시지를 발행 시 브로커가 처리
        registry.enableSimpleBroker(
            "/queue", // 1:1
            "/topic"  // 1:N
        )
        // 메시지를 핸들러로 라우팅 되는 Prefix
        registry.setApplicationDestinationPrefixes("/app")
    }

    override fun registerStompEndpoints(registry: StompEndpointRegistry) {
        // 웹소켓 연결 주소, 핸들러를 하나하나 추가할 필요 없다.
        registry.addEndpoint("/websocket").withSockJS()
    }
    
    // 클라이언트 인바운드 채널을 구성하는 메서드
    override fun configureClientInboundChannel(registration: ChannelRegistration) {
        // stompHandler를 인터셉터로 등록하여 STOMP 메시지 핸들링을 수행
        registration.interceptors(stompHandler)
    }

    // STOMP에서 64KB 이상의 데이터 전송을 못하는 문제 해결
    override fun configureWebSocketTransport(registry: WebSocketTransportRegistration) {
        registry.setMessageSizeLimit(160 * 64 * 1024)
        registry.setSendTimeLimit(100 * 10000)
        registry.setSendBufferSizeLimit(3 * 512 * 1024)
    }
}

@Component
class StompHandler: ChannelInterceptor {

    override fun preSend(message: Message<*>, channel: MessageChannel): Message<*>? {
        val accessor = StompHeaderAccessor.wrap(message)
        verifyAccessToken(accessor)
        when (accessor.command) {
            StompCommand.CONNECT -> // do something
            else -> Unit
        }
        return message
    }

    private fun verifyAccessToken(accessor: StompHeaderAccessor) {
        accessor.getHeader("authentication") as? String
            ?: throw RuntimeException("Access token not found in the headers.")
    }
}


@Controller
class GreetingController {

    // /app/hello 경로로 처리한다
    @MessageMapping("/hello")
    // /topic/greeting 브로커로 전송한다
    @SendTo("/topic/greeting")
    fun greeting(message: String): String {
        return "Hello $message"
    }
}