Support for Jetty 10

Closes gh-26123

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/Jetty10WebSocketHandlerAdapter.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2002-2021 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
+ *
+ *      https://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.adapter;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.function.Function;
+
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketConnect;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
+import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
+import org.eclipse.jetty.websocket.api.annotations.WebSocket;
+import org.eclipse.jetty.websocket.api.extensions.Frame;
+
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+import org.springframework.web.reactive.socket.CloseStatus;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.WebSocketMessage;
+import org.springframework.web.reactive.socket.WebSocketMessage.Type;
+import org.springframework.web.reactive.socket.WebSocketSession;
+
+/**
+ * Identical to {@link JettyWebSocketHandlerAdapter}, only excluding the
+ * {@code onWebSocketFrame} method, since the {@link Frame} argument has moved
+ * to a different package in Jetty 10.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.3.4
+ */
+@WebSocket
+public class Jetty10WebSocketHandlerAdapter {
+
+	private static final ByteBuffer EMPTY_PAYLOAD = ByteBuffer.wrap(new byte[0]);
+
+
+	private final WebSocketHandler delegateHandler;
+
+	private final Function<Session, JettyWebSocketSession> sessionFactory;
+
+	@Nullable
+	private JettyWebSocketSession delegateSession;
+
+
+	public Jetty10WebSocketHandlerAdapter(WebSocketHandler handler,
+			Function<Session, JettyWebSocketSession> sessionFactory) {
+
+		Assert.notNull(handler, "WebSocketHandler is required");
+		Assert.notNull(sessionFactory, "'sessionFactory' is required");
+		this.delegateHandler = handler;
+		this.sessionFactory = sessionFactory;
+	}
+
+
+	@OnWebSocketConnect
+	public void onWebSocketConnect(Session session) {
+		this.delegateSession = this.sessionFactory.apply(session);
+		this.delegateHandler.handle(this.delegateSession)
+				.checkpoint(session.getUpgradeRequest().getRequestURI() + " [JettyWebSocketHandlerAdapter]")
+				.subscribe(this.delegateSession);
+	}
+
+	@OnWebSocketMessage
+	public void onWebSocketText(String message) {
+		if (this.delegateSession != null) {
+			WebSocketMessage webSocketMessage = toMessage(Type.TEXT, message);
+			this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
+		}
+	}
+
+	@OnWebSocketMessage
+	public void onWebSocketBinary(byte[] message, int offset, int length) {
+		if (this.delegateSession != null) {
+			ByteBuffer buffer = ByteBuffer.wrap(message, offset, length);
+			WebSocketMessage webSocketMessage = toMessage(Type.BINARY, buffer);
+			this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
+		}
+	}
+
+// TODO: onWebSocketFrame can't be declared without compiling against Jetty 10
+//   Jetty 10: org.eclipse.jetty.websocket.api.Frame
+//   Jetty  9: org.eclipse.jetty.websocket.api.extensions.Frame
+
+//	@OnWebSocketFrame
+//	public void onWebSocketFrame(Frame frame) {
+//		if (this.delegateSession != null) {
+//			if (OpCode.PONG == frame.getOpCode()) {
+//				ByteBuffer buffer = (frame.getPayload() != null ? frame.getPayload() : EMPTY_PAYLOAD);
+//				WebSocketMessage webSocketMessage = toMessage(Type.PONG, buffer);
+//				this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
+//			}
+//		}
+//	}
+
+	private <T> WebSocketMessage toMessage(Type type, T message) {
+		WebSocketSession session = this.delegateSession;
+		Assert.state(session != null, "Cannot create message without a session");
+		if (Type.TEXT.equals(type)) {
+			byte[] bytes = ((String) message).getBytes(StandardCharsets.UTF_8);
+			DataBuffer buffer = session.bufferFactory().wrap(bytes);
+			return new WebSocketMessage(Type.TEXT, buffer);
+		}
+		else if (Type.BINARY.equals(type)) {
+			DataBuffer buffer = session.bufferFactory().wrap((ByteBuffer) message);
+			return new WebSocketMessage(Type.BINARY, buffer);
+		}
+		else if (Type.PONG.equals(type)) {
+			DataBuffer buffer = session.bufferFactory().wrap((ByteBuffer) message);
+			return new WebSocketMessage(Type.PONG, buffer);
+		}
+		else {
+			throw new IllegalArgumentException("Unexpected message type: " + message);
+		}
+	}
+
+	@OnWebSocketClose
+	public void onWebSocketClose(int statusCode, String reason) {
+		if (this.delegateSession != null) {
+			this.delegateSession.handleClose(CloseStatus.create(statusCode, reason));
+		}
+	}
+
+	@OnWebSocketError
+	public void onWebSocketError(Throwable cause) {
+		if (this.delegateSession != null) {
+			this.delegateSession.handleError(cause);
+		}
+	}
+
+}

spring-webflux/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -17,7 +17,9 @@
 package org.springframework.web.reactive.socket.client;
 
 import java.io.IOException;
+import java.lang.reflect.Method;
 import java.net.URI;
+import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -32,9 +34,12 @@ import reactor.core.publisher.Sinks;
 import org.springframework.context.Lifecycle;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.HttpHeaders;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.ReflectionUtils;
 import org.springframework.web.reactive.socket.HandshakeInfo;
 import org.springframework.web.reactive.socket.WebSocketHandler;
 import org.springframework.web.reactive.socket.adapter.ContextWebSocketHandler;
+import org.springframework.web.reactive.socket.adapter.Jetty10WebSocketHandlerAdapter;
 import org.springframework.web.reactive.socket.adapter.JettyWebSocketHandlerAdapter;
 import org.springframework.web.reactive.socket.adapter.JettyWebSocketSession;
 
@@ -53,6 +58,16 @@ import org.springframework.web.reactive.socket.adapter.JettyWebSocketSession;
  */
 public class JettyWebSocketClient implements WebSocketClient, Lifecycle {
 
+	private static ClassLoader loader = JettyWebSocketClient.class.getClassLoader();
+
+	private static final boolean jetty10Present;
+
+	static {
+		jetty10Present = ClassUtils.isPresent(
+				"org.eclipse.jetty.websocket.client.JettyUpgradeListener", loader);
+	}
+
+
 	private static final Log logger = LogFactory.getLog(JettyWebSocketClient.class);
 
 
@@ -60,6 +75,9 @@ public class JettyWebSocketClient implements WebSocketClient, Lifecycle {
 
 	private final boolean externallyManaged;
 
+	private final UpgradeHelper upgradeHelper =
+			(jetty10Present ? new Jetty10UpgradeHelper() : new Jetty9UpgradeHelper());
+
 
 	/**
 	 * Default constructor that creates and manages an instance of a Jetty
@@ -147,22 +165,19 @@ public class JettyWebSocketClient implements WebSocketClient, Lifecycle {
 					url, ContextWebSocketHandler.decorate(handler, contextView), completionSink);
 			ClientUpgradeRequest request = new ClientUpgradeRequest();
 			request.setSubProtocols(handler.getSubProtocols());
-			UpgradeListener upgradeListener = new DefaultUpgradeListener(headers);
-			try {
-				this.jettyClient.connect(jettyHandler, url, request, upgradeListener);
-				return completionSink.asMono();
-			}
-			catch (IOException ex) {
-				return Mono.error(ex);
-			}
+			return this.upgradeHelper.upgrade(
+					this.jettyClient, jettyHandler, url, request, headers, completionSink);
 		});
 	}
 
 	private Object createHandler(URI url, WebSocketHandler handler, Sinks.Empty<Void> completion) {
-		return new JettyWebSocketHandlerAdapter(handler, session -> {
+		Function<Session, JettyWebSocketSession> sessionFactory = session -> {
 			HandshakeInfo info = createHandshakeInfo(url, session);
 			return new JettyWebSocketSession(session, info, DefaultDataBufferFactory.sharedInstance, completion);
-		});
+		};
+		return (jetty10Present ?
+				new Jetty10WebSocketHandlerAdapter(handler, sessionFactory) :
+				new JettyWebSocketHandlerAdapter(handler, sessionFactory));
 	}
 
 	private HandshakeInfo createHandshakeInfo(URI url, Session jettySession) {
@@ -173,6 +188,34 @@ public class JettyWebSocketClient implements WebSocketClient, Lifecycle {
 	}
 
 
+	/**
+	 * Encapsulate incompatible changes between Jetty 9.4 and 10.
+	 */
+	private interface UpgradeHelper {
+
+		Mono<Void> upgrade(org.eclipse.jetty.websocket.client.WebSocketClient jettyClient,
+				Object jettyHandler, URI url, ClientUpgradeRequest request, HttpHeaders headers,
+				Sinks.Empty<Void> completionSink);
+	}
+
+
+	private static class Jetty9UpgradeHelper implements UpgradeHelper {
+
+		@Override
+		public Mono<Void> upgrade(org.eclipse.jetty.websocket.client.WebSocketClient jettyClient,
+				Object jettyHandler, URI url, ClientUpgradeRequest request, HttpHeaders headers,
+				Sinks.Empty<Void> completionSink) {
+
+			try {
+				jettyClient.connect(jettyHandler, url, request, new DefaultUpgradeListener(headers));
+				return completionSink.asMono();
+			}
+			catch (IOException ex) {
+				return Mono.error(ex);
+			}
+		}
+	}
+
 	private static class DefaultUpgradeListener implements UpgradeListener {
 
 		private final HttpHeaders headers;
@@ -192,4 +235,32 @@ public class JettyWebSocketClient implements WebSocketClient, Lifecycle {
 		}
 	}
 
+	private static class Jetty10UpgradeHelper implements UpgradeHelper {
+
+		// On Jetty 9 returns Future, on Jetty 10 returns CompletableFuture
+		private static final Method connectMethod;
+
+		static {
+			try {
+				Class<?> type = loader.loadClass("org.eclipse.jetty.websocket.client.WebSocketClient");
+				connectMethod = type.getMethod("connect", Object.class, URI.class, ClientUpgradeRequest.class);
+			}
+			catch (ClassNotFoundException | NoSuchMethodException ex) {
+				throw new IllegalStateException("No compatible Jetty version found", ex);
+			}
+		}
+
+		@Override
+		public Mono<Void> upgrade(org.eclipse.jetty.websocket.client.WebSocketClient jettyClient,
+				Object jettyHandler, URI url, ClientUpgradeRequest request, HttpHeaders headers,
+				Sinks.Empty<Void> completionSink) {
+
+			// TODO: pass JettyUpgradeListener argument to set headers from HttpHeaders (like we do for Jetty 9)
+			//  which would require a JDK Proxy since it is new in Jetty 10
+
+			ReflectionUtils.invokeMethod(connectMethod, jettyClient, jettyHandler, url, request);
+			return completionSink.asMono();
+		}
+	}
+
 }

spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/support/HandshakeWebSocketService.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2020 the original author or authors.
+ * Copyright 2002-2021 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.
@@ -68,16 +68,19 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle {
 
 	private static final boolean jettyPresent;
 
+	private static final boolean jetty10Present;
+
 	private static final boolean undertowPresent;
 
 	private static final boolean reactorNettyPresent;
 
 	static {
-		ClassLoader classLoader = HandshakeWebSocketService.class.getClassLoader();
-		tomcatPresent = ClassUtils.isPresent("org.apache.tomcat.websocket.server.WsHttpUpgradeHandler", classLoader);
-		jettyPresent = ClassUtils.isPresent("org.eclipse.jetty.websocket.server.WebSocketServerFactory", classLoader);
-		undertowPresent = ClassUtils.isPresent("io.undertow.websockets.WebSocketProtocolHandshakeHandler", classLoader);
-		reactorNettyPresent = ClassUtils.isPresent("reactor.netty.http.server.HttpServerResponse", classLoader);
+		ClassLoader loader = HandshakeWebSocketService.class.getClassLoader();
+		tomcatPresent = ClassUtils.isPresent("org.apache.tomcat.websocket.server.WsHttpUpgradeHandler", loader);
+		jettyPresent = ClassUtils.isPresent("org.eclipse.jetty.websocket.server.WebSocketServerFactory", loader);
+		jetty10Present = ClassUtils.isPresent("org.eclipse.jetty.websocket.server.JettyWebSocketServerContainer", loader);
+		undertowPresent = ClassUtils.isPresent("io.undertow.websockets.WebSocketProtocolHandshakeHandler", loader);
+		reactorNettyPresent = ClassUtils.isPresent("reactor.netty.http.server.HttpServerResponse", loader);
 	}
 
 
@@ -117,6 +120,9 @@ public class HandshakeWebSocketService implements WebSocketService, Lifecycle {
 		else if (jettyPresent) {
 			className = "JettyRequestUpgradeStrategy";
 		}
+		else if (jetty10Present) {
+			className = "Jetty10RequestUpgradeStrategy";
+		}
 		else if (undertowPresent) {
 			className = "UndertowRequestUpgradeStrategy";
 		}

spring-webflux/src/main/java/org/springframework/web/reactive/socket/server/upgrade/Jetty10RequestUpgradeStrategy.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2002-2021 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
+ *
+ *      https://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.server.upgrade;
+
+import java.lang.reflect.Method;
+import java.util.function.Supplier;
+
+import javax.servlet.ServletContext;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import reactor.core.publisher.Mono;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.target.EmptyTargetSource;
+import org.springframework.core.io.buffer.DataBufferFactory;
+import org.springframework.http.server.reactive.ServerHttpRequest;
+import org.springframework.http.server.reactive.ServerHttpRequestDecorator;
+import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.http.server.reactive.ServerHttpResponseDecorator;
+import org.springframework.lang.NonNull;
+import org.springframework.lang.Nullable;
+import org.springframework.util.ReflectionUtils;
+import org.springframework.web.reactive.socket.HandshakeInfo;
+import org.springframework.web.reactive.socket.WebSocketHandler;
+import org.springframework.web.reactive.socket.adapter.ContextWebSocketHandler;
+import org.springframework.web.reactive.socket.adapter.Jetty10WebSocketHandlerAdapter;
+import org.springframework.web.reactive.socket.adapter.JettyWebSocketSession;
+import org.springframework.web.reactive.socket.server.RequestUpgradeStrategy;
+import org.springframework.web.server.ServerWebExchange;
+
+/**
+ * A {@link RequestUpgradeStrategy} for use with Jetty 10.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.3.4
+ */
+public class Jetty10RequestUpgradeStrategy implements RequestUpgradeStrategy {
+
+	private static final Class<?> webSocketCreatorClass;
+
+	private static final Method getContainerMethod;
+
+	private static final Method upgradeMethod;
+
+	private static final Method setAcceptedSubProtocol;
+
+	static {
+		ClassLoader loader = Jetty10RequestUpgradeStrategy.class.getClassLoader();
+		try {
+			webSocketCreatorClass = loader.loadClass("org.eclipse.jetty.websocket.server.JettyWebSocketCreator");
+
+			Class<?> type = loader.loadClass("org.eclipse.jetty.websocket.server.JettyWebSocketServerContainer");
+			getContainerMethod = type.getMethod("getContainer", ServletContext.class);
+			upgradeMethod = ReflectionUtils.findMethod(type, "upgrade", (Class<?>[]) null);
+
+			type = loader.loadClass("org.eclipse.jetty.websocket.server.JettyServerUpgradeResponse");
+			setAcceptedSubProtocol = type.getMethod("setAcceptedSubProtocol", String.class);
+		}
+		catch (Exception ex) {
+			throw new IllegalStateException("No compatible Jetty version found", ex);
+		}
+	}
+
+
+	@Override
+	public Mono<Void> upgrade(
+			ServerWebExchange exchange, WebSocketHandler handler,
+			@Nullable String subProtocol, Supplier<HandshakeInfo> handshakeInfoFactory) {
+
+		ServerHttpRequest request = exchange.getRequest();
+		ServerHttpResponse response = exchange.getResponse();
+
+		HttpServletRequest servletRequest = ServerHttpRequestDecorator.getNativeRequest(request);
+		HttpServletResponse servletResponse = ServerHttpResponseDecorator.getNativeResponse(response);
+		ServletContext servletContext = servletRequest.getServletContext();
+
+		HandshakeInfo handshakeInfo = handshakeInfoFactory.get();
+		DataBufferFactory factory = response.bufferFactory();
+
+		// Trigger WebFlux preCommit actions and upgrade
+		return exchange.getResponse().setComplete()
+				.then(Mono.deferContextual(contextView -> {
+					Jetty10WebSocketHandlerAdapter adapter = new Jetty10WebSocketHandlerAdapter(
+							ContextWebSocketHandler.decorate(handler, contextView),
+							session -> new JettyWebSocketSession(session, handshakeInfo, factory));
+
+					try {
+						Object creator = createJettyWebSocketCreator(adapter, subProtocol);
+						Object container = ReflectionUtils.invokeMethod(getContainerMethod, null, servletContext);
+						ReflectionUtils.invokeMethod(upgradeMethod, container, creator, servletRequest, servletResponse);
+					}
+					catch (Exception ex) {
+						return Mono.error(ex);
+					}
+					return Mono.empty();
+				}));
+	}
+
+	private static Object createJettyWebSocketCreator(
+			Jetty10WebSocketHandlerAdapter adapter, @Nullable String protocol) {
+
+		ProxyFactory factory = new ProxyFactory(EmptyTargetSource.INSTANCE);
+		factory.addInterface(webSocketCreatorClass);
+		factory.addAdvice(new WebSocketCreatorInterceptor(adapter, protocol));
+		return factory.getProxy();
+	}
+
+
+	/**
+	 * Proxy for a JettyWebSocketCreator to supply the WebSocket handler and set the sub-protocol.
+	 */
+	private static class WebSocketCreatorInterceptor implements MethodInterceptor {
+
+		private final Jetty10WebSocketHandlerAdapter adapter;
+
+		@Nullable
+		private final String protocol;
+
+
+		public WebSocketCreatorInterceptor(
+				Jetty10WebSocketHandlerAdapter adapter, @Nullable String protocol) {
+
+			this.adapter = adapter;
+			this.protocol = protocol;
+		}
+
+		@Nullable
+		@Override
+		public Object invoke(@NonNull MethodInvocation invocation) {
+			if (this.protocol != null) {
+				ReflectionUtils.invokeMethod(
+						setAcceptedSubProtocol, invocation.getArguments()[2], this.protocol);
+			}
+			return this.adapter;
+		}
+	}
+
+}