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

Upgrade to Jetty 12

Browse Source 
This commit upgrades Spring Framework to Jetty 12.0.1, and Reactive HTTP
 Client 4.0.0.

Closes gh-30698
This commit is contained in:
Arjen Poutsma
2023-06-22 11:53:50 +02:00
parent 210b42b7d8
commit 6597727c86
32 changed files with 491 additions and 851 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketHandlerAdapter.java
View File

@@ -20,13 +20,14 @@ import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.function.Function;
import org.eclipse.jetty.websocket.api.Callback;
import org.eclipse.jetty.websocket.api.Frame;
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.OnWebSocketFrame;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen;
import org.eclipse.jetty.websocket.api.annotations.WebSocket;
import org.eclipse.jetty.websocket.core.OpCode;
@@ -71,8 +72,8 @@ public class JettyWebSocketHandlerAdapter {
}
@OnWebSocketConnect
public void onWebSocketConnect(Session session) {
@OnWebSocketOpen
public void onWebSocketOpen(Session session) {
this.delegateSession = this.sessionFactory.apply(session);
this.delegateHandler.handle(this.delegateSession)
.checkpoint(session.getUpgradeRequest().getRequestURI() + " [JettyWebSocketHandlerAdapter]")
@@ -88,21 +89,22 @@ public class JettyWebSocketHandlerAdapter {
}
@OnWebSocketMessage
public void onWebSocketBinary(byte[] message, int offset, int length) {
public void onWebSocketBinary(ByteBuffer buffer, Callback callback) {
if (this.delegateSession != null) {
ByteBuffer buffer = ByteBuffer.wrap(message, offset, length);
WebSocketMessage webSocketMessage = toMessage(Type.BINARY, buffer);
this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
callback.succeed();
}
}
@OnWebSocketFrame
public void onWebSocketFrame(Frame frame) {
public void onWebSocketFrame(Frame frame, Callback callback) {
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);
callback.succeed();
}
}
}

41
spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketSession.java
View File

@@ -20,17 +20,14 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import org.eclipse.jetty.websocket.api.RemoteEndpoint;
import org.eclipse.jetty.websocket.api.Callback;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.SuspendToken;
import org.eclipse.jetty.websocket.api.WriteCallback;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.HandshakeInfo;
@@ -47,10 +44,6 @@ import org.springframework.web.reactive.socket.WebSocketSession;
*/
public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Session> {
@Nullable
private volatile SuspendToken suspendToken;
public JettyWebSocketSession(Session session, HandshakeInfo info, DataBufferFactory factory) {
this(session, info, factory, null);
}
@@ -66,32 +59,26 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
@Override
protected boolean canSuspendReceiving() {
return true;
// Jetty 12 TODO: research suspend functionality in Jetty 12
return false;
}
@Override
protected void suspendReceiving() {
Assert.state(this.suspendToken == null, "Already suspended");
this.suspendToken = getDelegate().suspend();
}
@Override
protected void resumeReceiving() {
SuspendToken tokenToUse = this.suspendToken;
this.suspendToken = null;
if (tokenToUse != null) {
tokenToUse.resume();
}
}
@Override
protected boolean sendMessage(WebSocketMessage message) throws IOException {
DataBuffer dataBuffer = message.getPayload();
RemoteEndpoint remote = getDelegate().getRemote();
Session session = getDelegate();
if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
getSendProcessor().setReadyToSend(false);
String text = dataBuffer.toString(StandardCharsets.UTF_8);
remote.sendString(text, new SendProcessorCallback());
session.sendText(text, new SendProcessorCallback());
}
else {
if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
@@ -101,9 +88,9 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
while (iterator.hasNext()) {
ByteBuffer byteBuffer = iterator.next();
switch (message.getType()) {
case BINARY -> remote.sendBytes(byteBuffer, new SendProcessorCallback());
case PING -> remote.sendPing(byteBuffer);
case PONG -> remote.sendPong(byteBuffer);
case BINARY -> session.sendBinary(byteBuffer, new SendProcessorCallback());
case PING -> session.sendPing(byteBuffer, new SendProcessorCallback());
case PONG -> session.sendPong(byteBuffer, new SendProcessorCallback());
default -> throw new IllegalArgumentException("Unexpected message type: " + message.getType());
}
}
@@ -119,21 +106,23 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
@Override
public Mono<Void> close(CloseStatus status) {
getDelegate().close(status.getCode(), status.getReason());
return Mono.empty();
Callback.Completable callback = new Callback.Completable();
getDelegate().close(status.getCode(), status.getReason(), callback);
return Mono.fromFuture(callback);
}
private final class SendProcessorCallback implements WriteCallback {
private final class SendProcessorCallback implements Callback {
@Override
public void writeFailed(Throwable x) {
public void fail(Throwable x) {
getSendProcessor().cancel();
getSendProcessor().onError(x);
}
@Override
public void writeSuccess() {
public void succeed() {
getSendProcessor().setReadyToSend(true);
getSendProcessor().onWritePossible();
}

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

@@ -1,175 +0,0 @@
/*
* Copyright 2002-2022 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.client;
import java.io.IOException;
import java.net.URI;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import reactor.core.publisher.Mono;
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.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.JettyWebSocketHandlerAdapter;
import org.springframework.web.reactive.socket.adapter.JettyWebSocketSession;
/**
* A {@link WebSocketClient} implementation for use with Jetty
* {@link org.eclipse.jetty.websocket.client.WebSocketClient}.
* Only supported on Jetty 11, superseded by {@link StandardWebSocketClient}.
*
* <p><strong>Note: </strong> the Jetty {@code WebSocketClient} requires
* lifecycle management and must be started and stopped. This is automatically
* managed when this class is declared as a Spring bean and created with the
* default constructor. See constructor notes for more details.
*
* @author Violeta Georgieva
* @author Rossen Stoyanchev
* @author Juergen Hoeller
* @since 5.0
* @deprecated as of 6.0.3, in favor of {@link StandardWebSocketClient}
*/
@Deprecated(since = "6.0.3", forRemoval = true)
public class JettyWebSocketClient implements WebSocketClient, Lifecycle {
private static final Log logger = LogFactory.getLog(JettyWebSocketClient.class);
private final org.eclipse.jetty.websocket.client.WebSocketClient jettyClient;
private final boolean externallyManaged;
/**
* Default constructor that creates and manages an instance of a Jetty
* {@link org.eclipse.jetty.websocket.client.WebSocketClient WebSocketClient}.
* The instance can be obtained with {@link #getJettyClient()} for further
* configuration.
* <p><strong>Note: </strong> When this constructor is used {@link Lifecycle}
* methods of this class are delegated to the Jetty {@code WebSocketClient}.
*/
public JettyWebSocketClient() {
this.jettyClient = new org.eclipse.jetty.websocket.client.WebSocketClient();
this.externallyManaged = false;
}
/**
* Constructor that accepts an existing instance of a Jetty
* {@link org.eclipse.jetty.websocket.client.WebSocketClient WebSocketClient}.
* <p><strong>Note: </strong> Use of this constructor implies the Jetty
* {@code WebSocketClient} is externally managed and hence {@link Lifecycle}
* methods of this class are not delegated to it.
*/
public JettyWebSocketClient(org.eclipse.jetty.websocket.client.WebSocketClient jettyClient) {
this.jettyClient = jettyClient;
this.externallyManaged = true;
}
/**
* Return the underlying Jetty {@code WebSocketClient}.
*/
public org.eclipse.jetty.websocket.client.WebSocketClient getJettyClient() {
return this.jettyClient;
}
@Override
public void start() {
if (!this.externallyManaged) {
try {
this.jettyClient.start();
}
catch (Exception ex) {
throw new IllegalStateException("Failed to start Jetty WebSocketClient", ex);
}
}
}
@Override
public void stop() {
if (!this.externallyManaged) {
try {
this.jettyClient.stop();
}
catch (Exception ex) {
throw new IllegalStateException("Error stopping Jetty WebSocketClient", ex);
}
}
}
@Override
public boolean isRunning() {
return this.jettyClient.isRunning();
}
@Override
public Mono<Void> execute(URI url, WebSocketHandler handler) {
return execute(url, new HttpHeaders(), handler);
}
@Override
public Mono<Void> execute(URI url, HttpHeaders headers, WebSocketHandler handler) {
return executeInternal(url, headers, handler);
}
private Mono<Void> executeInternal(URI url, HttpHeaders headers, WebSocketHandler handler) {
Sinks.Empty<Void> completionSink = Sinks.empty();
return Mono.deferContextual(contextView -> {
if (logger.isDebugEnabled()) {
logger.debug("Connecting to " + url);
}
Object jettyHandler = createHandler(
url, ContextWebSocketHandler.decorate(handler, contextView), completionSink);
ClientUpgradeRequest request = new ClientUpgradeRequest();
request.setHeaders(headers);
request.setSubProtocols(handler.getSubProtocols());
try {
this.jettyClient.connect(jettyHandler, url, request);
return completionSink.asMono();
}
catch (IOException ex) {
return Mono.error(ex);
}
});
}
private Object createHandler(URI url, WebSocketHandler handler, Sinks.Empty<Void> completion) {
Function<Session, JettyWebSocketSession> sessionFactory = session -> {
HandshakeInfo info = createHandshakeInfo(url, session);
return new JettyWebSocketSession(session, info, DefaultDataBufferFactory.sharedInstance, completion);
};
return new JettyWebSocketHandlerAdapter(handler, sessionFactory);
}
private HandshakeInfo createHandshakeInfo(URI url, Session jettySession) {
HttpHeaders headers = new HttpHeaders();
headers.putAll(jettySession.getUpgradeResponse().getHeaders());
String protocol = headers.getFirst("Sec-WebSocket-Protocol");
return new HandshakeInfo(url, headers, Mono.empty(), protocol);
}
}

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

@@ -21,8 +21,8 @@ import java.util.function.Supplier;
import jakarta.servlet.ServletContext;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import org.eclipse.jetty.websocket.server.JettyWebSocketCreator;
import org.eclipse.jetty.websocket.server.JettyWebSocketServerContainer;
import org.eclipse.jetty.ee10.websocket.server.JettyWebSocketCreator;
import org.eclipse.jetty.ee10.websocket.server.JettyWebSocketServerContainer;
import reactor.core.publisher.Mono;
import org.springframework.core.io.buffer.DataBufferFactory;

3
spring-webflux/src/test/java/org/springframework/web/reactive/socket/AbstractReactiveWebSocketIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2022 the original author or authors.
* Copyright 2002-2023 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.
@@ -93,7 +93,6 @@ abstract class AbstractReactiveWebSocketIntegrationTests {
@SuppressWarnings("removal")
WebSocketClient[] clients = new WebSocketClient[] {
new TomcatWebSocketClient(),
new org.springframework.web.reactive.socket.client.JettyWebSocketClient(),
new ReactorNettyWebSocketClient(),
new ReactorNetty2WebSocketClient(),
new UndertowWebSocketClient(Xnio.getInstance().createWorker(OptionMap.EMPTY))