// Path: conduit/src/main/java/io/rtr/conduit/amqp/transport/TransportExecutor.java
// public class TransportExecutor extends ThreadPoolExecutor {
// 
//     // From ConsumerWorkService
//     private static final int DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2;
// 
//     private static ThreadFactory buildCountedDaemonThreadFactory() {
//         AtomicInteger threadCount = new AtomicInteger(0);
//         return run -> {
//             Thread thread = new Thread(run);
//             thread.setDaemon(true);
//             thread.setName(String.format("AMQPConnection-%s", threadCount.getAndIncrement()));
//             return thread;
//         };
//     }
// 
//     public TransportExecutor() {
//         this(DEFAULT_NUM_THREADS, buildCountedDaemonThreadFactory());
//     }
// 
//     /**
//      * Based on Executors.newFixedThreadPool
//      * @param nThreads
//      * @param threadFactory
//      */
//     public TransportExecutor(int nThreads, ThreadFactory threadFactory) {
//         super(nThreads, nThreads,
//                 0L, TimeUnit.MILLISECONDS,
//                 new LinkedBlockingQueue<Runnable>(),
//                 threadFactory);
//     }
// }


import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MetricsCollector;
import io.rtr.conduit.amqp.transport.TransportExecutor;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;
            factory.setSocketFactory(SSLSocketFactory.getDefault());
        }

        factory.setHost(host);
        factory.setPort(port);

        if (metricsCollector != null) {
            factory.setMetricsCollector(metricsCollector);
        }
    }

    public boolean isConnected() {
        return this.connection != null && this.connection.isOpen();
    }

    public synchronized void connect(AMQPConnectionProperties properties) throws IOException, TimeoutException {
        if (isConnected()) {
            return;
        }
        configureConnectionFactory(properties);
        initializeExecutor();
        connection = connectionFactory.newConnection(executor);
    }

    public synchronized void disconnect() throws IOException {
        //! We are going to assume that closing an already closed
        //  connection is considered success.
        if (isConnected()) {
            try {
                connection.close(connectionFactory.getConnectionTimeout());
// Path: conduit/src/main/java/io/rtr/conduit/amqp/transport/TransportExecutor.java
// public class TransportExecutor extends ThreadPoolExecutor {
// 
//     // From ConsumerWorkService
//     private static final int DEFAULT_NUM_THREADS = Runtime.getRuntime().availableProcessors() * 2;
// 
//     private static ThreadFactory buildCountedDaemonThreadFactory() {
//         AtomicInteger threadCount = new AtomicInteger(0);
//         return run -> {
//             Thread thread = new Thread(run);
//             thread.setDaemon(true);
//             thread.setName(String.format("AMQPConnection-%s", threadCount.getAndIncrement()));
//             return thread;
//         };
//     }
// 
//     public TransportExecutor() {
//         this(DEFAULT_NUM_THREADS, buildCountedDaemonThreadFactory());
//     }
// 
//     /**
//      * Based on Executors.newFixedThreadPool
//      * @param nThreads
//      * @param threadFactory
//      */
//     public TransportExecutor(int nThreads, ThreadFactory threadFactory) {
//         super(nThreads, nThreads,
//                 0L, TimeUnit.MILLISECONDS,
//                 new LinkedBlockingQueue<Runnable>(),
//                 threadFactory);
//     }
// }


// Path: conduit/src/main/java/io/rtr/conduit/amqp/impl/AMQPConnection.java
import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MetricsCollector;
import io.rtr.conduit.amqp.transport.TransportExecutor;
import javax.net.ssl.SSLSocketFactory;
import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;

            factory.setSocketFactory(SSLSocketFactory.getDefault());
        }

        factory.setHost(host);
        factory.setPort(port);

        if (metricsCollector != null) {
            factory.setMetricsCollector(metricsCollector);
        }
    }

    public boolean isConnected() {
        return this.connection != null && this.connection.isOpen();
    }

    public synchronized void connect(AMQPConnectionProperties properties) throws IOException, TimeoutException {
        if (isConnected()) {
            return;
        }
        configureConnectionFactory(properties);
        initializeExecutor();
        connection = connectionFactory.newConnection(executor);
    }

    public synchronized void disconnect() throws IOException {
        //! We are going to assume that closing an already closed
        //  connection is considered success.
        if (isConnected()) {
            try {
                connection.close(connectionFactory.getConnectionTimeout());
                connection = null;