diff --git a/src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java b/src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java index b60f972..5044540 100644 --- a/src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java +++ b/src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java @@ -176,6 +176,15 @@ public Future sendRequest() { 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 header : request().headers()) { @@ -232,9 +241,9 @@ public Future sendResponse() { responseSocket.closeHandler(v -> proxyResponseSocket.close()); proxyResponseSocket.closeHandler(v -> responseSocket.close()); } else { - // Find reproducer - System.err.println("Handle this case"); - ar3.cause().printStackTrace(); + // The client went away while the backend answered 101 + log.trace("Could not upgrade the client connection", ar3.cause()); + proxiedResponse.netSocket().close(); } }); } else { diff --git a/src/test/java/io/vertx/tests/WebSocketTest.java b/src/test/java/io/vertx/tests/WebSocketTest.java index 1f2b139..442c5ed 100644 --- a/src/test/java/io/vertx/tests/WebSocketTest.java +++ b/src/test/java/io/vertx/tests/WebSocketTest.java @@ -11,6 +11,7 @@ package io.vertx.tests; import io.vertx.core.Future; +import io.vertx.core.Promise; import io.vertx.core.buffer.Buffer; import io.vertx.core.http.*; import io.vertx.core.net.SocketAddress; @@ -260,4 +261,49 @@ public void testWebSocketExtensionsNegotiatedBetweenClientAndBackend(TestContext ws.writeTextMessage("hello"); })); } + + private static final Buffer UPGRADE = Buffer.buffer("GET /ws HTTP/1.1\r\n" + + "Host: localhost:8080\r\n" + + "Connection: Upgrade\r\n" + + "Upgrade: websocket\r\n" + + "Sec-WebSocket-Version: 13\r\n" + + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + + "\r\n"); + + @Test + public void testClientCloseDuringUpgradeClosesBackendConnection(TestContext ctx) { + Async backendClosed = ctx.async(); + Promise upgradeReceived = Promise.promise(); + SocketAddress backend = startNetBackend(ctx, 8081, so -> { + // Never answers the upgrade + so.handler(buff -> upgradeReceived.tryComplete()); + so.closeHandler(v -> backendClosed.complete()); + }); + startProxy(backend); + vertx.createNetClient().connect(8080, "localhost").onComplete(ctx.asyncAssertSuccess(so -> { + upgradeReceived.future().onComplete(ctx.asyncAssertSuccess(v -> so.close())); + so.write(UPGRADE); + })); + } + + @Test + public void testClientCloseBeforeOriginResolvedSendsNoUpgrade(TestContext ctx) { + Async backendClosed = ctx.async(); + Promise resolving = Promise.promise(); + SocketAddress backend = startNetBackend(ctx, 8081, so -> { + so.handler(buff -> ctx.fail("Upgrade sent for a client that has gone")); + so.closeHandler(v -> backendClosed.complete()); + }); + startProxy(proxy -> proxy.origin(OriginRequestProvider.selector(proxyContext -> { + Promise origin = Promise.promise(); + // Resolved only once the client has gone + proxyContext.request().proxiedRequest().connection().closeHandler(v -> origin.complete(backend)); + resolving.complete(); + return origin.future(); + }))); + vertx.createNetClient().connect(8080, "localhost").onComplete(ctx.asyncAssertSuccess(so -> { + resolving.future().onComplete(ctx.asyncAssertSuccess(v -> so.close())); + so.write(UPGRADE); + })); + } }