Changes from review.

Signed-off-by: Lachlan Roberts <lachlan.p.roberts@gmail.com>

Author: lachlan-roberts

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

@@ -225,13 +225,13 @@ protected Action process() {
 @Override
 protected void onCompleteSuccess() {
 iterator.close();
-completable.complete(null);
+completable.succeed();
 }
 
 @Override
 protected void onCompleteFailure(Throwable cause) {
 iterator.close();
-completable.completeExceptionally(cause);
+completable.fail(cause);
 }
 }.iterate();
 }

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

@@ -78,7 +78,7 @@ public Mono<Void> execute(URI url, @Nullable HttpHeaders headers, WebSocketHandl
 public void onHandshakeResponse(Request request, Response response) {
 String protocol = response.getHeaders().get(HttpHeader.SEC_WEBSOCKET_SUBPROTOCOL);
 HttpHeaders responseHeaders = new HttpHeaders();
-response.getHeaders().forEach(header -> responseHeaders.addAll(header.getName(), header.getValueList()));
+response.getHeaders().forEach(header -> responseHeaders.add(header.getName(), header.getValue()));
 handshakeInfo.set(new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol));
 }
 };
@@ -89,6 +89,8 @@ public void onHandshakeResponse(Request request, Response response) {
 try {
 this.client.connect(handlerAdapter, url, upgradeRequest, jettyUpgradeListener)
 .whenComplete((session, throwable) -> {
+// Only fail the completion if we have an error
+// as the JettyWebSocketSession will never be opened.
 if (throwable != null) {
 completion.tryEmitError(throwable);
 }