From 3b987c263c17fd0a903c03425f8f9384294d7dcc Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Tue, 20 Dec 2016 12:21:30 -0500 Subject: [PATCH] Sub-protocol negotiation for reactive WebSocket support Issue: SPR-14527 --- .../web/reactive/socket/HandshakeInfo.java | 28 +++++-- .../web/reactive/socket/WebSocketHandler.java | 6 +- .../client/ReactorNettyWebSocketClient.java | 43 ++++++++--- .../socket/client/RxNettyWebSocketClient.java | 69 +++++++++++++---- .../socket/client/WebSocketClientSupport.java | 40 ++++++++++ .../socket/server/RequestUpgradeStrategy.java | 7 +- .../support/HandshakeWebSocketService.java | 22 +++++- .../upgrade/DefaultServerEndpointConfig.java | 8 +- .../upgrade/JettyRequestUpgradeStrategy.java | 47 ++++++++++-- .../ReactorNettyRequestUpgradeStrategy.java | 29 +++----- .../RxNettyRequestUpgradeStrategy.java | 40 +++++----- .../upgrade/TomcatRequestUpgradeStrategy.java | 16 ++-- .../UndertowRequestUpgradeStrategy.java | 26 +++++-- .../server/WebSocketIntegrationTests.java | 74 ++++++++++++++++++- 14 files changed, 365 insertions(+), 90 deletions(-) create mode 100644 spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java index 8aa4ffefdc..8470067494 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java @@ -17,6 +17,7 @@ package org.springframework.web.reactive.socket; import java.net.URI; import java.security.Principal; +import java.util.Optional; import reactor.core.publisher.Mono; @@ -31,6 +32,7 @@ import org.springframework.util.Assert; * @since 5.0 * @see WebSocketSession#getHandshakeInfo() */ +@SuppressWarnings("OptionalUsedAsFieldOrParameterType") public class HandshakeInfo { private final URI uri; @@ -39,14 +41,20 @@ public class HandshakeInfo { private final Mono principalMono; + private final Optional protocol; + + + public HandshakeInfo(URI uri, HttpHeaders headers, Mono principal, + Optional subProtocol) { - public HandshakeInfo(URI uri, HttpHeaders headers, Mono principalMono) { Assert.notNull(uri, "URI is required."); Assert.notNull(headers, "HttpHeaders are required."); - Assert.notNull(principalMono, "Principal is required."); + Assert.notNull(principal, "Principal is required."); + Assert.notNull(subProtocol, "Sub-protocol is required."); this.uri = uri; this.headers = headers; - this.principalMono = principalMono; + this.principalMono = principal; + this.protocol = subProtocol; } @@ -58,19 +66,29 @@ public class HandshakeInfo { } /** - * Return the headers from the handshake HTTP request. + * Return the handshake HTTP headers. Those are the request headers for a + * server session and the response headers for a client session. */ public HttpHeaders getHeaders() { return this.headers; } /** - * Return the principal associated with the handshake HTTP request, if any. + * Return the principal associated with the handshake HTTP request. */ public Mono getPrincipal() { return this.principalMono; } + /** + * The sub-protocol negotiated at handshake time. + * @see + * https://tools.ietf.org/html/rfc6455#section-1.9 + */ + public Optional getSubProtocol() { + return this.protocol; + } + @Override public String toString() { diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/WebSocketHandler.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/WebSocketHandler.java index 587973e60f..85c7448bbf 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/WebSocketHandler.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/WebSocketHandler.java @@ -30,10 +30,10 @@ public interface WebSocketHandler { /** * Return the list of sub-protocols supported by this handler. - *

By default an empty list is returned. + *

By default an empty array is returned. */ - default List getSubProtocols() { - return Collections.emptyList(); + default String[] getSubProtocols() { + return new String[0]; } /** diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java index b96ba63025..f78d5e2e47 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java @@ -16,6 +16,7 @@ package org.springframework.web.reactive.socket.client; import java.net.URI; +import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; @@ -25,6 +26,7 @@ import reactor.ipc.netty.NettyOutbound; import reactor.ipc.netty.http.client.HttpClient; import reactor.ipc.netty.http.client.HttpClientOptions; import reactor.ipc.netty.http.client.HttpClientRequest; +import reactor.ipc.netty.http.client.HttpClientResponse; import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.HttpHeaders; @@ -39,7 +41,7 @@ import org.springframework.web.reactive.socket.adapter.ReactorNettyWebSocketSess * @author Rossen Stoyanchev * @since 5.0 */ -public class ReactorNettyWebSocketClient implements WebSocketClient { +public class ReactorNettyWebSocketClient extends WebSocketClientSupport implements WebSocketClient { private final HttpClient httpClient; @@ -61,30 +63,47 @@ public class ReactorNettyWebSocketClient implements WebSocketClient { @Override public Mono execute(URI url, HttpHeaders headers, WebSocketHandler handler) { - // We have to store the NettyOutbound fow now.. - // The alternative HttpClientResponse#receiveWebSocket does not work at present + // TODO: https://github.com/reactor/reactor-netty/issues/19 AtomicReference outboundRef = new AtomicReference<>(); + String[] protocols = getSubProtocols(headers, handler); + // TODO: https://github.com/reactor/reactor-netty/issues/20 + return this.httpClient .get(url.toString(), request -> { - addHeaders(request, headers); + addRequestHeaders(request, headers); NettyOutbound outbound = request.sendWebsocket(); outboundRef.set(outbound); return outbound; }) - .then(inbound -> { - ByteBufAllocator allocator = inbound.channel().alloc(); + .then(in -> { + HttpHeaders responseHeaders = getResponseHeaders(in); + String protocol = responseHeaders.getFirst(SEC_WEBSOCKET_PROTOCOL); + HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), + Optional.ofNullable(protocol)); + + ByteBufAllocator allocator = in.channel().alloc(); NettyDataBufferFactory factory = new NettyDataBufferFactory(allocator); - NettyOutbound outbound = outboundRef.get(); - HandshakeInfo info = new HandshakeInfo(url, headers, Mono.empty()); - WebSocketSession session = new ReactorNettyWebSocketSession(inbound, outbound, info, factory); + + NettyOutbound out = outboundRef.get(); + WebSocketSession session = new ReactorNettyWebSocketSession(in, out, info, factory); return handler.handle(session); }); } - private void addHeaders(HttpClientRequest request, HttpHeaders headers) { - headers.entrySet().stream() - .forEach(e -> request.requestHeaders().set(e.getKey(), e.getValue())); + private void addRequestHeaders(HttpClientRequest request, HttpHeaders headers) { + headers.keySet().stream() + .forEach(key -> headers.get(key).stream() + .forEach(value -> request.addHeader(key, value))); + } + + private HttpHeaders getResponseHeaders(HttpClientResponse response) { + HttpHeaders headers = new HttpHeaders(); + response.responseHeaders().forEach(entry -> { + String name = entry.getKey(); + headers.put(name, response.responseHeaders().getAll(name)); + }); + return headers; } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/RxNettyWebSocketClient.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/RxNettyWebSocketClient.java index 0757ae29c8..cc28a88f86 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/RxNettyWebSocketClient.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/RxNettyWebSocketClient.java @@ -17,6 +17,12 @@ package org.springframework.web.reactive.socket.client; import java.net.URI; import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; import java.util.function.Function; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; @@ -26,6 +32,7 @@ import io.netty.buffer.ByteBufAllocator; import io.reactivex.netty.protocol.http.client.HttpClient; import io.reactivex.netty.protocol.http.ws.WebSocketConnection; import io.reactivex.netty.protocol.http.ws.client.WebSocketRequest; +import io.reactivex.netty.protocol.http.ws.client.WebSocketResponse; import reactor.core.publisher.Mono; import reactor.util.function.Tuples; import rx.Observable; @@ -33,6 +40,7 @@ import rx.RxReactiveStreams; import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.HttpHeaders; +import org.springframework.util.ObjectUtils; import org.springframework.web.reactive.socket.HandshakeInfo; import org.springframework.web.reactive.socket.WebSocketHandler; import org.springframework.web.reactive.socket.WebSocketSession; @@ -44,7 +52,7 @@ import org.springframework.web.reactive.socket.adapter.RxNettyWebSocketSession; * @author Rossen Stoyanchev * @since 5.0 */ -public class RxNettyWebSocketClient implements WebSocketClient { +public class RxNettyWebSocketClient extends WebSocketClientSupport implements WebSocketClient { private final Function> httpClientFactory; @@ -91,32 +99,65 @@ public class RxNettyWebSocketClient implements WebSocketClient { @Override public Mono execute(URI url, HttpHeaders headers, WebSocketHandler handler) { - HandshakeInfo info = new HandshakeInfo(url, headers, Mono.empty()); - Observable completion = connectInternal(handler, info); + Observable completion = connectInternal(url, headers, handler); return Mono.from(RxReactiveStreams.toPublisher(completion)); } - private Observable connectInternal(WebSocketHandler handler, HandshakeInfo info) { - return createWebSocketRequest(info.getUri()) + private Observable connectInternal(URI url, HttpHeaders headers, WebSocketHandler handler) { + return createRequest(url, headers, handler) .flatMap(response -> { - ByteBufAllocator allocator = response.unsafeNettyChannel().alloc(); - NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(allocator); Observable conn = response.getWebSocketConnection(); - return Observable.zip(conn, Observable.just(bufferFactory), Tuples::of); + return Observable.zip(Observable.just(response), conn, Tuples::of); }) .flatMap(tuple -> { - WebSocketConnection conn = tuple.getT1(); - NettyDataBufferFactory bufferFactory = tuple.getT2(); - WebSocketSession session = new RxNettyWebSocketSession(conn, info, bufferFactory); + WebSocketResponse response = tuple.getT1(); + HttpHeaders responseHeaders = getResponseHeaders(response); + Optional protocol = Optional.ofNullable(response.getAcceptedSubProtocol()); + HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol); + + ByteBufAllocator allocator = response.unsafeNettyChannel().alloc(); + NettyDataBufferFactory factory = new NettyDataBufferFactory(allocator); + + WebSocketConnection conn = tuple.getT2(); + WebSocketSession session = new RxNettyWebSocketSession(conn, info, factory); return RxReactiveStreams.toObservable(handler.handle(session)); }); } - private WebSocketRequest createWebSocketRequest(URI url) { + private WebSocketRequest createRequest(URI url, HttpHeaders headers, WebSocketHandler handler) { + String query = url.getRawQuery(); - return this.httpClientFactory.apply(url) - .createGet(url.getRawPath() + (query != null ? "?" + query : "")) + String requestUrl = url.getRawPath() + (query != null ? "?" + query : ""); + + WebSocketRequest request = this.httpClientFactory.apply(url) + .createGet(requestUrl) + .setHeaders(toObjectValueMap(headers)) .requestWebSocketUpgrade(); + + String[] protocols = getSubProtocols(headers, handler); + if (!ObjectUtils.isEmpty(protocols)) { + request = request.requestSubProtocols(protocols); + } + + return request; + } + + private Map> toObjectValueMap(HttpHeaders headers) { + if (headers.isEmpty()) { + return Collections.emptyMap(); + } + Map> map = new HashMap<>(headers.size()); + headers.keySet().stream().forEach(key -> map.put(key, new ArrayList<>(headers.get(key)))); + return map; + } + + private HttpHeaders getResponseHeaders(WebSocketResponse response) { + HttpHeaders headers = new HttpHeaders(); + response.headerIterator().forEachRemaining(entry -> { + String name = entry.getKey().toString(); + headers.put(name, response.getAllHeaderValues(name)); + }); + return headers; } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java new file mode 100644 index 0000000000..dbdcf5e382 --- /dev/null +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.web.reactive.socket.client; + +import org.springframework.http.HttpHeaders; +import org.springframework.util.StringUtils; +import org.springframework.web.reactive.socket.WebSocketHandler; + +/** + * Base class for {@link WebSocketClient} implementations. + * + * @author Rossen Stoyanchev + * @since 5.0 + */ +public class WebSocketClientSupport { + + protected static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol"; + + + protected String[] getSubProtocols(HttpHeaders headers, WebSocketHandler handler) { + String value = headers.getFirst(SEC_WEBSOCKET_PROTOCOL); + return (value != null ? + StringUtils.commaDelimitedListToStringArray(value) : + handler.getSubProtocols()); + } + +} \ No newline at end of file diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/RequestUpgradeStrategy.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/RequestUpgradeStrategy.java index 0355befff7..1e0fbd513f 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/RequestUpgradeStrategy.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/RequestUpgradeStrategy.java @@ -15,6 +15,8 @@ */ package org.springframework.web.reactive.socket.server; +import java.util.Optional; + import reactor.core.publisher.Mono; import org.springframework.http.server.reactive.ServerHttpRequest; @@ -40,9 +42,12 @@ public interface RequestUpgradeStrategy { * Upgrade to a WebSocket session and handle it with the given handler. * @param exchange the current exchange * @param webSocketHandler handler for the WebSocket session + * @param subProtocol the selected sub-protocol got the handler * @return completion {@code Mono} to indicate the outcome of the * WebSocket session handling. */ - Mono upgrade(ServerWebExchange exchange, WebSocketHandler webSocketHandler); + @SuppressWarnings("OptionalUsedAsFieldOrParameterType") + Mono upgrade(ServerWebExchange exchange, WebSocketHandler webSocketHandler, + Optional subProtocol); } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java index 6db4bdc65c..5aa4ab355d 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java @@ -15,8 +15,10 @@ */ package org.springframework.web.reactive.socket.server.support; +import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Optional; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -30,6 +32,7 @@ import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; import org.springframework.web.reactive.socket.WebSocketHandler; import org.springframework.web.reactive.socket.server.RequestUpgradeStrategy; import org.springframework.web.reactive.socket.server.WebSocketService; @@ -49,6 +52,8 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle { private static final String SEC_WEBSOCKET_KEY = "Sec-WebSocket-Key"; + private static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol"; + private static final boolean tomcatPresent = ClassUtils.isPresent( "org.apache.tomcat.websocket.server.WsHttpUpgradeHandler", @@ -171,7 +176,7 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle { @Override - public Mono handleRequest(ServerWebExchange exchange, WebSocketHandler webSocketHandler) { + public Mono handleRequest(ServerWebExchange exchange, WebSocketHandler handler) { ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); @@ -190,7 +195,9 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle { return response.setComplete(); } - return getUpgradeStrategy().upgrade(exchange, webSocketHandler); + Optional subProtocol = selectSubProtocol(request, handler); + + return getUpgradeStrategy().upgrade(exchange, handler, subProtocol); } private boolean isWebSocketUpgrade(ServerHttpRequest request) { @@ -217,4 +224,15 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle { return true; } + private Optional selectSubProtocol(ServerHttpRequest request, WebSocketHandler handler) { + String protocolHeader = request.getHeaders().getFirst(SEC_WEBSOCKET_PROTOCOL); + if (protocolHeader == null) { + return Optional.empty(); + } + String[] protocols = handler.getSubProtocols(); + return StringUtils.commaDelimitedListToSet(protocolHeader).stream() + .filter(protocol -> Arrays.stream(protocols).anyMatch(protocol::equals)) + .findFirst(); + } + } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/DefaultServerEndpointConfig.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/DefaultServerEndpointConfig.java index dc3db8e575..da46ec13bc 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/DefaultServerEndpointConfig.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/DefaultServerEndpointConfig.java @@ -44,6 +44,8 @@ class DefaultServerEndpointConfig extends ServerEndpointConfig.Configurator private final Endpoint endpoint; + private List protocols = new ArrayList<>(); + /** * Constructor with a path and an {@code javax.websocket.Endpoint}. @@ -83,9 +85,13 @@ class DefaultServerEndpointConfig extends ServerEndpointConfig.Configurator return this.path; } + public void setSubprotocols(List protocols) { + this.protocols = protocols; + } + @Override public List getSubprotocols() { - return new ArrayList<>(); + return this.protocols; } @Override diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/JettyRequestUpgradeStrategy.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/JettyRequestUpgradeStrategy.java index e45bbcff7e..bd7b409e46 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/JettyRequestUpgradeStrategy.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/JettyRequestUpgradeStrategy.java @@ -17,6 +17,8 @@ package org.springframework.web.reactive.socket.server.upgrade; import java.io.IOException; +import java.security.Principal; +import java.util.Optional; import javax.servlet.ServletContext; import javax.servlet.http.HttpServletRequest; import javax.servlet.http.HttpServletResponse; @@ -46,9 +48,10 @@ import org.springframework.web.server.ServerWebExchange; * @author Rossen Stoyanchev * @since 5.0 */ +@SuppressWarnings("OptionalUsedAsFieldOrParameterType") public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Lifecycle { - private static final ThreadLocal adapterHolder = + private static final ThreadLocal adapterHolder = new NamedThreadLocal<>("JettyWebSocketHandlerAdapter"); @@ -68,7 +71,14 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Life this.running = true; try { this.factory = new WebSocketServerFactory(this.servletContext); - this.factory.setCreator((request, response) -> adapterHolder.get()); + this.factory.setCreator((request, response) -> { + WebSocketHandlerContainer container = adapterHolder.get(); + String protocol = container.getProtocol().orElse(null); + if (protocol != null) { + response.setAcceptedSubProtocol(protocol); + } + return container.getAdapter(); + }); this.factory.start(); } catch (Throwable ex) { @@ -100,7 +110,8 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Life @Override - public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler) { + public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler, + Optional subProtocol) { ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); @@ -108,7 +119,7 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Life HttpServletRequest servletRequest = getHttpServletRequest(request); HttpServletResponse servletResponse = getHttpServletResponse(response); - HandshakeInfo info = getHandshakeInfo(exchange); + HandshakeInfo info = getHandshakeInfo(exchange, subProtocol); DataBufferFactory factory = response.bufferFactory(); JettyWebSocketHandlerAdapter adapter = new JettyWebSocketHandlerAdapter(handler, info, factory); @@ -118,7 +129,7 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Life Assert.isTrue(isUpgrade, "Not a WebSocket handshake"); try { - adapterHolder.set(adapter); + adapterHolder.set(new WebSocketHandlerContainer(adapter, subProtocol)); this.factory.acceptWebSocket(servletRequest, servletResponse); } catch (IOException ex) { @@ -141,9 +152,10 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Life return ((ServletServerHttpResponse) response).getServletResponse(); } - private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange) { + private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange, Optional protocol) { ServerHttpRequest request = exchange.getRequest(); - return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal()); + Mono principal = exchange.getPrincipal(); + return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol); } private void startLazily(HttpServletRequest request) { @@ -159,4 +171,25 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy, Life } } + + private static class WebSocketHandlerContainer { + + private final JettyWebSocketHandlerAdapter adapter; + + private final Optional protocol; + + + public WebSocketHandlerContainer(JettyWebSocketHandlerAdapter adapter, Optional protocol) { + this.adapter = adapter; + this.protocol = protocol; + } + + public JettyWebSocketHandlerAdapter getAdapter() { + return this.adapter; + } + + public Optional getProtocol() { + return this.protocol; + } + } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/ReactorNettyRequestUpgradeStrategy.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/ReactorNettyRequestUpgradeStrategy.java index 80b19ce64f..9d98f31721 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/ReactorNettyRequestUpgradeStrategy.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/ReactorNettyRequestUpgradeStrategy.java @@ -15,14 +15,14 @@ */ package org.springframework.web.reactive.socket.server.upgrade; -import java.util.List; +import java.security.Principal; +import java.util.Optional; import reactor.core.publisher.Mono; import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.server.reactive.ReactorServerHttpResponse; import org.springframework.http.server.reactive.ServerHttpRequest; -import org.springframework.util.StringUtils; import org.springframework.web.reactive.socket.HandshakeInfo; import org.springframework.web.reactive.socket.WebSocketHandler; import org.springframework.web.reactive.socket.adapter.ReactorNettyWebSocketSession; @@ -35,31 +35,26 @@ import org.springframework.web.server.ServerWebExchange; * @author Rossen Stoyanchev * @since 5.0 */ +@SuppressWarnings("OptionalUsedAsFieldOrParameterType") public class ReactorNettyRequestUpgradeStrategy implements RequestUpgradeStrategy { @Override - public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler) { + public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler, + Optional subProtocol) { ReactorServerHttpResponse response = (ReactorServerHttpResponse) exchange.getResponse(); - HandshakeInfo handshakeInfo = getHandshakeInfo(exchange); + HandshakeInfo info = getHandshakeInfo(exchange, subProtocol); NettyDataBufferFactory bufferFactory = (NettyDataBufferFactory) response.bufferFactory(); - String protocols = StringUtils.arrayToCommaDelimitedString(getSubProtocols(handler)); - protocols = (StringUtils.hasText(protocols) ? protocols : null); - - return response.getReactorResponse().sendWebsocket(protocols, - (inbound, outbound) -> handler.handle( - new ReactorNettyWebSocketSession(inbound, outbound, handshakeInfo, bufferFactory))); + return response.getReactorResponse().sendWebsocket(subProtocol.orElse(null), + (in, out) -> handler.handle( + new ReactorNettyWebSocketSession(in, out, info, bufferFactory))); } - private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange) { + private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange, Optional protocol) { ServerHttpRequest request = exchange.getRequest(); - return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal()); - } - - private static String[] getSubProtocols(WebSocketHandler webSocketHandler) { - List subProtocols = webSocketHandler.getSubProtocols(); - return subProtocols.toArray(new String[subProtocols.size()]); + Mono principal = exchange.getPrincipal(); + return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol); } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/RxNettyRequestUpgradeStrategy.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/RxNettyRequestUpgradeStrategy.java index cca8172292..74b9533a86 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/RxNettyRequestUpgradeStrategy.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/RxNettyRequestUpgradeStrategy.java @@ -15,10 +15,11 @@ */ package org.springframework.web.reactive.socket.server.upgrade; -import java.util.List; +import java.security.Principal; +import java.util.Optional; +import io.reactivex.netty.protocol.http.ws.server.WebSocketHandshaker; import reactor.core.publisher.Mono; -import rx.Observable; import rx.RxReactiveStreams; import org.springframework.core.io.buffer.NettyDataBufferFactory; @@ -37,34 +38,39 @@ import org.springframework.web.server.ServerWebExchange; * @author Rossen Stoyanchev * @since 5.0 */ +@SuppressWarnings("OptionalUsedAsFieldOrParameterType") public class RxNettyRequestUpgradeStrategy implements RequestUpgradeStrategy { @Override - public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler) { + public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler, + Optional subProtocol) { RxNettyServerHttpResponse response = (RxNettyServerHttpResponse) exchange.getResponse(); - HandshakeInfo handshakeInfo = getHandshakeInfo(exchange); - NettyDataBufferFactory bufferFactory = (NettyDataBufferFactory) response.bufferFactory(); + HandshakeInfo info = getHandshakeInfo(exchange, subProtocol); + NettyDataBufferFactory factory = (NettyDataBufferFactory) response.bufferFactory(); - Observable completion = response.getRxNettyResponse() + WebSocketHandshaker handshaker = response.getRxNettyResponse() .acceptWebSocketUpgrade(conn -> { - WebSocketSession session = new RxNettyWebSocketSession(conn, handshakeInfo, bufferFactory); + WebSocketSession session = new RxNettyWebSocketSession(conn, info, factory); return RxReactiveStreams.toObservable(handler.handle(session)); - }) - .subprotocol(getSubProtocols(handler)); + }); - return Mono.from(RxReactiveStreams.toPublisher(completion)); + if (subProtocol.isPresent()) { + handshaker = handshaker.subprotocol(subProtocol.get()); + } + else { + // TODO: https://github.com/reactor/reactor-netty/issues/20 + handshaker = handshaker.subprotocol(new String[0]); + } + + return Mono.from(RxReactiveStreams.toPublisher(handshaker)); } - private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange) { + private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange, Optional protocol) { ServerHttpRequest request = exchange.getRequest(); - return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal()); - } - - private static String[] getSubProtocols(WebSocketHandler webSocketHandler) { - List subProtocols = webSocketHandler.getSubProtocols(); - return subProtocols.toArray(new String[subProtocols.size()]); + Mono principal = exchange.getPrincipal(); + return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol); } } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/TomcatRequestUpgradeStrategy.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/TomcatRequestUpgradeStrategy.java index 6f39df94c9..58413692ca 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/TomcatRequestUpgradeStrategy.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/TomcatRequestUpgradeStrategy.java @@ -17,7 +17,9 @@ package org.springframework.web.reactive.socket.server.upgrade; import java.io.IOException; +import java.security.Principal; import java.util.Collections; +import java.util.Optional; import javax.servlet.ServletContext; import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; @@ -46,13 +48,15 @@ import org.springframework.web.server.ServerWebExchange; * @author Violeta Georgieva * @since 5.0 */ +@SuppressWarnings("OptionalUsedAsFieldOrParameterType") public class TomcatRequestUpgradeStrategy implements RequestUpgradeStrategy { private static final String SERVER_CONTAINER_ATTR = "javax.websocket.server.ServerContainer"; @Override - public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler){ + public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler, + Optional subProtocol){ ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); @@ -60,12 +64,13 @@ public class TomcatRequestUpgradeStrategy implements RequestUpgradeStrategy { HttpServletRequest servletRequest = getHttpServletRequest(request); HttpServletResponse servletResponse = getHttpServletResponse(response); - HandshakeInfo info = getHandshakeInfo(exchange); + HandshakeInfo info = getHandshakeInfo(exchange, subProtocol); DataBufferFactory factory = response.bufferFactory(); Endpoint endpoint = new StandardWebSocketHandlerAdapter(handler, info, factory).getEndpoint(); String requestURI = servletRequest.getRequestURI(); - ServerEndpointConfig config = new DefaultServerEndpointConfig(requestURI, endpoint); + DefaultServerEndpointConfig config = new DefaultServerEndpointConfig(requestURI, endpoint); + config.setSubprotocols(subProtocol.map(Collections::singletonList).orElse(Collections.emptyList())); try { WsServerContainer container = getContainer(servletRequest); @@ -88,9 +93,10 @@ public class TomcatRequestUpgradeStrategy implements RequestUpgradeStrategy { return ((ServletServerHttpResponse) response).getServletResponse(); } - private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange) { + private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange, Optional protocol) { ServerHttpRequest request = exchange.getRequest(); - return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal()); + Mono principal = exchange.getPrincipal(); + return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol); } private WsServerContainer getContainer(HttpServletRequest request) { diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/UndertowRequestUpgradeStrategy.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/UndertowRequestUpgradeStrategy.java index b67c62ed55..25d22960ee 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/UndertowRequestUpgradeStrategy.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/UndertowRequestUpgradeStrategy.java @@ -16,9 +16,18 @@ package org.springframework.web.reactive.socket.server.upgrade; +import java.security.Principal; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Optional; +import java.util.Set; + import io.undertow.server.HttpServerExchange; import io.undertow.websockets.WebSocketConnectionCallback; import io.undertow.websockets.WebSocketProtocolHandshakeHandler; +import io.undertow.websockets.core.protocol.Handshake; +import io.undertow.websockets.core.protocol.version13.Hybi13Handshake; import reactor.core.publisher.Mono; import org.springframework.core.io.buffer.DataBufferFactory; @@ -38,16 +47,18 @@ import org.springframework.web.server.ServerWebExchange; * @author Violeta Georgieva * @since 5.0 */ +@SuppressWarnings("OptionalUsedAsFieldOrParameterType") public class UndertowRequestUpgradeStrategy implements RequestUpgradeStrategy { @Override - public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler) { + public Mono upgrade(ServerWebExchange exchange, WebSocketHandler handler, + Optional subProtocol) { ServerHttpRequest request = exchange.getRequest(); ServerHttpResponse response = exchange.getResponse(); - HandshakeInfo info = getHandshakeInfo(exchange); + HandshakeInfo info = getHandshakeInfo(exchange, subProtocol); DataBufferFactory bufferFactory = response.bufferFactory(); Assert.isTrue(request instanceof UndertowServerHttpRequest); @@ -56,8 +67,12 @@ public class UndertowRequestUpgradeStrategy implements RequestUpgradeStrategy { WebSocketConnectionCallback callback = new UndertowWebSocketHandlerAdapter(handler, info, bufferFactory); + Set protocols = subProtocol.map(Collections::singleton).orElse(Collections.emptySet()); + Hybi13Handshake handshake = new Hybi13Handshake(protocols, false); + List handshakes = Collections.singletonList(handshake); + try { - new WebSocketProtocolHandshakeHandler(callback).handleRequest(httpExchange); + new WebSocketProtocolHandshakeHandler(handshakes, callback).handleRequest(httpExchange); } catch (Exception ex) { return Mono.error(ex); @@ -66,9 +81,10 @@ public class UndertowRequestUpgradeStrategy implements RequestUpgradeStrategy { return Mono.empty(); } - private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange) { + private HandshakeInfo getHandshakeInfo(ServerWebExchange exchange, Optional protocol) { ServerHttpRequest request = exchange.getRequest(); - return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal()); + Mono principal = exchange.getPrincipal(); + return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol); } } diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java index 08ee46096a..368ff3f198 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java @@ -18,16 +18,22 @@ package org.springframework.web.reactive.socket.server; import java.net.URISyntaxException; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; +import org.hamcrest.Matchers; +import org.junit.Ignore; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.MonoProcessor; import reactor.core.publisher.ReplayProcessor; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.util.StringUtils; import org.springframework.web.reactive.HandlerMapping; import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping; +import org.springframework.web.reactive.socket.HandshakeInfo; import org.springframework.web.reactive.socket.WebSocketHandler; import org.springframework.web.reactive.socket.WebSocketMessage; import org.springframework.web.reactive.socket.WebSocketSession; @@ -36,6 +42,7 @@ import org.springframework.web.reactive.socket.client.RxNettyWebSocketClient; import org.springframework.web.reactive.socket.client.WebSocketClient; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; /** * Integration tests with server-side {@link WebSocketHandler}s. @@ -52,7 +59,7 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests @Test - public void echoReactorNettyClient() throws Exception { + public void echoReactorClient() throws Exception { testEcho(new ReactorNettyWebSocketClient()); } @@ -77,6 +84,49 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests assertEquals(input.collectList().blockMillis(5000), output.collectList().blockMillis(5000)); } + @Test + @Ignore("https://github.com/reactor/reactor-netty/issues/20") + public void subProtocolReactorNettyClient() throws Exception { + testSubProtocol(new ReactorNettyWebSocketClient()); + } + + @Test + public void subProtocolRxNettyClient() throws Exception { + testSubProtocol(new RxNettyWebSocketClient()); + } + + private void testSubProtocol(WebSocketClient client) throws URISyntaxException { + String protocol = "echo-v1"; + AtomicReference infoRef = new AtomicReference<>(); + MonoProcessor output = MonoProcessor.create(); + + client.execute(getUrl("/sub-protocol"), + new SubProtocolWebSocketHandler(protocol) { + + @Override + public Mono handle(WebSocketSession session) { + infoRef.set(session.getHandshakeInfo()); + return session.receive() + .map(WebSocketMessage::getPayloadAsText) + .subscribeWith(output) + .then(); + } + }) + .blockMillis(5000); + + HandshakeInfo info = infoRef.get(); + assertThat(info.getHeaders().getFirst("Upgrade"), Matchers.equalToIgnoringCase("websocket")); + assertEquals(protocol, info.getHeaders().getFirst("Sec-WebSocket-Protocol")); + assertEquals("Wrong protocol accepted", protocol, info.getSubProtocol().orElse("none")); + assertEquals("Wrong protocol detected on the server side", protocol, output.blockMillis(5000)); + } + + + @Test + public void customHeaders() throws Exception { + // TODO + } + @Configuration static class WebConfig { @@ -86,6 +136,7 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests Map map = new HashMap<>(); map.put("/echo", new EchoWebSocketHandler()); + map.put("/sub-protocol", new SubProtocolWebSocketHandler("echo-v1")); SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping(); mapping.setUrlMap(map); @@ -102,4 +153,25 @@ public class WebSocketIntegrationTests extends AbstractWebSocketIntegrationTests } } + private static class SubProtocolWebSocketHandler implements WebSocketHandler { + + private final String subProtocols; + + public SubProtocolWebSocketHandler(String subProtocols) { + this.subProtocols = subProtocols; + } + + @Override + public String[] getSubProtocols() { + return StringUtils.commaDelimitedListToStringArray(this.subProtocols); + } + + @Override + public Mono handle(WebSocketSession session) { + String protocol = session.getHandshakeInfo().getSubProtocol().orElse("none"); + WebSocketMessage message = session.textMessage(protocol); + return session.send(Mono.just(message)); + } + } + }