Skip to content

Commit e9cb9a9

Browse files
lachlan-robertslachlan-roberts
authored
Merge pull request #2 from webtide/JettyCoreHttpHandlerAdapter-WebSocket
Update spring Jetty websocket usage
2 parents d9202dc + cd9063b commit e9cb9a9

File tree

9 files changed

+357
-165
lines changed

9 files changed

+357
-165
lines changed

‎spring-web/src/main/java/org/springframework/http/server/reactive/JettyCoreServerHttpRequest.java

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -53,7 +53,6 @@
5353
* @author Greg Wilkins
5454
* @since 6.2
5555
*/
56-
// TODO: extend AbstractServerHttpRequest for websocket.
5756
class JettyCoreServerHttpRequest implements ServerHttpRequest {
5857
private static final MultiValueMap<String, String> EMPTY_QUERY = CollectionUtils.unmodifiableMultiValueMap(new LinkedMultiValueMap<>());
5958

‎spring-web/src/main/java/org/springframework/http/server/reactive/JettyCoreServerHttpResponse.java

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,7 +65,6 @@
6565
* @author Lachlan Roberts
6666
* @since 6.2
6767
*/
68-
// TODO: extend AbstractServerHttpResponse for websocket.
6968
class JettyCoreServerHttpResponse implements ServerHttpResponse, ZeroCopyHttpOutputMessage {
7069
private final AtomicBoolean committed = new AtomicBoolean(false);
7170

@@ -133,8 +132,8 @@ public Mono<Void> writeWith(Path file, long position, long count) {
133132
Callback.Completable callback = new Callback.Completable();
134133
mono = Mono.fromFuture(callback);
135134
try {
136-
// TODO: Why does intellij warn about possible blocking call?
137-
// Because it can block and we want to be fully asynchronous. Use AsynchronousFileChannel?
135+
// The method can block, but it is not expected to do so for any significant time.
136+
@SuppressWarnings("BlockingMethodInNonBlockingContext")
138137
SeekableByteChannel channel = Files.newByteChannel(file, StandardOpenOption.READ);
139138
new ContentWriterIteratingCallback(channel, position, count, this.response, callback).iterate();
140139
}

‎spring-webflux/spring-webflux.gradle

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@ dependencies {
2828
exclude group: "jakarta.servlet", module: "jakarta.servlet-api"
2929
}
3030
optional("org.eclipse.jetty.websocket:jetty-websocket-jetty-server")
31+
optional("org.eclipse.jetty.websocket:jetty-websocket-jetty-client")
3132
optional("org.freemarker:freemarker")
3233
optional("org.jetbrains.kotlin:kotlin-reflect")
3334
optional("org.jetbrains.kotlin:kotlin-stdlib")

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

Lines changed: 48 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -19,47 +19,39 @@
1919
import java.nio.ByteBuffer;
2020
import java.nio.charset.Charset;
2121
import java.nio.charset.StandardCharsets;
22+
import java.util.Objects;
2223
import java.util.function.Function;
2324
import java.util.function.IntPredicate;
2425

26+
import org.eclipse.jetty.util.BufferUtil;
2527
import org.eclipse.jetty.websocket.api.Callback;
26-
import org.eclipse.jetty.websocket.api.Frame;
2728
import org.eclipse.jetty.websocket.api.Session;
28-
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketClose;
29-
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketError;
30-
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketFrame;
31-
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketMessage;
32-
import org.eclipse.jetty.websocket.api.annotations.OnWebSocketOpen;
33-
import org.eclipse.jetty.websocket.api.annotations.WebSocket;
34-
import org.eclipse.jetty.websocket.core.OpCode;
29+
import org.reactivestreams.Subscriber;
30+
import org.reactivestreams.Subscription;
3531

3632
import org.springframework.core.io.buffer.CloseableDataBuffer;
3733
import org.springframework.core.io.buffer.DataBuffer;
3834
import org.springframework.core.io.buffer.DataBufferFactory;
39-
import org.springframework.lang.Nullable;
4035
import org.springframework.util.Assert;
4136
import org.springframework.web.reactive.socket.CloseStatus;
4237
import org.springframework.web.reactive.socket.WebSocketHandler;
4338
import org.springframework.web.reactive.socket.WebSocketMessage;
4439
import org.springframework.web.reactive.socket.WebSocketMessage.Type;
4540

4641
/**
47-
* Jetty {@link WebSocket @WebSocket} handler that delegates events to a
42+
* Jetty {@link org.eclipse.jetty.websocket.api.Session.Listener} handler that delegates events to a
4843
* reactive {@link WebSocketHandler} and its session.
4944
*
5045
* @author Violeta Georgieva
5146
* @author Rossen Stoyanchev
5247
* @since 5.0
5348
*/
54-
@WebSocket
55-
public class JettyWebSocketHandlerAdapter {
56-
private static final ByteBuffer EMPTY_PAYLOAD = ByteBuffer.wrap(new byte[0]);
57-
49+
public class JettyWebSocketHandlerAdapter implements Session.Listener {
5850
private final WebSocketHandler delegateHandler;
5951

6052
private final Function<Session, JettyWebSocketSession> sessionFactory;
6153

62-
@Nullable
54+
@SuppressWarnings("NotNullFieldNotInitialized")
6355
private JettyWebSocketSession delegateSession;
6456

6557
public JettyWebSocketHandlerAdapter(WebSocketHandler handler,
@@ -71,65 +63,63 @@ public JettyWebSocketHandlerAdapter(WebSocketHandler handler,
7163
this.sessionFactory = sessionFactory;
7264
}
7365

74-
@OnWebSocketOpen
66+
@Override
7567
public void onWebSocketOpen(Session session) {
76-
this.delegateSession = this.sessionFactory.apply(session);
68+
this.delegateSession = Objects.requireNonNull(this.sessionFactory.apply(session));
7769
this.delegateHandler.handle(this.delegateSession)
78-
.checkpoint(session.getUpgradeRequest().getRequestURI() + " [JettyWebSocketHandlerAdapter]")
79-
.subscribe(this.delegateSession);
70+
.subscribe(new Subscriber<>() {
71+
@Override
72+
public void onSubscribe(Subscription s) {
73+
s.request(Long.MAX_VALUE);
74+
}
75+
76+
@Override
77+
public void onNext(Void unused) {
78+
}
79+
80+
@Override
81+
public void onError(Throwable t) {
82+
delegateSession.onHandlerError(t);
83+
}
84+
85+
@Override
86+
public void onComplete() {
87+
delegateSession.onHandleComplete();
88+
}
89+
});
8090
}
8191

82-
@OnWebSocketMessage
92+
@Override
8393
public void onWebSocketText(String message) {
84-
if (this.delegateSession != null) {
85-
byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
86-
DataBuffer buffer = this.delegateSession.bufferFactory().wrap(bytes);
87-
WebSocketMessage webSocketMessage = new WebSocketMessage(Type.TEXT, buffer);
88-
this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
89-
}
94+
byte[] bytes = message.getBytes(StandardCharsets.UTF_8);
95+
DataBuffer buffer = this.delegateSession.bufferFactory().wrap(bytes);
96+
WebSocketMessage webSocketMessage = new WebSocketMessage(Type.TEXT, buffer);
97+
this.delegateSession.handleMessage(webSocketMessage);
9098
}
9199

92-
@OnWebSocketMessage
100+
@Override
93101
public void onWebSocketBinary(ByteBuffer byteBuffer, Callback callback) {
94-
if (this.delegateSession != null) {
95-
DataBuffer buffer = this.delegateSession.bufferFactory().wrap(byteBuffer);
96-
buffer = new JettyDataBuffer(buffer, callback);
97-
WebSocketMessage webSocketMessage = new WebSocketMessage(Type.BINARY, buffer);
98-
this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
99-
}
100-
else {
101-
callback.succeed();
102-
}
102+
DataBuffer buffer = this.delegateSession.bufferFactory().wrap(byteBuffer);
103+
buffer = new JettyDataBuffer(buffer, callback);
104+
WebSocketMessage webSocketMessage = new WebSocketMessage(Type.BINARY, buffer);
105+
this.delegateSession.handleMessage(webSocketMessage);
103106
}
104107

105-
@OnWebSocketFrame
106-
public void onWebSocketFrame(Frame frame, Callback callback) {
107-
if (this.delegateSession != null) {
108-
if (OpCode.PONG == frame.getOpCode()) {
109-
ByteBuffer byteBuffer = (frame.getPayload() != null ? frame.getPayload() : EMPTY_PAYLOAD);
110-
DataBuffer buffer = this.delegateSession.bufferFactory().wrap(byteBuffer);
111-
buffer = new JettyDataBuffer(buffer, callback);
112-
WebSocketMessage webSocketMessage = new WebSocketMessage(Type.PONG, buffer);
113-
this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
114-
return;
115-
}
116-
}
117-
118-
callback.succeed();
108+
@Override
109+
public void onWebSocketPong(ByteBuffer payload) {
110+
DataBuffer buffer = this.delegateSession.bufferFactory().wrap(BufferUtil.copy(payload));
111+
WebSocketMessage webSocketMessage = new WebSocketMessage(Type.PONG, buffer);
112+
this.delegateSession.handleMessage(webSocketMessage);
119113
}
120114

121-
@OnWebSocketClose
115+
@Override
122116
public void onWebSocketClose(int statusCode, String reason) {
123-
if (this.delegateSession != null) {
124-
this.delegateSession.handleClose(CloseStatus.create(statusCode, reason));
125-
}
117+
this.delegateSession.handleClose(CloseStatus.create(statusCode, reason));
126118
}
127119

128-
@OnWebSocketError
120+
@Override
129121
public void onWebSocketError(Throwable cause) {
130-
if (this.delegateSession != null) {
131-
this.delegateSession.handleError(cause);
132-
}
122+
this.delegateSession.handleError(cause);
133123
}
134124

135125

0 commit comments

Comments
 (0)