async tls handshake in jrt

1 files changed, 37 insertions, 5 deletions

diff --git a/jrt/src/com/yahoo/jrt/Transport.java b/jrt/src/com/yahoo/jrt/Transport.java
index 0a2f2a4b7cb..33ce6fe6ed0 100644
--- a/jrt/src/com/yahoo/jrt/Transport.java
+++ b/jrt/src/com/yahoo/jrt/Transport.java
@@ -52,6 +52,12 @@ public class Transport {
         public void run() { handleEnableWrite(conn); }
     }
 
+    private class HandshakeWorkDoneCmd implements Runnable {
+        private Connection conn;
+        HandshakeWorkDoneCmd(Connection conn) { this.conn = conn; }
+        public void run() { handleHandshakeWorkDone(conn); }
+    }
+
     private class SyncCmd implements Runnable {
         boolean done = false;
         public synchronized void waitDone() {
@@ -73,7 +79,7 @@ public class Transport {
     private Queue             queue;
     private Queue             myQueue;
     private Connector         connector;
-    private Closer            closer;
+    private Worker            worker;
     private Scheduler         scheduler;
     private int               state;
     private Selector          selector;
@@ -82,7 +88,7 @@ public class Transport {
     private void handleAddConnection(Connection conn) {
         if (conn.isClosed()) {
             if (conn.hasSocket()) {
-                closer.closeLater(conn);
+                worker.closeLater(conn);
             }
             return;
         }
@@ -97,7 +103,7 @@ public class Transport {
         }
         conn.fini();
         if (conn.hasSocket()) {
-            closer.closeLater(conn);
+            worker.closeLater(conn);
         }
     }
 
@@ -108,6 +114,18 @@ public class Transport {
         conn.enableWrite();
     }
 
+    private void handleHandshakeWorkDone(Connection conn) {
+        if (conn.isClosed()) {
+            return;
+        }
+        try {
+            conn.handleHandshakeWorkDone();
+        } catch (IOException e) {
+            conn.setLostReason(e);
+            handleCloseConnection(conn);
+        }
+    }
+
     private boolean postCommand(Runnable cmd) {
         boolean wakeup;
         synchronized (this) {
@@ -174,7 +192,7 @@ public class Transport {
         queue     = new Queue();
         myQueue   = new Queue();
         connector = new Connector(this);
-        closer    = new Closer(this);
+        worker    = new Worker(this);
         scheduler = new Scheduler(System.currentTimeMillis());
         state     = OPEN;
         try {
@@ -287,6 +305,20 @@ public class Transport {
         }
     }
 
+    void handshakeWorkDone(Connection conn) {
+        postCommand(new HandshakeWorkDoneCmd(conn));
+    }
+
+    /**
+     * Request that {@link Connection#doHandshakeWork()} be called (in any thread)
+     * followed by a call to {@link Connection#handleHandshakeWorkDone()} from the transport thread.
+     *
+     * @param conn the connection needing handshake work
+     */
+    void doHandshakeWork(Connection conn) {
+        worker.doHandshakeWork(conn);
+    }
+
     /**
      * Create a {@link Task} that can be scheduled for execution in
      * the transport thread.
@@ -379,7 +411,7 @@ public class Transport {
             handleCloseConnection(conn);
         }
         try { selector.close(); } catch (Exception e) {}
-        closer.shutdown().join();
+        worker.shutdown().join();
         connector.exit().join();
         try { cryptoEngine.close(); } catch (Exception e) {}
     }