jdk/src/jdk.incubator.httpclient/share/classes/jdk/incubator/http/AsyncSSLDelegate.java - platform/libcore - Git at Google

 /*
  * Copyright (c) 2015, 2016, 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
  * particular file as subject to the "Classpath" exception as provided
  * by Oracle in the LICENSE file that accompanied this code.
  *
  * This code is distributed in the hope that it will be useful, but WITHOUT
  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  * version 2 for more details (a copy is included in the LICENSE file that
  * accompanied this code).
  *
  * You should have received a copy of the GNU General Public License version
  * 2 along with this work; if not, write to the Free Software Foundation,
  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  *
  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  * or visit www.oracle.com if you need additional information or have any
  * questions.
  */

 package jdk.incubator.http;

 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Semaphore;
 import java.util.function.Consumer;
 import java.util.function.Supplier;

 import static javax.net.ssl.SSLEngineResult.Status.*;
 import javax.net.ssl.*;

 import jdk.incubator.http.internal.common.AsyncWriteQueue;
 import jdk.incubator.http.internal.common.ByteBufferPool;
 import jdk.incubator.http.internal.common.ByteBufferReference;
 import jdk.incubator.http.internal.common.Log;
 import jdk.incubator.http.internal.common.Queue;
 import jdk.incubator.http.internal.common.Utils;
 import static javax.net.ssl.SSLEngineResult.HandshakeStatus.*;
 import jdk.incubator.http.internal.common.ExceptionallyCloseable;

 /**
  * Asynchronous wrapper around SSLEngine. send and receive is fully non
  * blocking. When handshaking is required, a thread is created to perform
  * the handshake and application level sends do not take place during this time.
  *
  * Is implemented using queues and functions operating on the receiving end
  * of each queue.
  *
  * Application writes to:
  *        ||
  *        \/
  *     appOutputQ
  *        ||
  *        \/
  * appOutputQ read by "upperWrite" method which does SSLEngine.wrap
  * and does async write to PlainHttpConnection
  *
  * Reading side is as follows
  * --------------------------
  *
  * "upperRead" method reads off channelInputQ and calls SSLEngine.unwrap and
  * when decrypted data is returned, it is passed to the user's Consumer<ByteBuffer>
  *        /\
  *        ||
  *     channelInputQ
  *        /\
  *        ||
  * "lowerRead" method puts buffers into channelInputQ. It is invoked from
  * OP_READ events from the selector.
  *
  * Whenever handshaking is required, the doHandshaking() method is called
  * which creates a thread to complete the handshake. It takes over the
  * channelInputQ from upperRead, and puts outgoing packets on channelOutputQ.
  * Selector events are delivered to lowerRead and lowerWrite as normal.
  *
  * Errors
  *
  * Any exception thrown by the engine or channel, causes all Queues to be closed
  * the channel to be closed, and the error is reported to the user's
  * Consumer<Throwable>
  */
 class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {

     // outgoing buffers put in this queue first and may remain here
     // while SSL handshaking happening.
     final AsyncWriteQueue appOutputQ = new AsyncWriteQueue(this::upperWrite);

     // queue of wrapped ByteBuffers waiting to be sent on socket channel
     //final Queue<ByteBuffer> channelOutputQ;

     // Bytes read into this queue before being unwrapped. Backup on this
     // Q should only happen when the engine is stalled due to delegated tasks
     final Queue<ByteBufferReference> channelInputQ;

     // input occurs through the read() method which is expected to be called
     // when the selector signals some data is waiting to be read. All incoming
     // handshake data is handled in this method, which means some calls to
     // read() may return zero bytes of user data. This is not a sign of spinning,
     // just that the handshake mechanics are being executed.

     final SSLEngine engine;
     final SSLParameters sslParameters;
     //final SocketChannel chan;
     final HttpConnection lowerOutput;
     final HttpClientImpl client;
     // should be volatile to provide proper synchronization(visibility) action
     volatile Consumer<ByteBufferReference> asyncReceiver;
     volatile Consumer<Throwable> errorHandler;

     // Locks.
     final Object reader = new Object();
     // synchronizing handshake state
     final Semaphore handshaker = new Semaphore(1);
     // flag set when frame or writer is blocked waiting for handshake to finish
     //boolean writerBlocked;
     //boolean readerBlocked;

     // alpn[] may be null. upcall is callback which receives incoming decoded bytes off socket

     AsyncSSLDelegate(HttpConnection lowerOutput, HttpClientImpl client, String[] alpn)
     {
         SSLContext context = client.sslContext();
         //channelOutputQ = new Queue<>();
         //channelOutputQ.registerPutCallback(this::lowerWrite);
         engine = context.createSSLEngine();
         engine.setUseClientMode(true);
         SSLParameters sslp = client.sslParameters()
                                    .orElseGet(context::getSupportedSSLParameters);
         sslParameters = Utils.copySSLParameters(sslp);
         if (alpn != null) {
             sslParameters.setApplicationProtocols(alpn);
         }
         logParams(sslParameters);
         engine.setSSLParameters(sslParameters);
         this.lowerOutput = lowerOutput;
         this.client = client;
         this.channelInputQ = new Queue<>();
         this.channelInputQ.registerPutCallback(this::upperRead);
     }

     @Override
     public void writeAsync(ByteBufferReference[] src) throws IOException {
         appOutputQ.put(src);
     }

     @Override
     public void writeAsyncUnordered(ByteBufferReference[] buffers) throws IOException {
         appOutputQ.putFirst(buffers);
     }

     @Override
     public void flushAsync() throws IOException {
         if (appOutputQ.flush()) {
             lowerOutput.flushAsync();
         }
     }

     @Override
     public void closeExceptionally(Throwable t) {
         Utils.close(t, appOutputQ, channelInputQ, lowerOutput);
     }

     @Override
     public void close() {
         Utils.close(appOutputQ, channelInputQ, lowerOutput);
     }

     // The code below can be uncommented to shake out
     // the implementation by inserting random delays and trigger
     // handshake in the SelectorManager thread (upperRead)
     // static final java.util.Random random =
     //    new java.util.Random(System.currentTimeMillis());

     /**
      * Attempts to wrap buffers from appOutputQ and place them on the
      * channelOutputQ for writing. If handshaking is happening, then the
      * process stalls and last buffers taken off the appOutputQ are put back
      * into it until handshaking completes.
      *
      * This same method is called to try and resume output after a blocking
      * handshaking operation has completed.
      */
     private void upperWrite(ByteBufferReference[] refs, AsyncWriteQueue delayCallback) {
         // currently delayCallback is not used. Use it when it's needed to execute handshake in another thread.
         try {
             ByteBuffer[] buffers = ByteBufferReference.toBuffers(refs);
             int bytes = Utils.remaining(buffers);
             while (bytes > 0) {
                 EngineResult r = wrapBuffers(buffers);
                 int bytesProduced = r.bytesProduced();
                 int bytesConsumed = r.bytesConsumed();
                 bytes -= bytesConsumed;
                 if (bytesProduced > 0) {
                     lowerOutput.writeAsync(new ByteBufferReference[]{r.destBuffer});
                 }

                 // The code below can be uncommented to shake out
                 // the implementation by inserting random delays and trigger
                 // handshake in the SelectorManager thread (upperRead)

                 // int sleep = random.nextInt(100);
                 // if (sleep > 20) {
                 //   Thread.sleep(sleep);
                 // }

                 // handshaking is happening or is needed
                 if (r.handshaking()) {
                     Log.logTrace("Write: needs handshake");
                     doHandshakeNow("Write");
                 }
             }
             ByteBufferReference.clear(refs);
         } catch (Throwable t) {
             closeExceptionally(t);
             errorHandler.accept(t);
         }
     }

     private void startHandshake(String tag) {
         Runnable run = () -> {
             try {
                 doHandshakeNow(tag);
             } catch (Throwable t) {
                 Log.logTrace("{0}: handshake failed: {1}", tag, t);
                 closeExceptionally(t);
                 errorHandler.accept(t);
             }
         };
         client.executor().execute(run);
     }

     private void doHandshakeNow(String tag)
         throws IOException, InterruptedException
     {
         handshaker.acquire();
         try {
             channelInputQ.registerPutCallback(null);
             lowerOutput.flushAsync();
             Log.logTrace("{0}: Starting handshake...", tag);
             doHandshakeImpl();
             Log.logTrace("{0}: Handshake completed", tag);
             channelInputQ.registerPutCallback(this::upperRead);
         } finally {
             handshaker.release();
         }
     }

      /**
      * Executes entire handshake in calling thread.
      * Returns after handshake is completed or error occurs
      */
     private void doHandshakeImpl() throws IOException {
         while (true) {
             SSLEngineResult.HandshakeStatus status = engine.getHandshakeStatus();
             switch(status) {
                 case NEED_TASK: {
                     List<Runnable> tasks = obtainTasks();
                     for (Runnable task : tasks) {
                         task.run();
                     }
                 } break;
                 case NEED_WRAP:
                     handshakeWrapAndSend();
                     break;
                 case NEED_UNWRAP: case NEED_UNWRAP_AGAIN:
                     handshakeReceiveAndUnWrap();
                     break;
                 case FINISHED: case NOT_HANDSHAKING:
                     return;
                 default:
                     throw new InternalError("Unexpected Handshake Status: "
                                              + status);
             }
         }
     }

     // acknowledge a received CLOSE request from peer
     void doClosure() throws IOException {
         //while (!wrapAndSend(emptyArray))
             //;
     }

     List<Runnable> obtainTasks() {
         List<Runnable> l = new ArrayList<>();
         Runnable r;
         while ((r = engine.getDelegatedTask()) != null) {
             l.add(r);
         }
         return l;
     }

     @Override
     public void setAsyncCallbacks(Consumer<ByteBufferReference> asyncReceiver,
                                   Consumer<Throwable> errorReceiver,
                                   Supplier<ByteBufferReference> readBufferSupplier) {
         this.asyncReceiver = asyncReceiver;
         this.errorHandler = errorReceiver;
         // readBufferSupplier is not used,
         // because of AsyncSSLDelegate has its own appBufferPool
     }

     @Override
     public void startReading() {
         // maybe this class does not need to implement AsyncConnection
     }

     static class EngineResult {
         final SSLEngineResult result;
         final ByteBufferReference destBuffer;


         // normal result
         EngineResult(SSLEngineResult result) {
             this(result, null);
         }

         EngineResult(SSLEngineResult result, ByteBufferReference destBuffer) {
             this.result = result;
             this.destBuffer = destBuffer;
         }

         boolean handshaking() {
             SSLEngineResult.HandshakeStatus s = result.getHandshakeStatus();
             return s != FINISHED && s != NOT_HANDSHAKING;
         }

         int bytesConsumed() {
             return result.bytesConsumed();
         }

         int bytesProduced() {
             return result.bytesProduced();
         }

         SSLEngineResult.HandshakeStatus handshakeStatus() {
             return result.getHandshakeStatus();
         }

         SSLEngineResult.Status status() {
             return result.getStatus();
         }
     }

     EngineResult handshakeWrapAndSend() throws IOException {
         EngineResult r = wrapBuffer(Utils.EMPTY_BYTEBUFFER);
         if (r.bytesProduced() > 0) {
             lowerOutput.writeAsync(new ByteBufferReference[]{r.destBuffer});
             lowerOutput.flushAsync();
         }
         return r;
     }

     // called during handshaking. It blocks until a complete packet
     // is available, unwraps it and returns.
     void handshakeReceiveAndUnWrap() throws IOException {
         ByteBufferReference ref = channelInputQ.take();
         while (true) {
             // block waiting for input
             EngineResult r = unwrapBuffer(ref.get());
             SSLEngineResult.Status status = r.status();
             if (status == BUFFER_UNDERFLOW) {
                 // wait for another buffer to arrive
                 ByteBufferReference ref1 = channelInputQ.take();
                 ref = combine (ref, ref1);
                 continue;
             }
             // OK
             // theoretically possible we could receive some user data
             if (r.bytesProduced() > 0) {
                 asyncReceiver.accept(r.destBuffer);
             } else {
                 r.destBuffer.clear();
             }
             // it is also possible that a delegated task could be needed
             // even though they are handled in the calling function
             if (r.handshakeStatus() == NEED_TASK) {
                 obtainTasks().stream().forEach((task) -> task.run());
             }

             if (!ref.get().hasRemaining()) {
                 ref.clear();
                 return;
             }
         }
     }

     EngineResult wrapBuffer(ByteBuffer src) throws SSLException {
         ByteBuffer[] bufs = new ByteBuffer[1];
         bufs[0] = src;
         return wrapBuffers(bufs);
     }

     private final ByteBufferPool netBufferPool = new ByteBufferPool();
     private final ByteBufferPool appBufferPool = new ByteBufferPool();

     /**
      * provides buffer of sslEngine@getPacketBufferSize().
      * used for encrypted buffers after wrap or before unwrap.
      * @return ByteBufferReference
      */
     public ByteBufferReference getNetBuffer() {
         return netBufferPool.get(engine.getSession().getPacketBufferSize());
     }

     /**
      * provides buffer of sslEngine@getApplicationBufferSize().
      * @return ByteBufferReference
      */
     private ByteBufferReference getAppBuffer() {
         return appBufferPool.get(engine.getSession().getApplicationBufferSize());
     }

     EngineResult wrapBuffers(ByteBuffer[] src) throws SSLException {
         ByteBufferReference dst = getNetBuffer();
         while (true) {
             SSLEngineResult sslResult = engine.wrap(src, dst.get());
             switch (sslResult.getStatus()) {
                 case BUFFER_OVERFLOW:
                     // Shouldn't happen. We allocated buffer with packet size
                     // get it again if net buffer size was changed
                     dst = getNetBuffer();
                     break;
                 case CLOSED:
                 case OK:
                     dst.get().flip();
                     return new EngineResult(sslResult, dst);
                 case BUFFER_UNDERFLOW:
                     // Shouldn't happen.  Doesn't returns when wrap()
                     // underflow handled externally
                     return new EngineResult(sslResult);
                 default:
                     assert false;
             }
         }
     }

     EngineResult unwrapBuffer(ByteBuffer srcbuf) throws IOException {
         ByteBufferReference dst = getAppBuffer();
         while (true) {
             SSLEngineResult sslResult = engine.unwrap(srcbuf, dst.get());
             switch (sslResult.getStatus()) {
                 case BUFFER_OVERFLOW:
                     // may happen only if app size buffer was changed.
                     // get it again if app buffer size changed
                     dst = getAppBuffer();
                     break;
                 case CLOSED:
                     doClosure();
                     throw new IOException("Engine closed");
                 case BUFFER_UNDERFLOW:
                     dst.clear();
                     return new EngineResult(sslResult);
                 case OK:
                      dst.get().flip();
                      return new EngineResult(sslResult, dst);
             }
         }
     }

     /**
      * Asynchronous read input. Call this when selector fires.
      * Unwrap done in upperRead because it also happens in
      * doHandshake() when handshake taking place
      */
     public void asyncReceive(ByteBufferReference buffer) {
         try {
             channelInputQ.put(buffer);
         } catch (Throwable t) {
             closeExceptionally(t);
             errorHandler.accept(t);
         }
     }

     private ByteBufferReference pollInput() throws IOException {
         return channelInputQ.poll();
     }

     private ByteBufferReference pollInput(ByteBufferReference next) throws IOException {
         return next == null ? channelInputQ.poll() : next;
     }

     public void upperRead() {
         ByteBufferReference src;
         ByteBufferReference next = null;
         synchronized (reader) {
             try {
                 src = pollInput();
                 if (src == null) {
                     return;
                 }
                 while (true) {
                     EngineResult r = unwrapBuffer(src.get());
                     switch (r.result.getStatus()) {
                         case BUFFER_UNDERFLOW:
                             // Buffer too small. Need to combine with next buf
                             next = pollInput(next);
                             if (next == null) {
                                 // no data available.
                                 // push buffer back until more data available
                                 channelInputQ.pushback(src);
                                 return;
                             } else {
                                 src = shift(src, next);
                                 if (!next.get().hasRemaining()) {
                                     next.clear();
                                     next = null;
                                 }
                             }
                             break;
                         case OK:
                             // check for any handshaking work
                             if (r.handshaking()) {
                                 // handshaking is happening or is needed
                                 // so we put the buffer back on Q to process again
                                 // later.
                                 Log.logTrace("Read: needs handshake");
                                 channelInputQ.pushback(src);
                                 startHandshake("Read");
                                 return;
                             }
                             asyncReceiver.accept(r.destBuffer);
                     }
                     if (src.get().hasRemaining()) {
                         continue;
                     }
                     src.clear();
                     src = pollInput(next);
                     next = null;
                     if (src == null) {
                         return;
                     }
                 }
             } catch (Throwable t) {
                 closeExceptionally(t);
                 errorHandler.accept(t);
             }
         }
     }

     ByteBufferReference shift(ByteBufferReference ref1, ByteBufferReference ref2) {
         ByteBuffer buf1 = ref1.get();
         if (buf1.capacity() < engine.getSession().getPacketBufferSize()) {
             ByteBufferReference newRef = getNetBuffer();
             ByteBuffer newBuf = newRef.get();
             newBuf.put(buf1);
             buf1 = newBuf;
             ref1.clear();
             ref1 = newRef;
         } else {
             buf1.compact();
         }
         ByteBuffer buf2 = ref2.get();
         Utils.copy(buf2, buf1, Math.min(buf1.remaining(), buf2.remaining()));
         buf1.flip();
         return ref1;
     }


     ByteBufferReference combine(ByteBufferReference ref1, ByteBufferReference ref2) {
         ByteBuffer buf1 = ref1.get();
         ByteBuffer buf2 = ref2.get();
         int avail1 = buf1.capacity() - buf1.remaining();
         if (buf2.remaining() < avail1) {
             buf1.compact();
             buf1.put(buf2);
             buf1.flip();
             ref2.clear();
             return ref1;
         }
         int newsize = buf1.remaining() + buf2.remaining();
         ByteBuffer newbuf = ByteBuffer.allocate(newsize); // getting rid of buffer pools
         newbuf.put(buf1);
         newbuf.put(buf2);
         newbuf.flip();
         ref1.clear();
         ref2.clear();
         return ByteBufferReference.of(newbuf);
     }

     SSLParameters getSSLParameters() {
         return sslParameters;
     }

     static void logParams(SSLParameters p) {
         if (!Log.ssl()) {
             return;
         }

         if (p == null) {
             Log.logSSL("SSLParameters: Null params");
             return;
         }

         final StringBuilder sb = new StringBuilder("SSLParameters:");
         final List<Object> params = new ArrayList<>();
         if (p.getCipherSuites() != null) {
             for (String cipher : p.getCipherSuites()) {
                 sb.append("\n    cipher: {")
                   .append(params.size()).append("}");
                 params.add(cipher);
             }
         }

         // SSLParameters.getApplicationProtocols() can't return null
         for (String approto : p.getApplicationProtocols()) {
             sb.append("\n    application protocol: {")
               .append(params.size()).append("}");
             params.add(approto);
         }

         if (p.getProtocols() != null) {
             for (String protocol : p.getProtocols()) {
                 sb.append("\n    protocol: {")
                   .append(params.size()).append("}");
                 params.add(protocol);
             }
         }

         if (p.getServerNames() != null) {
             for (SNIServerName sname : p.getServerNames()) {
                  sb.append("\n    server name: {")
                   .append(params.size()).append("}");
                 params.add(sname.toString());
             }
         }
         sb.append('\n');

         Log.logSSL(sb.toString(), params.toArray());
     }

     String getSessionInfo() {
         StringBuilder sb = new StringBuilder();
         String application = engine.getApplicationProtocol();
         SSLSession sess = engine.getSession();
         String cipher = sess.getCipherSuite();
         String protocol = sess.getProtocol();
         sb.append("Handshake complete alpn: ")
           .append(application)
           .append(", Cipher: ")
           .append(cipher)
           .append(", Protocol: ")
           .append(protocol);
         return sb.toString();
     }
 }
	/*
	* Copyright (c) 2015, 2016, 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
	* particular file as subject to the "Classpath" exception as provided
	* by Oracle in the LICENSE file that accompanied this code.
	*
	* This code is distributed in the hope that it will be useful, but WITHOUT
	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
	* version 2 for more details (a copy is included in the LICENSE file that
	* accompanied this code).
	*
	* You should have received a copy of the GNU General Public License version
	* 2 along with this work; if not, write to the Free Software Foundation,
	* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
	*
	* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
	* or visit www.oracle.com if you need additional information or have any
	* questions.
	*/

	package jdk.incubator.http;

	import java.io.IOException;
	import java.nio.ByteBuffer;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.concurrent.Semaphore;
	import java.util.function.Consumer;
	import java.util.function.Supplier;

	import static javax.net.ssl.SSLEngineResult.Status.*;
	import javax.net.ssl.*;

	import jdk.incubator.http.internal.common.AsyncWriteQueue;
	import jdk.incubator.http.internal.common.ByteBufferPool;
	import jdk.incubator.http.internal.common.ByteBufferReference;
	import jdk.incubator.http.internal.common.Log;
	import jdk.incubator.http.internal.common.Queue;
	import jdk.incubator.http.internal.common.Utils;
	import static javax.net.ssl.SSLEngineResult.HandshakeStatus.*;
	import jdk.incubator.http.internal.common.ExceptionallyCloseable;

	/**
	* Asynchronous wrapper around SSLEngine. send and receive is fully non
	* blocking. When handshaking is required, a thread is created to perform
	* the handshake and application level sends do not take place during this time.
	*
	* Is implemented using queues and functions operating on the receiving end
	* of each queue.
	*
	* Application writes to:
	* \|\|
	* \/
	* appOutputQ
	* \|\|
	* \/
	* appOutputQ read by "upperWrite" method which does SSLEngine.wrap
	* and does async write to PlainHttpConnection
	*
	* Reading side is as follows
	* --------------------------
	*
	* "upperRead" method reads off channelInputQ and calls SSLEngine.unwrap and
	* when decrypted data is returned, it is passed to the user's Consumer<ByteBuffer>
	* /\
	* \|\|
	* channelInputQ
	* /\
	* \|\|
	* "lowerRead" method puts buffers into channelInputQ. It is invoked from
	* OP_READ events from the selector.
	*
	* Whenever handshaking is required, the doHandshaking() method is called
	* which creates a thread to complete the handshake. It takes over the
	* channelInputQ from upperRead, and puts outgoing packets on channelOutputQ.
	* Selector events are delivered to lowerRead and lowerWrite as normal.
	*
	* Errors
	*
	* Any exception thrown by the engine or channel, causes all Queues to be closed
	* the channel to be closed, and the error is reported to the user's
	* Consumer<Throwable>
	*/
	class AsyncSSLDelegate implements ExceptionallyCloseable, AsyncConnection {

	// outgoing buffers put in this queue first and may remain here
	// while SSL handshaking happening.
	final AsyncWriteQueue appOutputQ = new AsyncWriteQueue(this::upperWrite);

	// queue of wrapped ByteBuffers waiting to be sent on socket channel
	//final Queue<ByteBuffer> channelOutputQ;

	// Bytes read into this queue before being unwrapped. Backup on this
	// Q should only happen when the engine is stalled due to delegated tasks
	final Queue<ByteBufferReference> channelInputQ;

	// input occurs through the read() method which is expected to be called
	// when the selector signals some data is waiting to be read. All incoming
	// handshake data is handled in this method, which means some calls to
	// read() may return zero bytes of user data. This is not a sign of spinning,
	// just that the handshake mechanics are being executed.

	final SSLEngine engine;
	final SSLParameters sslParameters;
	//final SocketChannel chan;
	final HttpConnection lowerOutput;
	final HttpClientImpl client;
	// should be volatile to provide proper synchronization(visibility) action
	volatile Consumer<ByteBufferReference> asyncReceiver;
	volatile Consumer<Throwable> errorHandler;

	// Locks.
	final Object reader = new Object();
	// synchronizing handshake state
	final Semaphore handshaker = new Semaphore(1);
	// flag set when frame or writer is blocked waiting for handshake to finish
	//boolean writerBlocked;
	//boolean readerBlocked;

	// alpn[] may be null. upcall is callback which receives incoming decoded bytes off socket

	AsyncSSLDelegate(HttpConnection lowerOutput, HttpClientImpl client, String[] alpn)
	{
	SSLContext context = client.sslContext();
	//channelOutputQ = new Queue<>();
	//channelOutputQ.registerPutCallback(this::lowerWrite);
	engine = context.createSSLEngine();
	engine.setUseClientMode(true);
	SSLParameters sslp = client.sslParameters()
	.orElseGet(context::getSupportedSSLParameters);
	sslParameters = Utils.copySSLParameters(sslp);
	if (alpn != null) {
	sslParameters.setApplicationProtocols(alpn);
	}
	logParams(sslParameters);
	engine.setSSLParameters(sslParameters);
	this.lowerOutput = lowerOutput;
	this.client = client;
	this.channelInputQ = new Queue<>();
	this.channelInputQ.registerPutCallback(this::upperRead);
	}

	@Override
	public void writeAsync(ByteBufferReference[] src) throws IOException {
	appOutputQ.put(src);
	}

	@Override
	public void writeAsyncUnordered(ByteBufferReference[] buffers) throws IOException {
	appOutputQ.putFirst(buffers);
	}

	@Override
	public void flushAsync() throws IOException {
	if (appOutputQ.flush()) {
	lowerOutput.flushAsync();
	}
	}

	@Override
	public void closeExceptionally(Throwable t) {
	Utils.close(t, appOutputQ, channelInputQ, lowerOutput);
	}

	@Override
	public void close() {
	Utils.close(appOutputQ, channelInputQ, lowerOutput);
	}

	// The code below can be uncommented to shake out
	// the implementation by inserting random delays and trigger
	// handshake in the SelectorManager thread (upperRead)
	// static final java.util.Random random =
	// new java.util.Random(System.currentTimeMillis());

	/**
	* Attempts to wrap buffers from appOutputQ and place them on the
	* channelOutputQ for writing. If handshaking is happening, then the
	* process stalls and last buffers taken off the appOutputQ are put back
	* into it until handshaking completes.
	*
	* This same method is called to try and resume output after a blocking
	* handshaking operation has completed.
	*/
	private void upperWrite(ByteBufferReference[] refs, AsyncWriteQueue delayCallback) {
	// currently delayCallback is not used. Use it when it's needed to execute handshake in another thread.
	try {
	ByteBuffer[] buffers = ByteBufferReference.toBuffers(refs);
	int bytes = Utils.remaining(buffers);
	while (bytes > 0) {
	EngineResult r = wrapBuffers(buffers);
	int bytesProduced = r.bytesProduced();
	int bytesConsumed = r.bytesConsumed();
	bytes -= bytesConsumed;
	if (bytesProduced > 0) {
	lowerOutput.writeAsync(new ByteBufferReference[]{r.destBuffer});
	}

	// The code below can be uncommented to shake out
	// the implementation by inserting random delays and trigger
	// handshake in the SelectorManager thread (upperRead)

	// int sleep = random.nextInt(100);
	// if (sleep > 20) {
	// Thread.sleep(sleep);
	// }

	// handshaking is happening or is needed
	if (r.handshaking()) {
	Log.logTrace("Write: needs handshake");
	doHandshakeNow("Write");
	}
	}
	ByteBufferReference.clear(refs);
	} catch (Throwable t) {
	closeExceptionally(t);
	errorHandler.accept(t);
	}
	}

	private void startHandshake(String tag) {
	Runnable run = () -> {
	try {
	doHandshakeNow(tag);
	} catch (Throwable t) {
	Log.logTrace("{0}: handshake failed: {1}", tag, t);
	closeExceptionally(t);
	errorHandler.accept(t);
	}
	};
	client.executor().execute(run);
	}

	private void doHandshakeNow(String tag)
	throws IOException, InterruptedException
	{
	handshaker.acquire();
	try {
	channelInputQ.registerPutCallback(null);
	lowerOutput.flushAsync();
	Log.logTrace("{0}: Starting handshake...", tag);
	doHandshakeImpl();
	Log.logTrace("{0}: Handshake completed", tag);
	channelInputQ.registerPutCallback(this::upperRead);
	} finally {
	handshaker.release();
	}
	}

	/**
	* Executes entire handshake in calling thread.
	* Returns after handshake is completed or error occurs
	*/
	private void doHandshakeImpl() throws IOException {
	while (true) {
	SSLEngineResult.HandshakeStatus status = engine.getHandshakeStatus();
	switch(status) {
	case NEED_TASK: {
	List<Runnable> tasks = obtainTasks();
	for (Runnable task : tasks) {
	task.run();
	}
	} break;
	case NEED_WRAP:
	handshakeWrapAndSend();
	break;
	case NEED_UNWRAP: case NEED_UNWRAP_AGAIN:
	handshakeReceiveAndUnWrap();
	break;
	case FINISHED: case NOT_HANDSHAKING:
	return;
	default:
	throw new InternalError("Unexpected Handshake Status: "
	+ status);
	}
	}
	}

	// acknowledge a received CLOSE request from peer
	void doClosure() throws IOException {
	//while (!wrapAndSend(emptyArray))
	//;
	}

	List<Runnable> obtainTasks() {
	List<Runnable> l = new ArrayList<>();
	Runnable r;
	while ((r = engine.getDelegatedTask()) != null) {
	l.add(r);
	}
	return l;
	}

	@Override
	public void setAsyncCallbacks(Consumer<ByteBufferReference> asyncReceiver,
	Consumer<Throwable> errorReceiver,
	Supplier<ByteBufferReference> readBufferSupplier) {
	this.asyncReceiver = asyncReceiver;
	this.errorHandler = errorReceiver;
	// readBufferSupplier is not used,
	// because of AsyncSSLDelegate has its own appBufferPool
	}

	@Override
	public void startReading() {
	// maybe this class does not need to implement AsyncConnection
	}

	static class EngineResult {
	final SSLEngineResult result;
	final ByteBufferReference destBuffer;


	// normal result
	EngineResult(SSLEngineResult result) {
	this(result, null);
	}

	EngineResult(SSLEngineResult result, ByteBufferReference destBuffer) {
	this.result = result;
	this.destBuffer = destBuffer;
	}

	boolean handshaking() {
	SSLEngineResult.HandshakeStatus s = result.getHandshakeStatus();
	return s != FINISHED && s != NOT_HANDSHAKING;
	}

	int bytesConsumed() {
	return result.bytesConsumed();
	}

	int bytesProduced() {
	return result.bytesProduced();
	}

	SSLEngineResult.HandshakeStatus handshakeStatus() {
	return result.getHandshakeStatus();
	}

	SSLEngineResult.Status status() {
	return result.getStatus();
	}
	}

	EngineResult handshakeWrapAndSend() throws IOException {
	EngineResult r = wrapBuffer(Utils.EMPTY_BYTEBUFFER);
	if (r.bytesProduced() > 0) {
	lowerOutput.writeAsync(new ByteBufferReference[]{r.destBuffer});
	lowerOutput.flushAsync();
	}
	return r;
	}

	// called during handshaking. It blocks until a complete packet
	// is available, unwraps it and returns.
	void handshakeReceiveAndUnWrap() throws IOException {
	ByteBufferReference ref = channelInputQ.take();
	while (true) {
	// block waiting for input
	EngineResult r = unwrapBuffer(ref.get());
	SSLEngineResult.Status status = r.status();
	if (status == BUFFER_UNDERFLOW) {
	// wait for another buffer to arrive
	ByteBufferReference ref1 = channelInputQ.take();
	ref = combine (ref, ref1);
	continue;
	}
	// OK
	// theoretically possible we could receive some user data
	if (r.bytesProduced() > 0) {
	asyncReceiver.accept(r.destBuffer);
	} else {
	r.destBuffer.clear();
	}
	// it is also possible that a delegated task could be needed
	// even though they are handled in the calling function
	if (r.handshakeStatus() == NEED_TASK) {
	obtainTasks().stream().forEach((task) -> task.run());
	}

	if (!ref.get().hasRemaining()) {
	ref.clear();
	return;
	}
	}
	}

	EngineResult wrapBuffer(ByteBuffer src) throws SSLException {
	ByteBuffer[] bufs = new ByteBuffer[1];
	bufs[0] = src;
	return wrapBuffers(bufs);
	}

	private final ByteBufferPool netBufferPool = new ByteBufferPool();
	private final ByteBufferPool appBufferPool = new ByteBufferPool();

	/**
	* provides buffer of sslEngine@getPacketBufferSize().
	* used for encrypted buffers after wrap or before unwrap.
	* @return ByteBufferReference
	*/
	public ByteBufferReference getNetBuffer() {
	return netBufferPool.get(engine.getSession().getPacketBufferSize());
	}

	/**
	* provides buffer of sslEngine@getApplicationBufferSize().
	* @return ByteBufferReference
	*/
	private ByteBufferReference getAppBuffer() {
	return appBufferPool.get(engine.getSession().getApplicationBufferSize());
	}

	EngineResult wrapBuffers(ByteBuffer[] src) throws SSLException {
	ByteBufferReference dst = getNetBuffer();
	while (true) {
	SSLEngineResult sslResult = engine.wrap(src, dst.get());
	switch (sslResult.getStatus()) {
	case BUFFER_OVERFLOW:
	// Shouldn't happen. We allocated buffer with packet size
	// get it again if net buffer size was changed
	dst = getNetBuffer();
	break;
	case CLOSED:
	case OK:
	dst.get().flip();
	return new EngineResult(sslResult, dst);
	case BUFFER_UNDERFLOW:
	// Shouldn't happen. Doesn't returns when wrap()
	// underflow handled externally
	return new EngineResult(sslResult);
	default:
	assert false;
	}
	}
	}

	EngineResult unwrapBuffer(ByteBuffer srcbuf) throws IOException {
	ByteBufferReference dst = getAppBuffer();
	while (true) {
	SSLEngineResult sslResult = engine.unwrap(srcbuf, dst.get());
	switch (sslResult.getStatus()) {
	case BUFFER_OVERFLOW:
	// may happen only if app size buffer was changed.
	// get it again if app buffer size changed
	dst = getAppBuffer();
	break;
	case CLOSED:
	doClosure();
	throw new IOException("Engine closed");
	case BUFFER_UNDERFLOW:
	dst.clear();
	return new EngineResult(sslResult);
	case OK:
	dst.get().flip();
	return new EngineResult(sslResult, dst);
	}
	}
	}

	/**
	* Asynchronous read input. Call this when selector fires.
	* Unwrap done in upperRead because it also happens in
	* doHandshake() when handshake taking place
	*/
	public void asyncReceive(ByteBufferReference buffer) {
	try {
	channelInputQ.put(buffer);
	} catch (Throwable t) {
	closeExceptionally(t);
	errorHandler.accept(t);
	}
	}

	private ByteBufferReference pollInput() throws IOException {
	return channelInputQ.poll();
	}

	private ByteBufferReference pollInput(ByteBufferReference next) throws IOException {
	return next == null ? channelInputQ.poll() : next;
	}

	public void upperRead() {
	ByteBufferReference src;
	ByteBufferReference next = null;
	synchronized (reader) {
	try {
	src = pollInput();
	if (src == null) {
	return;
	}
	while (true) {
	EngineResult r = unwrapBuffer(src.get());
	switch (r.result.getStatus()) {
	case BUFFER_UNDERFLOW:
	// Buffer too small. Need to combine with next buf
	next = pollInput(next);
	if (next == null) {
	// no data available.
	// push buffer back until more data available
	channelInputQ.pushback(src);
	return;
	} else {
	src = shift(src, next);
	if (!next.get().hasRemaining()) {
	next.clear();
	next = null;
	}
	}
	break;
	case OK:
	// check for any handshaking work
	if (r.handshaking()) {
	// handshaking is happening or is needed
	// so we put the buffer back on Q to process again
	// later.
	Log.logTrace("Read: needs handshake");
	channelInputQ.pushback(src);
	startHandshake("Read");
	return;
	}
	asyncReceiver.accept(r.destBuffer);
	}
	if (src.get().hasRemaining()) {
	continue;
	}
	src.clear();
	src = pollInput(next);
	next = null;
	if (src == null) {
	return;
	}
	}
	} catch (Throwable t) {
	closeExceptionally(t);
	errorHandler.accept(t);
	}
	}
	}

	ByteBufferReference shift(ByteBufferReference ref1, ByteBufferReference ref2) {
	ByteBuffer buf1 = ref1.get();
	if (buf1.capacity() < engine.getSession().getPacketBufferSize()) {
	ByteBufferReference newRef = getNetBuffer();
	ByteBuffer newBuf = newRef.get();
	newBuf.put(buf1);
	buf1 = newBuf;
	ref1.clear();
	ref1 = newRef;
	} else {
	buf1.compact();
	}
	ByteBuffer buf2 = ref2.get();
	Utils.copy(buf2, buf1, Math.min(buf1.remaining(), buf2.remaining()));
	buf1.flip();
	return ref1;
	}


	ByteBufferReference combine(ByteBufferReference ref1, ByteBufferReference ref2) {
	ByteBuffer buf1 = ref1.get();
	ByteBuffer buf2 = ref2.get();
	int avail1 = buf1.capacity() - buf1.remaining();
	if (buf2.remaining() < avail1) {
	buf1.compact();
	buf1.put(buf2);
	buf1.flip();
	ref2.clear();
	return ref1;
	}
	int newsize = buf1.remaining() + buf2.remaining();
	ByteBuffer newbuf = ByteBuffer.allocate(newsize); // getting rid of buffer pools
	newbuf.put(buf1);
	newbuf.put(buf2);
	newbuf.flip();
	ref1.clear();
	ref2.clear();
	return ByteBufferReference.of(newbuf);
	}

	SSLParameters getSSLParameters() {
	return sslParameters;
	}

	static void logParams(SSLParameters p) {
	if (!Log.ssl()) {
	return;
	}

	if (p == null) {
	Log.logSSL("SSLParameters: Null params");
	return;
	}

	final StringBuilder sb = new StringBuilder("SSLParameters:");
	final List<Object> params = new ArrayList<>();
	if (p.getCipherSuites() != null) {
	for (String cipher : p.getCipherSuites()) {
	sb.append("\n cipher: {")
	.append(params.size()).append("}");
	params.add(cipher);
	}
	}

	// SSLParameters.getApplicationProtocols() can't return null
	for (String approto : p.getApplicationProtocols()) {
	sb.append("\n application protocol: {")
	.append(params.size()).append("}");
	params.add(approto);
	}

	if (p.getProtocols() != null) {
	for (String protocol : p.getProtocols()) {
	sb.append("\n protocol: {")
	.append(params.size()).append("}");
	params.add(protocol);
	}
	}

	if (p.getServerNames() != null) {
	for (SNIServerName sname : p.getServerNames()) {
	sb.append("\n server name: {")
	.append(params.size()).append("}");
	params.add(sname.toString());
	}
	}
	sb.append('\n');

	Log.logSSL(sb.toString(), params.toArray());
	}

	String getSessionInfo() {
	StringBuilder sb = new StringBuilder();
	String application = engine.getApplicationProtocol();
	SSLSession sess = engine.getSession();
	String cipher = sess.getCipherSuite();
	String protocol = sess.getProtocol();
	sb.append("Handshake complete alpn: ")
	.append(application)
	.append(", Cipher: ")
	.append(cipher)
	.append(", Protocol: ")
	.append(protocol);
	return sb.toString();
	}
	}