Ensure the connection is closed when websocket handshake fails

src/main/java/reactor/netty/http/client/WebsocketClientOperations.java

@@ -112,21 +112,20 @@ public void onInboundNext(ChannelHandlerContext ctx, Object msg) {
 			setNettyResponse(response);
 
 			if (notRedirected(response)) {
-
-
 				try {
 					handshaker.finishHandshake(channel(), response);
 					listener().onStateChange(this, HttpClientState.RESPONSE_RECEIVED);
 				}
 				catch (Exception e) {
 					onInboundError(e);
+					//"FutureReturnValueIgnored" this is deliberate
+					ctx.close();
 				}
 				finally {
 					//Release unused content (101 status)
 					response.content()
 					        .release();
 				}
-
 			}
 			else {
 				response.content()

src/test/java/reactor/netty/http/client/WebsocketTest.java

@@ -25,6 +25,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
+import java.util.function.Function;
 import java.util.function.Predicate;
 
 import io.netty.buffer.Unpooled;
@@ -35,6 +37,7 @@
 import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
 import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakeException;
 import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
 import io.netty.handler.codec.http.websocketx.WebSocketFrame;
 import io.netty.handler.codec.http.websocketx.WebSocketHandshakeException;
@@ -1439,4 +1442,57 @@ private void doTestIssue970(boolean compress) {
 		            .verify(Duration.ofSeconds(30));
 		assertThat(clientHandler.get()).isEqualTo(compress);
 	}
+
+	@Test
+	void testConnectionClosedWhenFailedUpgrade_NoErrorHandling() throws Exception {
+		doTestConnectionClosedWhenFailedUpgrade(httpClient -> httpClient, null);
+	}
+
+	@Test
+	void testConnectionClosedWhenFailedUpgrade_ClientErrorHandling() throws Exception {
+		AtomicReference<Throwable> error = new AtomicReference<>();
+		doTestConnectionClosedWhenFailedUpgrade(
+				httpClient -> httpClient.doOnRequestError((req, t) -> error.set(t)), null);
+		assertThat(error.get()).isNotNull()
+				.isInstanceOf(WebSocketClientHandshakeException.class);
+		assertThat(((WebSocketClientHandshakeException) error.get()).response().status())
+				.isEqualTo(HttpResponseStatus.NOT_FOUND);
+	}
+
+	@Test
+	void testConnectionClosedWhenFailedUpgrade_PublisherErrorHandling() throws Exception {
+		AtomicReference<Throwable> error = new AtomicReference<>();
+		doTestConnectionClosedWhenFailedUpgrade(httpClient -> httpClient, error::set);
+		assertThat(error.get()).isNotNull()
+				.isInstanceOf(WebSocketClientHandshakeException.class);
+		assertThat(((WebSocketClientHandshakeException) error.get()).response().status())
+				.isEqualTo(HttpResponseStatus.NOT_FOUND);
+	}
+
+	private void doTestConnectionClosedWhenFailedUpgrade(
+			Function<HttpClient, HttpClient> clientCustomizer,
+			Consumer<Throwable> errorConsumer) throws Exception {
+		httpServer =
+				HttpServer.create()
+				          .port(0)
+				          .handle((req, res) -> res.sendNotFound())
+				          .wiretap(true)
+				          .bindNow();
+
+		CountDownLatch latch = new CountDownLatch(1);
+		HttpClient client =
+				HttpClient.create()
+				          .remoteAddress(httpServer::address)
+				          .wiretap(true)
+				          .tcpConfiguration(tcpClient -> tcpClient.doOnConnected(
+				                  conn -> conn.channel().closeFuture().addListener(f -> latch.countDown())));
+
+		clientCustomizer.apply(client)
+		                .websocket()
+		                .uri("/")
+		                .connect()
+		                .subscribe(null, errorConsumer, null);
+
+		assertThat(latch.await(5, TimeUnit.SECONDS)).as("latch await").isTrue();
+	}
 }