목표: 실시간으로 보내지는 데이터를 받는다.
우선 원래 계획했던대로 Webflux Websocket을 사용하려고 한다.
제대로 들어가기 전에 인바운드 아웃바운드에 대한 설명을 적고 간다.
Inbound: 외부 네트워크에서 발생해 내부 네트워크로 들어오는 것
Outbound: 내부 네트워크에서 발생해 외부 네트워크로 나가는 것
WebSocketHandler
public class MyWebSocketHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
return null;
}
}WebSocketHandler의 handle은 WebSocketSession을 받아 세션의 애플리케이션 처리가 완료되었음을 알려주기 위해 Mono<Void>를 반환한다. 세션은 두 스트림으로 다뤄지는데 하나는 inbound이고 하나는 outbound이다. 스트림을 처리하는 두 가지 방법은 다음과 같다. (session이 호출할 수 있는 함수)
WebeSocketSession메서드Flux<WebSocketMessage> receive(): 인바운드 메세지에 접근하고 연결이 끊기면 종료한다.Mono<Void> send(Publisher<WebSocketMessage>): 발신 메세지의 소스를 가져오고, 메세지를 쓰고, 완료하면 Mono를 반환한다.
WebSocketHandler는 인바운드 및 아웃바운드 스트림을 통합된 플로우를 구성하고 Mono<Void>를 반환해야 한다.
플로우는 다음과 같을 때 종료된다.
- 인바운드나 아웃바운드 메세지 스트림이 완료될 때
- 인바운드 스트림은 완료되고 아웃바운드 스트림은 계속 실행될 때
- 선택한 지점에서
WebSocketSession의close메서드를 통해
가장 쉬운 방법은 인바운드 스트림을 핸들링하는 것이다.
public class MyWebSocketHandler implements WebSocketHandler {
@Override
public Mono<Void> handle(WebSocketSession session) {
return session.receive() // 인바운드 매세지 접근
.doOnNext(message -> {
// 메세지 다루기
})
.concatMap(message -> {
// 메세지 내용을 사용하는 nested asynchronous 연산
})
.then();// Mono<Void>를 반환
}
}
아웃바운드 인바운드 둘 다 핸들링
@Override
public Mono<Void> handle(WebSocketSession session) {
Mono<Void> input = session.receive()
.doOnNext(message -> { // 인바운드 메세지 핸들링
// ...
})
.concatMap(message -> {
// ...
})
.then();
Flux<String> source = ... ;
Mono<Void> output = session.send(source.map(session::textMessage)); // 아웃고잉 메세지 보내기
return Mono.zip(input, output).then(); // 스트림 복귀, Mono<void> 반환
}DataBuffer
여길 참고하라고 함. Netty 경우는 좀 특이한데 쓰지 않을 것이므로 패스
Handshake
WebSocketHandlerAdapter가 WebSocketService로 위임한다. HandshakeWebSocketService 인스턴스가 WebSocket 리퀘스트를 확인하고 RequestUpgradeStrategy를 사용한다.
HandshakeWebSocketService 는 sessionAttributePredicate속성을 보여준다고 한다.
Server Configuration
각 서버의 RequestUpgradeStrateg는 WebSocket의 특정 구성을 알려준다. Webflux Java 구성을 사용한다면 Webflux config처럼 사용할 수 있다. 사용하지 않으면 아래를 참고
@Configuration
class WebConfig {
@Bean
public WebSocketHandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter(webSocketService());
}
@Bean
public WebSocketService webSocketService() {
TomcatRequestUpgradeStrategy strategy = new TomcatRequestUpgradeStrategy();
strategy.setMaxSessionIdleTimeout(0L);
return new HandshakeWebSocketService(strategy);
}
}CORS
CORS를 구성하는 가장 쉬운 방법은 WebSocketHandler에 CorsConfigurationSource을 임플리멘트해 CorsConfiguration을 반환시키는 것임.
출처
