jdk-sandbox: comparison src/java.net.http/share/classes/jdk/internal/net/http/common/SSLFlowDelegate.java

equal deleted inserted replaced

-:d3aa93570779
+:a47b8125b7cc
 /*
-* Copyright (c) 2017, 2018, Oracle and/or its affiliates. All rights reserved.
+* Copyright (c) 2017, 2019, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.  Oracle designates this
 import javax.net.ssl.SSLEngine;
 import javax.net.ssl.SSLEngineResult;
 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
 import javax.net.ssl.SSLEngineResult.Status;
 import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLHandshakeException;
 import java.io.IOException;
 import java.lang.ref.Reference;
 import java.lang.ref.ReferenceQueue;
 import java.lang.ref.WeakReference;
 import java.nio.ByteBuffer;
 final CompletableFuture<String> alpnCF; // completes on initial handshake
 final Monitorable monitor = isMonitored ? this::monitor : null; // prevent GC until SSLFD is stopped
 volatile boolean close_notify_received;
 final CompletableFuture<Void> readerCF;
 final CompletableFuture<Void> writerCF;
+final CompletableFuture<Void> stopCF;
 final Consumer<ByteBuffer> recycler;
 static AtomicInteger scount = new AtomicInteger(1);
 final int id;
 /**
 this.handshakeState = new AtomicInteger(NOT_HANDSHAKING);
 this.readerCF = reader.completion();
 this.writerCF = reader.completion();
 readerCF.exceptionally(this::stopOnError);
 writerCF.exceptionally(this::stopOnError);
+this.stopCF = CompletableFuture.allOf(reader.completion(), writer.completion())
-CompletableFuture.allOf(reader.completion(), writer.completion())
 .thenRun(this::normalStop);
 this.alpnCF = new MinimalFuture<>();
 // connect the Reader to the downReader and the
 // Writer to the downWriter.
 @Override
 public String toString() {
 return "READER: " + super.toString() + ", readBuf: " + readBuf.toString()
 + ", count: " + count.toString() + ", scheduler: "
 + (scheduler.isStopped() ? "stopped" : "running")
-+ ", status: " + lastUnwrapStatus;
++ ", status: " + lastUnwrapStatus
++ ", handshakeState: " + handshakeState.get()
++ ", engine: " + engine.getHandshakeStatus();
 }
 private void reallocReadBuf() {
 int sz = readBuf.capacity();
 ByteBuffer newb = ByteBuffer.allocate(sz * 2);
 return;
 }
 if (complete && result.status() == Status.CLOSED) {
 if (debugr.on()) debugr.log("Closed: completing");
 outgoing(Utils.EMPTY_BB_LIST, true);
+// complete ALPN if not yet completed
+setALPN();
 return;
 }
 if (result.handshaking()) {
 handshaking = true;
 if (debugr.on()) debugr.log("handshaking");
 if (doHandshake(result, READER)) continue; // need unwrap
 else break; // doHandshake will have triggered the write scheduler if necessary
 } else {
-if ((handshakeState.getAndSet(NOT_HANDSHAKING) & ~DOING_TASKS) == HANDSHAKING) {
+if (trySetALPN()) {
-handshaking = false;
-applicationBufferSize = engine.getSession().getApplicationBufferSize();
-packetBufferSize = engine.getSession().getPacketBufferSize();
-setALPN();
 resumeActivity();
 }
 }
 } catch (IOException ex) {
 errorCommon(ex);
 if (result.status() == Status.CLOSED) {
 if (!upstreamCompleted) {
 upstreamCompleted = true;
 upstreamSubscription.cancel();
+// complete ALPN if not yet completed
+setALPN();
 }
 if (result.bytesProduced() <= 0)
 return;
 if (!completing && !completed) {
 if (result.handshaking()) {
 if (debugw.on()) debugw.log("handshaking");
 doHandshake(result, WRITER);  // ok to ignore return
 handshaking = true;
 } else {
-if ((handshakeState.getAndSet(NOT_HANDSHAKING) & ~DOING_TASKS) == HANDSHAKING) {
+if (trySetALPN()) {
-applicationBufferSize = engine.getSession().getApplicationBufferSize();
-packetBufferSize = engine.getSession().getPacketBufferSize();
-setALPN();
 resumeActivity();
 }
 }
 cleanList(writeList); // tidy up the source list
 sendResultBytes(result);
 if (stopped)
 return;
 stopped = true;
 reader.stop();
 writer.stop();
+// make sure the alpnCF is completed.
+if (!alpnCF.isDone()) {
+Throwable alpn = new SSLHandshakeException(
+"Connection closed before successful ALPN negotiation");
+alpnCF.completeExceptionally(alpn);
+}
 if (isMonitored) Monitor.remove(monitor);
 }
-private Void stopOnError(Throwable currentlyUnused) {
+private Void stopOnError(Throwable error) {
 // maybe log, etc
+// ensure the ALPN is completed
+// We could also do this in SSLTube.SSLSubscriberWrapper
+// onError/onComplete - with the caveat that the ALP CF
+// would get completed externally. Doing it here keeps
+// it all inside SSLFlowDelegate.
+if (!alpnCF.isDone()) {
+alpnCF.completeExceptionally(error);
+}
 normalStop();
 return null;
 }
 private void cleanList(List<ByteBuffer> l) {
 }
 break;
 }
 } while (true);
 if (debug.on()) debug.log("finished task execution");
+HandshakeStatus hs = engine.getHandshakeStatus();
+if (hs == HandshakeStatus.FINISHED || hs == HandshakeStatus.NOT_HANDSHAKING) {
+// We're no longer handshaking, try setting ALPN
+trySetALPN();
+}
 resumeActivity();
 } catch (Throwable t) {
 handleError(t);
 }
 });
+}
+boolean trySetALPN() {
+// complete ALPN CF if needed.
+if ((handshakeState.getAndSet(NOT_HANDSHAKING) & ~DOING_TASKS) == HANDSHAKING) {
+applicationBufferSize = engine.getSession().getApplicationBufferSize();
+packetBufferSize = engine.getSession().getPacketBufferSize();
+setALPN();
+return true;
+}
+return false;
 }
 // FIXME: acknowledge a received CLOSE request from peer
 EngineResult doClosure(EngineResult r) throws IOException {
 if (debug.on())

changeset 53350	a47b8125b7cc
parent 52902	e3398b2e1ab0
child 58649	6b6bf0de534b