WebSockets

See equivalent in the Servlet stack

이 참고 문서의 이 부분은 reactive-stack WebSocket messaging에 대한 지원을 다룹니다.

Introduction to WebSocket

WebSocket 프로토콜, RFC 6455는 하나의 TCP 연결을 통해 client와 server 사이에 full-duplex, 양방향 통신 채널을 설정하는 표준화된 방식을 제공합니다. 이는 HTTP와는 다른 TCP 프로토콜이지만, 포트 80과 443을 사용하고 기존 방화벽 규칙의 재사용을 허용하면서 HTTP 위에서 동작하도록 설계되었습니다.

WebSocket 상호 작용은 HTTP Upgrade 헤더를 사용하여 WebSocket 프로토콜로 업그레이드(이 경우에는 전환)하는 HTTP 요청으로 시작됩니다. 다음 예제는 이러한 상호 작용을 보여줍니다:

1GET /spring-websocket-portfolio/portfolio HTTP/1.1
2Host: localhost:8080
3Upgrade: websocket (1)
4Connection: Upgrade (2)
5Sec-WebSocket-Key: Uc9l9TMkWGbHFD2qnFHltg==
6Sec-WebSocket-Protocol: v10.stomp, v11.stomp
7Sec-WebSocket-Version: 13
8Origin: http://localhost:8080

1	Upgrade 헤더.
2	Upgrade 연결 사용.

일반적인 200 상태 코드 대신, WebSocket 지원이 있는 server는 다음과 유사한 출력을 반환합니다:

1HTTP/1.1 101 Switching Protocols (1)
2Upgrade: websocket
3Connection: Upgrade
4Sec-WebSocket-Accept: 1qVdfYHU9hPOl4JYYNXF623Gzn0=
5Sec-WebSocket-Protocol: v10.stomp

1	프로토콜 전환

핸드셰이크가 성공하면, HTTP 업그레이드 요청의 기반이 되는 TCP 소켓은 client와 server 모두가 메시지를 계속 보내고 받을 수 있도록 열린 상태로 유지됩니다.

WebSocket이 어떻게 동작하는지에 대한 완전한 소개는 이 문서의 범위를 벗어납니다. RFC 6455, HTML5의 WebSocket 장, 또는 Web 상의 많은 소개와 튜토리얼을 참조하십시오.

WebSocket server가 웹 서버(예: nginx) 뒤에서 실행되는 경우, WebSocket 업그레이드 요청을 WebSocket server로 전달하도록 이를 구성해야 할 가능성이 큽니다. 마찬가지로, 애플리케이션이 클라우드 환경에서 실행되는 경우, WebSocket 지원과 관련하여 클라우드 제공자의 안내를 확인하십시오.

HTTP Versus WebSocket

WebSocket은 HTTP와 호환되도록 설계되었고 HTTP 요청으로 시작되지만, 두 프로토콜은 매우 다른 아키텍처와 애플리케이션 프로그래밍 모델로 이어진다는 점을 이해하는 것이 중요합니다.

HTTP와 REST에서는 애플리케이션이 많은 URL로 모델링됩니다. 애플리케이션과 상호 작용하기 위해 client는 이러한 URL에 요청-응답 스타일로 접근합니다. 서버는 HTTP URL, 메서드, 헤더를 기반으로 적절한 핸들러로 요청을 라우팅합니다.

대조적으로, WebSocket에서는 초기 연결을 위한 URL이 일반적으로 하나만 있습니다. 이후에는 모든 애플리케이션 메시지가 동일한 TCP 연결 위에서 흐릅니다. 이는 완전히 다른 비동기, 이벤트 기반, 메시징 아키텍처를 가리킵니다.

WebSocket은 또한 저수준 전송 프로토콜이며, HTTP와 달리 메시지 내용에 대해 어떠한 시맨틱도 규정하지 않습니다. 이는 client와 server가 메시지 시맨틱에 합의하지 않는 한 메시지를 라우팅하거나 처리할 방법이 없다는 것을 의미합니다.

WebSocket client와 server는 HTTP 핸드셰이크 요청의 Sec-WebSocket-Protocol 헤더를 통해 (예: STOMP와 같은) 상위 수준 메시징 프로토콜의 사용을 협상할 수 있습니다. 그것이 없는 경우, 자체적인 규칙을 만들어야 합니다.

When to Use WebSockets

WebSocket은 웹 페이지를 동적이고 인터랙티브하게 만들 수 있습니다. 그러나 많은 경우, AJAX와 HTTP 스트리밍 또는 롱 폴링의 조합이 더 단순하고 효과적인 솔루션을 제공할 수 있습니다.

예를 들어, 뉴스, 메일, 소셜 피드는 동적으로 업데이트될 필요가 있지만, 몇 분마다 한 번씩 그렇게 해도 전혀 문제가 없을 수 있습니다. 반면, 협업, 게임, 금융 앱은 훨씬 더 실시간에 가까워야 합니다.

지연 시간만으로는 결정 요소가 되지 않습니다. 메시지 볼륨이 상대적으로 낮은 경우(예: 네트워크 장애 모니터링) HTTP 스트리밍 또는 폴링이 효과적인 솔루션을 제공할 수 있습니다. WebSocket 사용에 가장 적합한 경우는 낮은 지연 시간, 높은 빈도, 높은 볼륨의 조합입니다.

또한 Internet 상에서는, 여러분의 통제 밖에 있는 제한적인 프록시가 Upgrade 헤더를 전달하도록 구성되지 않았거나 유휴 상태로 보이는 장기 연결을 종료하기 때문에 WebSocket 상호 작용을 금지할 수 있다는 점을 명심하십시오. 이는 방화벽 내부의 내부 애플리케이션에 WebSocket을 사용하는 것이 public facing 애플리케이션에 사용하는 것보다 더 간단한 결정이라는 것을 의미합니다.

WebSocket API

See equivalent in the Servlet stack

Spring Framework는 WebSocket 메시지를 처리하는 client-side 및 server-side 애플리케이션을 작성하는 데 사용할 수 있는 WebSocket API를 제공합니다.

Server

See equivalent in the Servlet stack

WebSocket server를 만들려면 먼저 WebSocketHandler를 만들 수 있습니다. 다음 예제는 그 방법을 보여줍니다:

1import org.springframework.web.reactive.socket.WebSocketHandler;
2import org.springframework.web.reactive.socket.WebSocketSession;
3
4public class MyWebSocketHandler implements WebSocketHandler {
5
6	@Override
7	public Mono<Void> handle(WebSocketSession session) {
8		// ...
9	}
10}

1import org.springframework.web.reactive.socket.WebSocketHandler
2import org.springframework.web.reactive.socket.WebSocketSession
3
4class MyWebSocketHandler : WebSocketHandler {
5
6	override fun handle(session: WebSocketSession): Mono<Void> {
7		// ...
8	}
9}

그런 다음 이를 URL에 매핑할 수 있습니다:

1@Configuration
2class WebConfig {
3
4	@Bean
5	public HandlerMapping handlerMapping() {
6		Map<String, WebSocketHandler> map = new HashMap<>();
7		map.put("/path", new MyWebSocketHandler());
8		int order = -1; // before annotated controllers
9
10		return new SimpleUrlHandlerMapping(map, order);
11	}
12}

1@Configuration
2class WebConfig {
3
4	@Bean
5	fun handlerMapping(): HandlerMapping {
6		val map = mapOf("/path" to MyWebSocketHandler())
7		val order = -1 // before annotated controllers
8
9		return SimpleUrlHandlerMapping(map, order)
10	}
11}

WebFlux Config를 사용하는 경우 추가로 할 일은 없으며, 그렇지 않고 WebFlux 설정을 사용하지 않는 경우 아래와 같이 WebSocketHandlerAdapter를 선언해야 합니다:

1@Configuration
2class WebConfig {
3
4	// ...
5
6	@Bean
7	public WebSocketHandlerAdapter handlerAdapter() {
8		return new WebSocketHandlerAdapter();
9	}
10}

1@Configuration
2class WebConfig {
3
4	// ...
5
6	@Bean
7	fun handlerAdapter() = WebSocketHandlerAdapter()
8}

WebSocketHandler

WebSocketHandler의 handle 메서드는 WebSocketSession을 받고 애플리케이션에서 세션 처리가 완료되는 시점을 나타내기 위해 Mono<Void>를 반환합니다. 세션은 inbound와 outbound 메시지 각각에 대한 두 개의 스트림을 통해 처리됩니다.

다음 표는 스트림을 처리하는 두 메서드를 설명합니다:

WebSocketSession method	Description
Flux<WebSocketMessage> receive()	inbound 메시지 스트림에 대한 접근을 제공하며 연결이 close될 때 완료됩니다.
Mono<Void> send(Publisher<WebSocketMessage>)	outgoing 메시지의 소스를 받아 메시지를 기록하고,<br>소스가 완료되고 기록이 완료되면 완료되는 Mono<Void>를 반환합니다.

WebSocketHandler는 inbound와 outbound 스트림을 통합된 플로우로 구성하고, 해당 플로우의 완료를 반영하는 Mono<Void>를 반환해야 합니다. 애플리케이션 요구 사항에 따라 통합된 플로우는 다음과 같은 시점에 완료됩니다:

• inbound 또는 outbound 메시지 스트림 중 하나가 완료될 때.
• inbound 스트림이 완료될 때(즉, 연결이 close될 때), outbound 스트림은 무한인 경우.
• 선택한 시점에 WebSocketSession의 close 메서드를 통해.

inbound와 outbound 메시지 스트림이 함께 구성되면, Reactive Streams 시그널이 activity 종료를 알리므로 연결이 open 상태인지 여부를 확인할 필요가 없습니다. inbound 스트림은 completion 또는 error 시그널을 받고, outbound 스트림은 cancellation 시그널을 받습니다.

handler의 가장 기본적인 구현은 inbound 스트림을 처리하는 것입니다. 다음 예제는 그러한 구현을 보여줍니다:

1class ExampleHandler implements WebSocketHandler {
2
3	@Override
4	public Mono<Void> handle(WebSocketSession session) {
5		return session.receive()            // (1)
6				.doOnNext(message -> {
7					// ...                 // (2)
8				})
9				.concatMap(message -> {
10					// ...                 // (3)
11				})
12				.then();                   // (4)
13	}
14}

1	inbound 메시지 스트림에 access합니다.
2	각 메시지에 대해 무언가를 수행합니다.
3	메시지 내용을 사용하는 중첩 비동기 작업을 수행합니다.
4	수신이 완료될 때 완료되는 Mono<Void>를 반환합니다.

1class ExampleHandler : WebSocketHandler {
2
3	override fun handle(session: WebSocketSession): Mono<Void> {
4		return session.receive()            // (1)
5				.doOnNext {
6					// ...                 // (2)
7				}
8				.concatMap {
9					// ...                 // (3)
10				}
11				.then()                    // (4)
12	}
13}

1	inbound 메시지 스트림에 access합니다.
2	각 메시지에 대해 무언가를 수행합니다.
3	메시지 내용을 사용하는 중첩 비동기 작업을 수행합니다.
4	수신이 완료될 때 완료되는 Mono<Void>를 반환합니다.

중첩, 비동기 작업의 경우, pooled data buffer를 사용하는 underlying server(예: Netty)에서는 message.retain()을 호출해야 할 수 있습니다. 그렇지 않으면 data buffer를 읽을 기회를 갖기 전에 release될 수 있습니다. 더 자세한 배경은 Data Buffers and Codecs을 참조하십시오.

다음 구현은 inbound와 outbound 스트림을 결합합니다:

1class ExampleHandler implements WebSocketHandler {
2
3	@Override
4	public Mono<Void> handle(WebSocketSession session) {
5
6		Flux<WebSocketMessage> output = session.receive()                // (1)
7				.doOnNext(message -> {
8					// ...
9				})
10				.concatMap(message -> {
11					// ...
12				})
13				.map(value -> session.textMessage("Echo " + value));     // (2)
14
15		return session.send(output);                                     // (3)
16	}
17}

1	inbound 메시지 스트림을 처리합니다.
2	outbound 메시지를 생성하여 통합된 플로우를 생성합니다.
3	수신을 계속하는 동안 완료되지 않는 Mono<Void>를 반환합니다.

1class ExampleHandler : WebSocketHandler {
2
3	override fun handle(session: WebSocketSession): Mono<Void> {
4
5		val output = session.receive()                    // (1)
6				.doOnNext {
7					// ...
8				}
9				.concatMap {
10					// ...
11				}
12				.map { session.textMessage("Echo $it") }   // (2)
13
14		return session.send(output)                        // (3)
15	}
16}

1	inbound 메시지 스트림을 처리합니다.
2	outbound 메시지를 생성하여 통합된 플로우를 생성합니다.
3	수신을 계속하는 동안 완료되지 않는 Mono<Void>를 반환합니다.

inbound와 outbound 스트림은 서로 독립적일 수 있으며, 다음 예제에서 보듯이 completion을 위해서만 결합될 수 있습니다:

1class ExampleHandler implements WebSocketHandler {
2
3	@Override
4	public Mono<Void> handle(WebSocketSession session) {
5
6		Mono<Void> input = session.receive()                            // (1)
7				.doOnNext(message -> {
8					// ...
9				})
10				.concatMap(message -> {
11					// ...
12				})
13				.then();
14
15		Flux<String> source = ... ;
16		Mono<Void> output = session.send(source.map(session::textMessage)); // (2)
17
18		return input.and(output);                                        // (3)
19	}
20}

1	inbound 메시지 스트림을 처리합니다.
2	outgoing 메시지를 전송합니다.
3	스트림을 결합하고, 스트림 중 하나가 끝나면 완료되는 Mono<Void>를 반환합니다.

1class ExampleHandler : WebSocketHandler {
2
3	override fun handle(session: WebSocketSession): Mono<Void> {
4
5		val input = session.receive()                                // (1)
6				.doOnNext {
7					// ...
8				}
9				.concatMap {
10					// ...
11				}
12				.then()
13
14		val source: Flux<String> = ...
15		val output = session.send(source.map(session::textMessage))  // (2)
16
17		return input.and(output)                                     // (3)
18	}
19}

1	inbound 메시지 스트림을 처리합니다.
2	outgoing 메시지를 전송합니다.
3	스트림을 결합하고, 스트림 중 하나가 끝나면 완료되는 Mono<Void>를 반환합니다.

DataBuffer

DataBuffer는 WebFlux에서 바이트 버퍼를 표현합니다. reference의 Spring Core 부분에는 Data Buffers and Codecs section에 이에 대한 더 많은 내용이 있습니다.

이해해야 할 핵심 포인트는 Netty와 같은 일부 server에서는 바이트 버퍼가 풀링되고 참조 카운팅되며, 메모리 누수를 방지하기 위해 consume될 때 release되어야 한다는 것입니다.

Netty에서 실행할 때, 애플리케이션이 input data buffer를 유지하려는 경우 DataBufferUtils.retain(dataBuffer)를 사용하여 버퍼가 release되지 않도록 해야 하며, 그 후 버퍼가 consume되면 DataBufferUtils.release(dataBuffer)를 사용해야 합니다.

Handshake

See equivalent in the Servlet stack

WebSocketHandlerAdapter는 WebSocketService에 위임합니다. 기본적으로 이는 HandshakeWebSocketService 인스턴스이며, WebSocket 요청에 대한 기본적인 체크를 수행한 다음 사용 중인 server에 대한 RequestUpgradeStrategy를 사용합니다. 현재 Reactor Netty, Tomcat, Jetty에 대한 내장 지원이 있습니다.

HandshakeWebSocketService는 sessionAttributePredicate 프로퍼티를 노출하며, 이를 통해 WebSession에서 속성을 추출하여 WebSocketSession의 속성에 삽입하기 위한 Predicate<String>을 설정할 수 있습니다.

Server Configuration

See equivalent in the Servlet stack

각 server에 대한 RequestUpgradeStrategy는 underlying WebSocket server 엔진에 특정한 설정을 노출합니다. WebFlux Java 설정을 사용할 때는 WebFlux Config의 해당 section에 표시된 대로 이러한 프로퍼티를 커스터마이즈할 수 있으며, 그렇지 않고 WebFlux 설정을 사용하지 않는 경우에는 아래와 같이 사용합니다:

1@Configuration
2class WebConfig {
3
4	@Bean
5	public WebSocketHandlerAdapter handlerAdapter() {
6		return new WebSocketHandlerAdapter(webSocketService());
7	}
8
9	@Bean
10	public WebSocketService webSocketService() {
11		TomcatRequestUpgradeStrategy strategy = new TomcatRequestUpgradeStrategy();
12		strategy.setMaxSessionIdleTimeout(0L);
13		return new HandshakeWebSocketService(strategy);
14	}
15}

1@Configuration
2class WebConfig {
3
4	@Bean
5	fun handlerAdapter() =
6			WebSocketHandlerAdapter(webSocketService())
7
8	@Bean
9	fun webSocketService(): WebSocketService {
10		val strategy = TomcatRequestUpgradeStrategy().apply {
11			setMaxSessionIdleTimeout(0L)
12		}
13		return HandshakeWebSocketService(strategy)
14	}
15}

사용 가능한 옵션을 확인하려면 server에 대한 업그레이드 전략을 확인하십시오. 현재 Tomcat과 Jetty만 그러한 옵션을 노출합니다.

CORS

See equivalent in the Servlet stack

CORS를 구성하고 WebSocket endpoint에 대한 접근을 제한하는 가장 쉬운 방법은 WebSocketHandler가 CorsConfigurationSource를 구현하도록 하고, 허용된 origin, 헤더 및 기타 세부 정보를 포함하는 CorsConfiguration을 반환하도록 하는 것입니다.

그렇게 할 수 없는 경우, SimpleUrlHandler에서 corsConfigurations 프로퍼티를 설정하여 URL 패턴별로 CORS 설정을 지정할 수도 있습니다. 둘 다 지정된 경우, CorsConfiguration의 combine 메서드를 사용하여 결합됩니다.

Client

Spring WebFlux는 Reactor Netty, Tomcat, Jetty, standard Java(즉, JSR-356)에 대한 구현을 가진 WebSocketClient 추상화를 제공합니다.

Tomcat client는 사실상 standard Java client의 확장으로, back pressure를 위해 메시지 수신을 suspend하기 위한 Tomcat-specific API를 활용할 수 있도록 WebSocketSession 처리를 위한 일부 추가 기능이 포함되어 있습니다.

WebSocket 세션을 시작하려면 client 인스턴스를 생성하고 그 execute 메서드를 사용할 수 있습니다:

1WebSocketClient client = new ReactorNettyWebSocketClient();
2
3URI url = new URI("ws://localhost:8080/path");
4client.execute(url, session ->
5		session.receive()
6				.doOnNext(System.out::println)
7				.then());

1val client = ReactorNettyWebSocketClient()
2
3val url = URI("ws://localhost:8080/path")
4client.execute(url) { session ->
5	session.receive()
6			.doOnNext(::println)
7			.then()
8}

Jetty와 같은 일부 client는 Lifecycle을 구현하며 사용하기 전에 stop 및 start되어야 합니다. 모든 client에는 underlying WebSocket client의 설정과 관련된 생성자 옵션이 있습니다.