DispatcherHandler

DispatcherHandler

 

Spring WebFlux에서 WebFluxConfigurationSupport를 통해서 DispatcherHandler를 생성하고 이를 "webHandler"라는 이름의 Bean으로 등록한다는 것을 확인했습니다. 또한 HttpHandlerAutoConfiguration는 "webHandler" 라는 이름의 Bean을 찾아서 WebHttpHandlerBuilder에 조합하고, 최종적으로 HttpHandler Bean을 생성해 등록한다는 것도 확인했습니다.

 

이번 글에서는 DispatcherHandler에 대해 깊게 살펴보겠습니다.

 

 

Spring WebFlux는 Spring MVC와 마찬가지로 Front Controller 패턴을 따릅니다. 이때 WebFlux 쪽에서 Front Controller 역할을 수행하는 핵심 컴포넌트가 바로 DispatcherHandler 입니다.

public class DispatcherHandler implements WebHandler, PreFlightRequestHandler, ApplicationContextAware {
    @Nullable
    private List<HandlerMapping> handlerMappings;
    
    @Nullable
    private List<HandlerAdapter> handlerAdapters;
    
    @Nullable
    private List<HandlerResultHandler> resultHandlers;
    ...   
}

DispatcherHandler는 자신이 모든 일을 직접 처리하지 않고, 특정 타입의 Bean들을 찾아서 위임합니다.

 

빈 타입	역  할
HandlerMapping	요청을 어떤 핸들러가 처리할지 매핑
HandlerAdapter	매핑된 핸들러를 실제 호출하는 방법을 캡슐화
HandlerResultHandler	핸들러 실행 결과를 처리하고 응답을 마무리

 

 

HandlerMapping

 

ServerWebExchange를 보고 어떤 핸들러가 이 요청을 처리할지 결정합니다. 지원하는 핸들러가 없다면 Mono.empty()를 반환합니다.

public interface HandlerMapping {
    Mono<Object> getHandler(ServerWebExchange exchange);
}

대표적인 구현체로는 다음 3가지가 있습니다.

• RequestMappingHandlerMapping : @RequestMapping, @GetMapping 등 애노테이션 기반 컨트롤러
• RouterFunctionMapping : 함수형 라우팅(Functional Endpoints)
• SimpleUrlHandlerMapping : 단순 URL 패턴 매핑

SimpleUrlHandlerMapping에 매핑되는 WebHandler는 DispatcherHandler가 아닌 사용자 정의 Handler를 의미합니다.

 

 

HandlerAdapter

 

HandlerMapping이 반환환 핸들러 중 자신이 지원하는 타입의 핸들러를 판별하여 실제 실행을 담당하는 컴포넌트입니다. supports()로 처리 가능 여부를 판단하고, handle()을 통해 요청을 실행한 뒤 HandlerResult를 반환합니다.

public interface HandlerAdapter {
    boolean supports(Object handler);

    Mono<HandlerResult> handle(ServerWebExchange exchange, Object handler);
}

구분 방식은 HandlerMapping과 유사하지만, 차이점은 핸들러 실행 단계에서 WebHandler와 WebSocketHandler를 분리하여 각각에 맞는 처리 방식으로 실행한다는 점입니다.

 

 

HandlerResultHandler

 

HandlerResultHandler는 HandlerAdapter가 반환한 HandlerResult 중 자신이 처리할 수 있는 결과 타입을 판별하여 실제 응답 작성까지 담당하는 컴포넌트입니다. supports()로 처리 가능 여부를 판단하고, handleResult()를 통해 결과를 HTTP Response로 변환하여 클라이언트에게 반환합니다.

public interface HandlerResultHandler {
    boolean supports(HandlerResult result);

    Mono<Void> handleResult(ServerWebExchange exchange, HandlerResult result);
}

대표적인 구현체로는 다음과 같이 4가지가 있습니다.

• ResponseEntityResultHandler : ResponseEntity, HttpEntity 반환 처리
• ResponseBodyResultHander : @ResponseBody / @RestController 메서드 반환값 처리
• ViewResolutionResultHandler : String(뷰 이름), Rendering, Model, Map, View
• ServerResponseResultHandler : ServerResponse 처리 (함수형 엔드포인트)

 

 

Handler Model

 

WebFlux에서 모든 요청은 DispatcherHandler를 통해 진입하고, 이후 HandlerMapping·HandlerAdapter·HandlerResultHandler 단계를 거쳐 최종적으로 응답이 만들어집니다. 이 과정에서 실제 요청을 처리하는 방식(Handler Model)은 하나가 아닌 여러 형태로 존재하며, 이번 글에서는 그 중 대표적인 네 가지 방식에 대해 살펴보겠습니다.

• Functional Endpoints
• Annotated Controllers
• SSE(Server-Sent Events)
• WebSocket

 

Functional Endpoints

 

Functional Endpoints는 WebFlux에서 제공하는 함수형 라우팅 기반 요청 처리 방식입니다.

 

HandlerFunction

HandlerFunction은 요청을 처리하고 응답을 생성하는 함수입니다. Spring MVC에서 Controller 메서드에 해당하는 역할이며, ServerRequest → ServerResponse 형태를 가집니다.

@FunctionalInterface
public interface HandlerFunction<T extends ServerResponse> {
    Mono<T> handle(ServerRequest request);
}

 

ServerRequest, ServerResponse

ServerRequest는 Functional Endpoint에서 요청 정보를 다루기 위해 제공되는 객체로, HandlerFunction의 인자로 전달됩니다.

ServerWebExchange를 내부적으로 포함하고 있으며, Body 파싱을 위한 messageReaders 제공합니다.

public interface ServerRequest {
    List<HttpMessageReader<?>> messageReaders();
    <T> Mono<T> bodyToMono(Class<? extends T> elementClass);
    <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference);
    <T> Flux<T> bodyToFlux(Class<? extends T> elementClass);
    <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference);
    Map<String, String> pathVariables();
    ServerWebExchange exchange();
}

 

ServerResponse는 HandlerFunction이 반환하는 응답 객체이며, Builder 기반으로 작성됩니다.

public interface ServerResponse {
    static BodyBuilder status(HttpStatus status) {...}
    static BodyBuilder status(int status) {...}
    static BodyBuilder ok() {...}
    static BodyBuilder created(URI location) {...}
    static BodyBuilder accepted() {...}
    static HeadersBuilder<?> noContent() {...}
    static BodyBuilder seeOther(URI location) {...}
    static BodyBuilder temporaryRedirect(URI location) {...}
    static BodyBuilder permanentRedirect(URI location) {...}
    static BodyBuilder badRequest() {...}
    static HeadersBuilder<?> notFound() {...}
}

public interface BodyBuilder extends HeadersBuilder<BodyBuilder> {
    BodyBuilder contentLength(long contentLength);

    BodyBuilder contentType(MediaType contentType);

    BodyBuilder hint(String key, Object value);

    Mono<ServerResponse> bodyValue(Object body);

    Mono<ServerResponse> body(Object producer, Class<?> elementClass);

    Mono<ServerResponse> render(String name, Map<String, ?> model);
}

public interface HeadersBuilder<B extends HeadersBuilder<B>> {
    B header(String headerName, String... headerValues);

    B headers(Consumer<HttpHeaders> headersConsumer);

    B cookie(ResponseCookie cookie);

    B cookies(Consumer<MultiValueMap<String, ResponseCookie>> cookiesConsumer);

    B allow(HttpMethod... allowedMethods);

    B lastModified(ZonedDateTime lastModified);

    B location(URI location);

    B cacheControl(CacheControl cacheControl);

    Mono<ServerResponse> build();
}

 

RouterFunciton

RouterFunction은 URL 경로를 HandlerFunction으로 라우팅하는 역할을 담당합니다. Controller에서 @GetMapping("/users")와 같은 매핑 애노테이션 역할을 대신합니다. 

@FunctionalInterface
public interface RouterFunction<T extends ServerResponse> {
    Mono<HandlerFunction<T>> route(ServerRequest request);
}

 

일반적으로 RouterFunction을 직접 구현하기보다는 RouterFunctions 유틸리티 클래스의 메서드를 사용하여 생성하는 방식이 권장됩니다.

RouterFunction<ServerResponse> route = RouterFunctions.route()
        .GET("/hello-world", 
                request -> ServerResponse.ok().bodyValue("Hello World")
        ).build();

 

RequestPredicate

라우팅 시 HTTP method 외에도 추가 조건을 두고 싶을 때 RequestPredicate를 함께 사용할 수 있습니다. 이를 활용하면 헤더 또는 MediaType 기반 라우팅 등 다양한 제약 조건을 표현할 수 있습니다.

RouterFunction<ServerResponse> route = RouterFunctions.route()
        .GET("/hello-world",
                accept(MediaType.TEXT_PLAIN),
                request -> ServerResponse.ok().bodyValue("Hello World"))
        .build();

 

HandlerFunction을 람다식으로 작성하지 않고 별도의 핸들러 클래스에 작성한 뒤 메서드 참조와 SAM Conversion을 이용하면 다음과 같이 작성할 수 있습니다.

RouterFunction<ServerResponse> route = route()
	.path("/person", builder -> builder
		.GET("/{id}", accept(APPLICATION_JSON), handler::getPerson)
		.GET(accept(APPLICATION_JSON), handler::listPeople)
		.POST(handler::createPerson))
	.build();

 

 

작성한 RouterFunction을 HTTP 서버에서 실행하는 방법은 다음 두 가지 방법이 있습니다.

1. RouterFunctions.toHttpHandler(router)로 변환한 뒤 ReactorHttpHandlerAdapter로 래핑하여 직접 띄우기
2. RouterFunction을 @Bean으로 등록하여 HandlerMapping에 자동으로 매핑하기

 

 

Annotated Controllers

 

Annotated Controller는 Spring WebFlux에서 사용되는 애노테이션 기반의 요청 처리 모델로, 기존 Spring MVC 개발 방식과 가장 유사하게 @Controller, @RestController, @RequestMapping 등의 애노테이션을 활용해 HTTP 요청을 매핑하고, 메서드의 반환값을 통해 응답을 생성합니다.

@Controller, @RestController

Spring Framework 6 이전에는 @Component + @ReqeustMapping 조합으로도 컨트롤러로 인식될 수 있었지만, 현재는 반드시 @Controller  또는 @RestController를 사용해야 컨트롤러로 인식됩니다.

 

Handler Methods

@RequestMapping 기반의 핸들러 메서드는 매우 유연한 메서드 시그니처를 가질 수 있으며,
컨트롤러 메서드의 파라미터(Arguments) 와 반환 타입(Return Values) 으로 사용할 수 있는 타입도 다양하게 지원됩니다.

Arguments 중 @RequestBody처럼 요청 본문(body)을 읽어야 하는 경우에는 Reactive 타입(Mono, Flux)을 지원하며, PathVariable, RequestParam, Header, Cookie 같은 단순 매핑 파라미터는 Servlet Stack과 거의 동일한 방식으로 사용할 수 있습니다.

 

대표적인 Handler Method Arguments는 다음과 같습니다.

유형	예시
경로/쿼리 기반 값	@PathVariable, @RequestParam
헤더/쿠키 기반 값	@RequestHeader, @CookieValue
요청/응답 객체	ServerWebExchange, ServerHttpRequest, ServerHttpResponse
세션/보안 정보	WebSession, Principal, Authentication
Validation	@Valid, @Validated, Errors
Reactive Body 처리	Mono<T>, Flux<T>

 

Return Values 역시 Servlet Stack과 거의 동일하나 ModelAndView 대신 Rendering을 지원한다는 차이가 있습니다.

 

대표적은 Handler Method Return Values는 다음과 같습니다.

반환 타입	설명
Mono<T>	단일 값 비동기 반환
Flux<T>	다중 스트림 반환 (SSE/Streaming 처리에 적합)
T (일반 객체)	즉시 값 반환하면 내부적으로 Mono로 처리됨
Mono<Void> / Void	바디 없이 성공 응답 (200/204 응답)
ResponseEntity<T>	상태 코드 / 헤더 / 바디 제어 가능
Mono<ResponseEntity<T>>	ResponseEntity + Reactive 조합
HttpHeaders	헤더만 포함된 응답 (body 없음)
Rendering	템플릿 렌더링을 위한 반환 타입 (View 기반)
ServerResponse	WebFlux.fn(Functional style) 반환 타입
Publisher<T>	Reactive Streams 일반 형태

 

다은은 JSON REST 컨트롤러의 예시입니다.

@RestController
@RequestMapping("/api")
public class UserController {

    @GetMapping("/users")
    public Flux<User> getUsers() {
        return Flux.just(
                new User("jin", 28),
                new User("hana", 26)
        );
    }
    
    @PostMapping("/user")
    public Mono<User> createUser(@RequestBody Mono<User> userMono) {
        return userMono
                .map(user -> new User(user.name(), user.age()));
    }
}

 

 

SSE (Server-Sent Events)

 

SSE(Server-Sent Events)는 서버가 클라이언트에게 지속적으로 데이터를 Push(stream) 할 수 있는 방식으로, HTTP 기반 Streaming 기술입니다. 클라이언트는 HTTP 연결을 유지한 채, 서버가 보내는 실시간 이벤트를 계속 수신합니다.

HTTP Streaming

HTTP Streaming(HTTP 스트리밍)은 클라이언트가 한 번 HTTP 요청을 보내면 서버가 응답을 즉시 끝내지 않고 연결을 유지한 채 데이터가 생기는 대로 전송하는 방식입니다. 즉, 응답을 한 번에 완성해 보내는 것이 아니라, chunk 단위로 계속 흘려보낼 수 있습니다.

방식	특징
Chunked Transfer-Encoding	Content-Length를 모르므로 Transfer-Encoding: chunked 를 사용하여 chunk 단위로 전송
EOF 기반 스트리밍	Connection: close 를 통해 서버가 종료할 때까지 데이터를 계속 받음

HTTP Streaming은 HTTP/1.1 이상에서만 사용 가능합니다.

 

SSE : 전송 방식 및 데이터 포맷

SSE는 보통 Chunked Transfer-Encoding 기반으로 작동합니다.

서버는 데이터를 chunk 단위로 여러 줄로 구성된 문자열을 전달하며, newline을 기준으로 이벤트를 구분합니다.

그리고 데이터 포맷의 경우 test/event-stream 을 사용하며, 각 메시지는 text 기반에 <field>: <value> 형식태로 구성됩니다.

 

다음은 일반적인 표준 응답과 SSE 응답 예시입니다.

 

다음은 SSE에서 사용되는 대표 필드들입니다.

필드	역할
id	이벤트 고유 ID (재연결 시 Last-Event-ID를 전달하면 이어받을 수 있습니다.)
data	실제 메시지(payload)로 Multi-line data도 가능
event	사용자 정의 이벤트 타입
retry	재연결(wait) 대기 시간(ms) 지정
comment	주석 또는 heartbeat 용으로 filed 부분이 비어 있음 (:로 시작)

 

ServerSentEventHttpMessageWriter

RequestMappingHandlerMapping이 @RequestMapping의 정보들(path, produces 등)을 읽어서 매칭되는 핸들러를 찾고, 그 값들을 exchange attribute에 저장합니다. 요청을 처리하고 핸들러에서 리턴한 값을 ResponseBodyResultHandler가 받아서 writeBody()를 호출하면 exchange에서 bestMediaType을 읽어와 알맞은 HttpMessageWriter를 선택합니다. ServerSentEvent의 경우 ServerSentEventHttpMessageWriter가 처리하게 됩니다.

protected Mono<Void> writeBody(@Nullable Object body, MethodParameter bodyParameter,
       @Nullable MethodParameter actualParam, ServerWebExchange exchange) {
       
    ...
    MediaType bestMediaType;
    bestMediaType = selectMediaType(exchange, () -> getMediaTypesFor(elementType));
    ...
    if (bestMediaType != null) {
        for (HttpMessageWriter<?> writer : getMessageWriters()) {
            if (writer.canWrite(actualElementType, bestMediaType)) {
                return writer.write((Publisher) publisher, actualType, elementType,
                        bestMediaType, exchange.getRequest(), exchange.getResponse(),
                        Hints.from(Hints.LOG_PREFIX_HINT, logPrefix));
            }
        }
    }
    ...
}

 

ServerSentEventHttpMessageWriter는 핸들러에서 반환된 객체를 SeverSentEvent 형태로 encode하여 응답 바디에 write 합니다.

public class ServerSentEventHttpMessageWriter implements HttpMessageWriter<Object> {

    @Override
    public Mono<Void> write(Publisher<?> input, ResolvableType elementType, @Nullable MediaType mediaType,
            ReactiveHttpOutputMessage message, Map<String, Object> hints) {
        ...
        return message.writeAndFlushWith(encode(input, elementType, mediaType, bufferFactory, hints));
    }

    private Flux<Publisher<DataBuffer>> encode(Publisher<?> input, ResolvableType elementType,
                MediaType mediaType, DataBufferFactory factory, Map<String, Object> hints) {

        return Flux.from(input).map(element -> {
            ServerSentEvent<?> sse = (element instanceof ServerSentEvent<?> serverSentEvent ?
                    serverSentEvent : ServerSentEvent.builder().data(element).build());
            
            ...
            return result.doOnDiscard(DataBuffer.class, DataBufferUtils::release);
        });
    }

    ...

}

 

위 encode() 메서드를 보면, 반환 값이 ServerSentEvent 타입의 인스턴스인지 확인 후 맞으면 그대로 사용하고, 아니라면 ServerSentEvent.builder() 를 사용해 래핑합니다.

 

다음은 SSE 컨트롤러의 예시입니다.

@RestController
public class SseController {

    @ResponseBody
    @GetMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> sse(
            @RequestHeader(name = "Last-Event-ID", required = false, defaultValue = "0")
            Long lastEventId
    ) {
        return Flux.range(0, 5)
                .delayElements(Duration.ofMillis(100))
                .map(i -> ServerSentEvent.<String>builder()
                        .event("add")                            
                        .id(String.valueOf(i + lastEventId + 1)) 
                        .data("data-" + i)                       
                        .comment("comment-" + i)                 
                        .build()
                );
    }
}

 

 

WebSocket

 

WebSocket은 클라이언트와 서버 간의 실시간 양방향 통신을 가능하게 하는 프로토콜이며, HTTP 기반이 아닌 독립적인 애플리케이션 계층(OSI 7계층) 프로토콜이며, 실제 데이터 전송은 전송 계층(OSI 4계층, TCP) 위에서 이루어집니다. HTTP와 달리 지속 연결을 유지하기 때문에 오버헤드가 적다는 장점이 있습니다.

WebSocket은 @Controller 처럼 반환 결과를 HTTP 응답으로 렌더링하지 않고
WebSocketHandler를 통해 직접 메시지를 송수신하는 구조이므로 HandlerResultHandler가 사용되지 않습니다.

 

WebSocket Upgrade

초기 연결은 HTTP 요청 형태로 시작하여, 서버가 WebSocket 연결을 허용하면 상태 코드 101 Switching Protocols를 응답하며 HTTP 연결에서 WebSocket 프로토콜로 변경됩니다.

 

SimpleUrlHandlerMapping

SimpleUrlHandlerMapping 은 Spring WebFlux/Spring MVC 모두에서 사용되는 HandlerMapping 구현체 중 하나로,
URL 패턴(Path) → Handler 매핑 정보를 명시적으로 등록해두고 사용하는 방식의 매핑 클래스입니다.

즉, @RequestMapping 기반 자동 스캔이 아닌, 수동으로 URL 매핑을 등록하는 방식입니다.

 

SimpleUrlHandlerMapping 를 보면 내부적으로 URL과 연결되는 Handler를 저장하는 urlMap을 갖고 있습니다.

public class SimpleUrlHandlerMapping extends AbstractUrlHandlerMapping {

    private final Map<String, Object> urlMap = new LinkedHashMap<>();
    ...   
}

 

WebSocket 서버를 구성하기 위해서는 WebSocketHandler를 구현하고, 이를 특정 경로와 함께 SimpleUrlHandlerMapping에 등록해주면 됩니다. 이렇게 하면 지정된 URL로 WebSocket 요청이 들어올 경우 해당 Handler가 실행됩니다.

public class MyWebSocketHandler implements WebSocketHandler {

	@Override
	public Mono<Void> handle(WebSocketSession session) {
		// ...
	}
}

@Configuration
class WebConfig {

	@Bean
	public HandlerMapping handlerMapping() {
		Map<String, WebSocketHandler> map = new HashMap<>();
		map.put("/path", new MyWebSocketHandler());
		int order = -1; // before annotated controllers

		return new SimpleUrlHandlerMapping(map, order);
	}
}

 

WebSocketHandler

WebSocket 요청이 WebSocket으로 업그레이드된 이후 클라이언트와 주고받는 메시지를 처리하는 핵심 핸들러 인터페이스입니다.

인자로 WebSocketSession을 받고 Mono<Void>를 반환합니다.

public interface WebSocketHandler {

    /**
     * Invoked when a new WebSocket connection is established, and allows
     * handling of the session.
     *
     * <p>See the class-level doc and the reference manual for more details and
     * examples of how to handle the session.
     * @param session the session to handle
     * @return indicates when application handling of the session is complete,
     * which should reflect the completion of the inbound message stream
     * (i.e. connection closing) and possibly the completion of the outbound
     * message stream and the writing of messages
     */
    Mono<Void> handle(WebSocketSession session);

}

 

WebSocketSession

WebSocketSession은 WebSocket 연결이 수립되면 생성되는 실 연결 객체(Session)입니다.

public interface WebSocketSession {

    /**
     * Return the id for the session.
     */
    String getId();

    /**
     * Return information from the handshake request.
     */
    HandshakeInfo getHandshakeInfo();

    /**
     * Return a {@code DataBuffer} Factory to create message payloads.
     */
    DataBufferFactory bufferFactory();

    /**
     * Return the map with attributes associated with the WebSocket session.
     */
    Map<String, Object> getAttributes();

    /**
     * Provides access to the stream of inbound messages.
     */
    Flux<WebSocketMessage> receive();

    /**
     * Give a source of outgoing messages, write the messages and return a
     * {@code Mono<Void>} that completes when the source completes and writing
     * is done.
     */
    Mono<Void> send(Publisher<WebSocketMessage> messages);

    /**
     * Whether the underlying connection is open.
     */
    boolean isOpen();

    /**
     * Close the WebSocket session with {@link CloseStatus#NORMAL}.
     */
    default Mono<Void> close() {
       return close(CloseStatus.NORMAL);
    }

    /**
     * Close the WebSocket session with the given status.
     */
    Mono<Void> close(CloseStatus status);

    
    
    // WebSocketMessage factory methods

    /**
     * Factory method to create a text {@link WebSocketMessage} using the
     * {@link #bufferFactory()} for the session.
     */
    WebSocketMessage textMessage(String payload);

    /**
     * Factory method to create a binary WebSocketMessage using the
     * {@link #bufferFactory()} for the session.
     */
    WebSocketMessage binaryMessage(Function<DataBufferFactory, DataBuffer> payloadFactory);

    /**
     * Factory method to create a ping WebSocketMessage using the
     * {@link #bufferFactory()} for the session.
     */
    WebSocketMessage pingMessage(Function<DataBufferFactory, DataBuffer> payloadFactory);

    /**
     * Factory method to create a pong WebSocketMessage using the
     * {@link #bufferFactory()} for the session.
     */
    WebSocketMessage pongMessage(Function<DataBufferFactory, DataBuffer> payloadFactory);

}

 

이 중에서 마지막 4개의 메서드는 WebSocketMessage를 만드는 팩토리 메서드입니다.

public class WebSocketMessage {

    private final Type type;

    private final DataBuffer payload;

    /**
     * Return the message type (text, binary, etc).
     */
    public Type getType() {
       return this.type;
    }

    /**
     * Return the message payload.
     */
    public DataBuffer getPayload() {
       return this.payload;
    }

    /**
     * A variant of {@link #getPayloadAsText(Charset)} that uses {@code UTF-8}
     * for decoding the raw content to text.
     */
    public String getPayloadAsText() {
       return getPayloadAsText(StandardCharsets.UTF_8);
    }

    /**
     * A shortcut for decoding the raw content of the message to text with the
     * given character encoding. This is useful for text WebSocket messages, or
     * otherwise when the payload is expected to contain text.
     */
    public String getPayloadAsText(Charset charset) {
       return this.payload.toString(charset);
    }

    /**
     * WebSocket message types.
     */
    public enum Type {
       TEXT,
       BINARY,
       PING,
       PONG
    }

}

 

다음은 WebSocketHandler의 예시입니다.

class ExampleHandler implements WebSocketHandler {

	@Override
	public Mono<Void> handle(WebSocketSession session) {

		Flux<WebSocketMessage> output = session.receive()				
				.doOnNext(message -> {
					// ...
				})
				.concatMap(message -> {
					// ...
				})
				.map(value -> session.textMessage("Echo " + value));	

		return session.send(output);									
	}
}