Skip to content

[5.2] Handle exceptions on joined WebSocket sockets #164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tsegismont merged 1 commit into from
Oct 20, 2026
Merged

[5.2] Handle exceptions on joined WebSocket sockets #164

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
128 changes: 71 additions & 57 deletions src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,38 +174,7 @@ public Future<ProxyResponse> sendRequest() {
return next.interceptor.handleProxyRequest(this);
} else {
if (isWebSocket) {
HttpServerRequest proxiedRequest = request().proxiedRequest();
return resolveOrigin(this).compose(request -> {
// The client and backend sockets are only joined once the backend answers 101: a client that goes away
// before that would leave the request to the backend open. The handler is set before the check, so a client
// closing in between is caught by one or the other.
HttpServerResponse clientResponse = proxiedRequest.response();
clientResponse.exceptionHandler(throwable -> request.reset(0L, throwable));
if (clientResponse.closed()) {
request.reset();
return Future.failedFuture(new HttpClosedException("Client connection closed"));
}
request.setMethod(request().getMethod());
request.setURI(request().getURI());
for (Map.Entry<String, String> header : request().headers()) {
String name = header.getKey();
if (name.equalsIgnoreCase(CONNECTION.toString())) {
// Firefox is known to send an unexpected connection header value
// Connection=keep-alive, Upgrade
// It leads to a failure in websocket proxying
// So we make sure the standard value is sent to the backend
request.headers().set(CONNECTION, UPGRADE);
} else if (!name.equalsIgnoreCase(HOST.toString())) {
request.headers().add(name, header.getValue());
}
}
Future<HttpClientResponse> responseFuture = request.connect();
ReadStream<Buffer> readStream = request().getBody().stream();
readStream.handler(request::write);
readStream.resume();
proxiedRequest.resume();
return responseFuture;
}).map(response -> new ProxiedResponse((ProxiedRequest) request(), request().proxiedRequest().response(), response));
return sendWebSocketRequest();
}
return sendProxyRequest(request);
}
Expand All @@ -226,31 +195,7 @@ public Future<Void> sendResponse() {
return previous.interceptor.handleProxyResponse(this);
} else {
if (isWebSocket) {
HttpClientResponse proxiedResponse = response().proxiedResponse();
if (response.getStatusCode() == 101) {
HttpServerResponse clientResponse = request().proxiedRequest().response();
clientResponse.setStatusCode(101);
clientResponse.headers().addAll(response.headers());
Future<NetSocket> otherso = request.proxiedRequest().toNetSocket();
otherso.onComplete(ar3 -> {
if (ar3.succeeded()) {
NetSocket responseSocket = ar3.result();
NetSocket proxyResponseSocket = proxiedResponse.netSocket();
responseSocket.handler(proxyResponseSocket::write);
proxyResponseSocket.handler(responseSocket::write);
responseSocket.closeHandler(v -> proxyResponseSocket.close());
proxyResponseSocket.closeHandler(v -> responseSocket.close());
} else {
// The client went away while the backend answered 101
log.trace("Could not upgrade the client connection", ar3.cause());
proxiedResponse.netSocket().close();
}
});
} else {
request().proxiedRequest().resume();
end(request(), proxiedResponse.statusCode());
}
return Future.succeededFuture();
return sendWebSocketResponse();
}
return response.send();
}
Expand All @@ -274,6 +219,75 @@ private Future<Void> sendProxyResponse(ProxyResponse response) {

return sendResponse();
}

private Future<ProxyResponse> sendWebSocketRequest() {
HttpServerRequest proxiedRequest = request().proxiedRequest();
return resolveOrigin(this).compose(request -> {
// The client and backend sockets are only joined once the backend answers 101: a client that goes away
// before that would leave the request to the backend open. The handler is set before the check, so a client
// closing in between is caught by one or the other.
HttpServerResponse clientResponse = proxiedRequest.response();
clientResponse.exceptionHandler(throwable -> request.reset(0L, throwable));
if (clientResponse.closed()) {
request.reset();
return Future.failedFuture(new HttpClosedException("Client connection closed"));
}
request.setMethod(request().getMethod());
request.setURI(request().getURI());
for (Map.Entry<String, String> header : request().headers()) {
String name = header.getKey();
if (name.equalsIgnoreCase(CONNECTION.toString())) {
request.headers().set(CONNECTION, UPGRADE);
} else if (name.equalsIgnoreCase(CONTENT_LENGTH.toString())) {
// A tunnel request cannot declare a content-length, the body (e.g. WebSocket V00 key3) is written
// to the tunnel instead and the backend infers its length from the WebSocket handshake headers
} else if (!name.equalsIgnoreCase(HOST.toString())) {
request.headers().add(name, header.getValue());
}
}
Future<HttpClientResponse> responseFuture = request.connect();
ReadStream<Buffer> readStream = request().getBody().stream();
readStream.handler(request::write);
readStream.resume();
proxiedRequest.resume();
return responseFuture;
}).map(response -> new ProxiedResponse((ProxiedRequest) request(), request().proxiedRequest().response(), response));
}

private Future<Void> sendWebSocketResponse() {
HttpClientResponse proxiedResponse = response().proxiedResponse();
if (response.getStatusCode() == 101) {
HttpServerResponse clientResponse = request().proxiedRequest().response();
clientResponse.setStatusCode(101);
clientResponse.headers().addAll(response.headers());
Future<NetSocket> otherso = request.proxiedRequest().toNetSocket();
otherso.onComplete(ar3 -> {
if (ar3.succeeded()) {
NetSocket responseSocket = ar3.result();
NetSocket proxyResponseSocket = proxiedResponse.netSocket();
responseSocket.handler(proxyResponseSocket::write);
proxyResponseSocket.handler(responseSocket::write);
responseSocket.closeHandler(v -> proxyResponseSocket.close());
proxyResponseSocket.closeHandler(v -> responseSocket.close());
responseSocket.exceptionHandler(t -> {
log.trace("Client WebSocket error", t);
proxyResponseSocket.close();
});
proxyResponseSocket.exceptionHandler(t -> {
log.trace("Backend WebSocket error", t);
responseSocket.close();
});
} else {
log.trace("Could not upgrade the client connection", ar3.cause());
proxiedResponse.netSocket().close();
}
});
} else {
request().proxiedRequest().resume();
end(request(), proxiedResponse.statusCode());
}
return Future.succeededFuture();
}
}

private static class ProxyInterceptorEntry {
Expand Down
42 changes: 42 additions & 0 deletions src/test/java/io/vertx/tests/WebSocketTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,12 @@ public void testWebSocketExtensionsNegotiatedBetweenClientAndBackend(TestContext
}));
}

private static final Buffer UPGRADE_RESPONSE = Buffer.buffer("HTTP/1.1 101 Switching Protocols\r\n" +
"Upgrade: websocket\r\n" +
"Connection: Upgrade\r\n" +
"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=\r\n" +
"\r\n");

private static final Buffer UPGRADE = Buffer.buffer("GET /ws HTTP/1.1\r\n" +
"Host: localhost:8080\r\n" +
"Connection: Upgrade\r\n" +
Expand All @@ -286,6 +292,42 @@ public void testClientCloseDuringUpgradeClosesBackendConnection(TestContext ctx)
}));
}

@Test
public void testClientCloseAfterUpgradeClosesBackendConnection(TestContext ctx) {
Async backendClosed = ctx.async();
SocketAddress backend = startNetBackend(ctx, 8081, so -> {
so.handler(buff -> {
so.write(UPGRADE_RESPONSE);
});
so.closeHandler(v -> backendClosed.complete());
});
startProxy(backend);
vertx.createNetClient().connect(8080, "localhost").onComplete(ctx.asyncAssertSuccess(so -> {
so.handler(buff -> {
// Upgrade response received by the client side, now close abruptly
so.close();
});
so.write(UPGRADE);
}));
}

@Test
public void testBackendCloseAfterUpgradeClosesClientConnection(TestContext ctx) {
Async clientClosed = ctx.async();
SocketAddress backend = startNetBackend(ctx, 8081, so -> {
so.handler(buff -> {
so.write(UPGRADE_RESPONSE);
// Close the backend side after upgrade completes
vertx.setTimer(100, id -> so.close());
});
});
startProxy(backend);
vertx.createNetClient().connect(8080, "localhost").onComplete(ctx.asyncAssertSuccess(so -> {
so.closeHandler(v -> clientClosed.complete());
so.write(UPGRADE);
}));
}

@Test
public void testClientCloseBeforeOriginResolvedSendsNoUpgrade(TestContext ctx) {
Async backendClosed = ctx.async();
Expand Down
Loading