Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Sub-protocol negotiation for reactive WebSocket support

Browse Source 
Issue: SPR-14527
This commit is contained in:
Rossen Stoyanchev
2016-12-20 12:21:30 -05:00
parent 241d5f7a3b
commit 3b987c263c
14 changed files with 365 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/HandshakeInfo.java
View File

@@ -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<Principal> principalMono;
private final Optional<String> protocol;
public HandshakeInfo(URI uri, HttpHeaders headers, Mono<Principal> principal,
Optional<String> subProtocol) {
public HandshakeInfo(URI uri, HttpHeaders headers, Mono<Principal> 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<Principal> getPrincipal() {
return this.principalMono;
}
/**
* The sub-protocol negotiated at handshake time.
* @see <a href="https://tools.ietf.org/html/rfc6455#section-1.9">
* https://tools.ietf.org/html/rfc6455#section-1.9</a>
*/
public Optional<String> getSubProtocol() {
return this.protocol;
}
@Override
public String toString() {

6
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/WebSocketHandler.java
View File

@@ -30,10 +30,10 @@ public interface WebSocketHandler {
/**
* Return the list of sub-protocols supported by this handler.
* <p>By default an empty list is returned.
* <p>By default an empty array is returned.
*/
default List<String> getSubProtocols() {
return Collections.emptyList();
default String[] getSubProtocols() {
return new String[0];
}
/**

43
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/ReactorNettyWebSocketClient.java
View File

@@ -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<Void> 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<NettyOutbound> 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;
}
}

69
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/RxNettyWebSocketClient.java
View File

@@ -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<URI, HttpClient<ByteBuf, ByteBuf>> httpClientFactory;
@@ -91,32 +99,65 @@ public class RxNettyWebSocketClient implements WebSocketClient {
@Override
public Mono<Void> execute(URI url, HttpHeaders headers, WebSocketHandler handler) {
HandshakeInfo info = new HandshakeInfo(url, headers, Mono.empty());
Observable<Void> completion = connectInternal(handler, info);
Observable<Void> completion = connectInternal(url, headers, handler);
return Mono.from(RxReactiveStreams.toPublisher(completion));
}
private Observable<Void> connectInternal(WebSocketHandler handler, HandshakeInfo info) {
return createWebSocketRequest(info.getUri())
private Observable<Void> 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<WebSocketConnection> 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<ByteBuf> response = tuple.getT1();
HttpHeaders responseHeaders = getResponseHeaders(response);
Optional<String> 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<ByteBuf> createWebSocketRequest(URI url) {
private WebSocketRequest<ByteBuf> 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<ByteBuf> 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<String, List<Object>> toObjectValueMap(HttpHeaders headers) {
if (headers.isEmpty()) {
return Collections.emptyMap();
}
Map<String, List<Object>> map = new HashMap<>(headers.size());
headers.keySet().stream().forEach(key -> map.put(key, new ArrayList<>(headers.get(key))));
return map;
}
private HttpHeaders getResponseHeaders(WebSocketResponse<ByteBuf> response) {
HttpHeaders headers = new HttpHeaders();
response.headerIterator().forEachRemaining(entry -> {
String name = entry.getKey().toString();
headers.put(name, response.getAllHeaderValues(name));
});
return headers;
}
}

40
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java Normal file
View File

@@ -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());
}
}

7
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/RequestUpgradeStrategy.java
View File

@@ -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<Void>} to indicate the outcome of the
* WebSocket session handling.
*/
Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler webSocketHandler);
@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler webSocketHandler,
Optional<String> subProtocol);
}

22
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java
View File

@@ -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<Void> handleRequest(ServerWebExchange exchange, WebSocketHandler webSocketHandler) {
public Mono<Void> 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<String> 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<String> 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();
}
}

8
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/DefaultServerEndpointConfig.java
View File

@@ -44,6 +44,8 @@ class DefaultServerEndpointConfig extends ServerEndpointConfig.Configurator
private final Endpoint endpoint;
private List<String> 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<String> protocols) {
this.protocols = protocols;
}
@Override
public List<String> getSubprotocols() {
return new ArrayList<>();
return this.protocols;
}
@Override

47
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/JettyRequestUpgradeStrategy.java
View File

@@ -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<JettyWebSocketHandlerAdapter> adapterHolder =
private static final ThreadLocal<WebSocketHandlerContainer> 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<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler) {
public Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler,
Optional<String> 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<String> protocol) {
ServerHttpRequest request = exchange.getRequest();
return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal());
Mono<Principal> 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<String> protocol;
public WebSocketHandlerContainer(JettyWebSocketHandlerAdapter adapter, Optional<String> protocol) {
this.adapter = adapter;
this.protocol = protocol;
}
public JettyWebSocketHandlerAdapter getAdapter() {
return this.adapter;
}
public Optional<String> getProtocol() {
return this.protocol;
}
}
}

29
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/ReactorNettyRequestUpgradeStrategy.java
View File

@@ -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<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler) {
public Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler,
Optional<String> 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<String> protocol) {
ServerHttpRequest request = exchange.getRequest();
return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal());
}
private static String[] getSubProtocols(WebSocketHandler webSocketHandler) {
List<String> subProtocols = webSocketHandler.getSubProtocols();
return subProtocols.toArray(new String[subProtocols.size()]);
Mono<Principal> principal = exchange.getPrincipal();
return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol);
}
}

40
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/RxNettyRequestUpgradeStrategy.java
View File

@@ -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<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler) {
public Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler,
Optional<String> 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<Void> 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<String> protocol) {
ServerHttpRequest request = exchange.getRequest();
return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal());
}
private static String[] getSubProtocols(WebSocketHandler webSocketHandler) {
List<String> subProtocols = webSocketHandler.getSubProtocols();
return subProtocols.toArray(new String[subProtocols.size()]);
Mono<Principal> principal = exchange.getPrincipal();
return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol);
}
}

16
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/TomcatRequestUpgradeStrategy.java
View File

@@ -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<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler){
public Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler,
Optional<String> 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<String> protocol) {
ServerHttpRequest request = exchange.getRequest();
return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal());
Mono<Principal> principal = exchange.getPrincipal();
return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol);
}
private WsServerContainer getContainer(HttpServletRequest request) {

26
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/server/upgrade/UndertowRequestUpgradeStrategy.java
View File

@@ -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<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler) {
public Mono<Void> upgrade(ServerWebExchange exchange, WebSocketHandler handler,
Optional<String> 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<String> protocols = subProtocol.map(Collections::singleton).orElse(Collections.emptySet());
Hybi13Handshake handshake = new Hybi13Handshake(protocols, false);
List<Handshake> 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<String> protocol) {
ServerHttpRequest request = exchange.getRequest();
return new HandshakeInfo(request.getURI(), request.getHeaders(), exchange.getPrincipal());
Mono<Principal> principal = exchange.getPrincipal();
return new HandshakeInfo(request.getURI(), request.getHeaders(), principal, protocol);
}
}

74
spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java
View File

@@ -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<HandshakeInfo> infoRef = new AtomicReference<>();
MonoProcessor<Object> output = MonoProcessor.create();
client.execute(getUrl("/sub-protocol"),
new SubProtocolWebSocketHandler(protocol) {
@Override
public Mono<Void> 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<String, WebSocketHandler> 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<Void> handle(WebSocketSession session) {
String protocol = session.getHandshakeInfo().getSubProtocol().orElse("none");
WebSocketMessage message = session.textMessage(protocol);
return session.send(Mono.just(message));
}
}
}