// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.jrt;


class Connector {

    private class Run implements Runnable {
        public void run() {
            try {
                Connector.this.run();
            } catch (Throwable problem) {
                parent.handleFailure(problem, Connector.this);
            }
        }
    }

    private Thread      thread = new Thread(new Run(), "<jrt-connector>");
    private Transport   parent;
    private ThreadQueue connectQueue = new ThreadQueue();
    private boolean     done = false;
    private boolean     exit = false;

    public Connector(Transport parent) {
        this.parent = parent;
        thread.setDaemon(true);
        thread.start();
    }

    public void connectLater(Connection c) {
        if ( ! connectQueue.enqueue(c)) {
            c.transportThread().addConnection(c);
        }
    }

    private void run() {
        try {
            while (true) {
                Connection conn = (Connection) connectQueue.dequeue();
                conn.transportThread().addConnection(conn.connect());
            }
        } catch (EndOfQueueException e) {}
        synchronized (this) {
            done = true;
            notifyAll();
            while (!exit) {
                try { wait(); } catch (InterruptedException x) {}
            }
        }
    }

    public Connector shutdown() {
        connectQueue.close();
        return this;
    }

    public synchronized void waitDone() {
        while (!done) {
            try { wait(); } catch (InterruptedException x) {}
        }
    }

    public synchronized Connector exit() {
        exit = true;
        notifyAll();
        return this;
    }

    public void join() {
        while (true) {
            try {
                thread.join();
                return;
            } catch (InterruptedException e) {}
        }
    }
}

@@ -140,353 +140,419 @@
 t;%0A%0A
-%0Aclass Connector %7B%0A%0A    private class Run implements Runnable %7B%0A        public void run() %7B%0A            try %7B%0A                Conn
+import com.yahoo.concurrent.ThreadFactoryFactory;%0A%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Ex
 ec
+u
 tor
-.this.run();%0A            %7D catch (Throwable problem) %7B%0A                parent.handleFailure(problem, Connector.this);%0A            %7D%0A        %7D%0A    %7D%0A%0A    private Thread      thread = new Thread(new Run(), %22%3Cjrt-
+s;%0Aimport java.util.concurrent.RejectedExecutionException;%0Aimport java.util.concurrent.TimeUnit;%0A%0Aclass Connector %7B%0A%0A    private final Transport   parent;%0A    private final ExecutorService executor = Executors.newCachedThreadPool(ThreadFactoryFactory.getDaemonThreadFactory(%22jrt.
 conn
@@ -560,10 +560,10 @@
 ctor
-%3E
 %22
+)
 );%0A 
@@ -577,161 +577,271 @@
 ate 
-Transport   parent;%0A    private ThreadQueue connectQueue = new ThreadQueue
+boolean     done = false;%0A%0A    private void connect(Connection conn) %7B%0A        try %7B%0A            conn.transportThread().addConnection(conn.connect
 ()
+)
 ;%0A    
-private boolean     done = false;%0A    private boolean     exit = false;
+    %7D catch (Throwable problem) %7B%0A            parent.handleFailure(problem, Connector.this);%0A        %7D%0A    %7D
 %0A%0A  
@@ -913,64 +913,8 @@
 nt;%0A
-        thread.setDaemon(true);%0A        thread.start();%0A
     
@@ -957,16 +957,19 @@
 ection c
+onn
 ) %7B%0A    
@@ -976,38 +976,110 @@
     
-if ( ! connectQueue.enqueue(c)
+try %7B%0A            executor.execute(() -%3E connect(conn));%0A        %7D catch (RejectedExecutionException e
 ) %7B%0A
@@ -1091,16 +1091,19 @@
        c
+onn
 .transpo
@@ -1128,16 +1128,19 @@
 ection(c
+onn
 );%0A     
@@ -1136,32 +1136,33 @@
 onn);%0A        %7D%0A
+%0A
     %7D%0A%0A    priva
@@ -1161,535 +1161,171 @@
    p
-rivate void run() %7B%0A        try %7B%0A            while (true) %7B%0A                Connection conn = (Connection) connectQueue.dequeue();%0A                conn.transportThread().addConnection(conn.connect());%0A            %7D%0A        %7D catch (EndOfQueueException e) %7B%7D%0A        synchronized (this) %7B%0A            done = true;%0A            notifyAll();%0A            while (!exit) %7B%0A                try %7B wait(); %7D catch (InterruptedException x) %7B%7D%0A            %7D%0A        %7D%0A    %7D%0A%0A    public Connector shutdown() %7B%0A        connectQueue.close();
+ublic Connector shutdown() %7B%0A        executor.shutdown();%0A        join();%0A        synchronized (this) %7B%0A            done = true;%0A            notifyAll();%0A        %7D
 %0A   
@@ -1541,29 +1541,8 @@
 ) %7B%0A
-        exit = true;%0A
     
@@ -1672,20 +1672,54 @@
     
-thread.join(
+executor.awaitTermination(60, TimeUnit.SECONDS
 );%0A 

Use a cached threadpool for execution of connect. To avoid failing connect blocking other connects.