// Path: src/main/java/com/pgasync/Connectible.java
// public interface Connectible extends QueryExecutor {
// 
//     CompletableFuture<Connection> getConnection();
// 
//     CompletableFuture<Void> close();
// }
// 
// Path: src/main/java/com/pgasync/Connection.java
// public interface Connection extends QueryExecutor {
// 
//     CompletableFuture<PreparedStatement> prepareStatement(String sql, Oid... parametersTypes);
// 
//     /**
//      * The typical scenario of using notifications is as follows:
//      * - {@link #subscribe(String, Consumer)}
//      * - Some calls of @onNotification callback
//      * - {@link Listening#unlisten()}
//      *
//      * @param channel Channel name to listen to.
//      * @param onNotification Callback, thar is cinvoked every time notification arrives.
//      * @return CompletableFuture instance, completed when subscription will be registered at the backend.
//      */
//     CompletableFuture<Listening> subscribe(String channel, Consumer<String> onNotification);
// 
//     CompletableFuture<Void> close();
// }
// 
// Path: src/main/java/com/pgasync/SqlException.java
// public class SqlException extends RuntimeException {
// 
//     private static final long serialVersionUID = 1L;
//     private static final int MAX_CAUSE_DEPTH = 100;
// 
//     final String code;
// 
//     public SqlException(String level, String code, String message) {
//         super(level + ": SQLSTATE=" + code + ", MESSAGE=" + message);
//         this.code = code;
//     }
// 
//     public SqlException(String message) {
//         super(message);
//         this.code = null;
//     }
// 
//     public SqlException(Throwable cause) {
//         super(cause);
//         this.code = null;
//     }
// 
//     public String getCode() {
//         return code;
//     }
// 
//     @FunctionalInterface
//     public interface CheckedRunnable {
// 
//         void run() throws Exception;
//     }
// 
//     public static boolean ifCause(Throwable th, Consumer<SqlException> action, CheckedRunnable others) throws Exception {
//         int depth = 1;
//         while (depth++ < MAX_CAUSE_DEPTH && th != null && !(th instanceof SqlException)) {
//             th = th.getCause();
//         }
//         if (th instanceof SqlException) {
//             action.accept((SqlException) th);
//             return true;
//         } else {
//             others.run();
//             return false;
//         }
//     }
// }


import static java.util.concurrent.TimeUnit.SECONDS;
import static org.hamcrest.CoreMatchers.*;
import static org.junit.Assert.assertThat;
import com.pgasync.Connectible;
import com.pgasync.Connection;
import com.pgasync.SqlException;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import java.util.Deque;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.SynchronousQueue;
import java.util.stream.IntStream;
import static java.lang.System.currentTimeMillis;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
    public void connectionPoolPipelinesQueries() throws InterruptedException {
        int count = 5;
        double sleep = 0.5;
        Deque<Long> results = new LinkedBlockingDeque<>();
        long startWrite = currentTimeMillis();
        for (int i = 0; i < count; ++i) {
            pool.completeQuery("select " + i + ", pg_sleep(" + sleep + ")")
                    .thenAccept(r -> results.add(currentTimeMillis()))
                    .exceptionally(th -> {
                        throw new AssertionError("failed", th);
                    });
        }
        long writeTime = currentTimeMillis() - startWrite;

        long remoteWaitTimeSeconds = (long) (sleep * count);
        SECONDS.sleep(2 + remoteWaitTimeSeconds);
        long readTime = results.getLast() - results.getFirst();

        assertThat(results.size(), is(count));
        assertThat(MILLISECONDS.toSeconds(writeTime), is(0L));
        assertThat(MILLISECONDS.toSeconds(readTime + 999) >= remoteWaitTimeSeconds, is(true));
    }

    private Connection getConnection() throws InterruptedException {
        SynchronousQueue<Connection> connQueue = new SynchronousQueue<>();
        pool.getConnection()
                .thenAccept(connQueue::offer);
        return c = connQueue.take();
    }

// Path: src/main/java/com/pgasync/Connectible.java
// public interface Connectible extends QueryExecutor {
// 
//     CompletableFuture<Connection> getConnection();
// 
//     CompletableFuture<Void> close();
// }
// 
// Path: src/main/java/com/pgasync/Connection.java
// public interface Connection extends QueryExecutor {
// 
//     CompletableFuture<PreparedStatement> prepareStatement(String sql, Oid... parametersTypes);
// 
//     /**
//      * The typical scenario of using notifications is as follows:
//      * - {@link #subscribe(String, Consumer)}
//      * - Some calls of @onNotification callback
//      * - {@link Listening#unlisten()}
//      *
//      * @param channel Channel name to listen to.
//      * @param onNotification Callback, thar is cinvoked every time notification arrives.
//      * @return CompletableFuture instance, completed when subscription will be registered at the backend.
//      */
//     CompletableFuture<Listening> subscribe(String channel, Consumer<String> onNotification);
// 
//     CompletableFuture<Void> close();
// }
// 
// Path: src/main/java/com/pgasync/SqlException.java
// public class SqlException extends RuntimeException {
// 
//     private static final long serialVersionUID = 1L;
//     private static final int MAX_CAUSE_DEPTH = 100;
// 
//     final String code;
// 
//     public SqlException(String level, String code, String message) {
//         super(level + ": SQLSTATE=" + code + ", MESSAGE=" + message);
//         this.code = code;
//     }
// 
//     public SqlException(String message) {
//         super(message);
//         this.code = null;
//     }
// 
//     public SqlException(Throwable cause) {
//         super(cause);
//         this.code = null;
//     }
// 
//     public String getCode() {
//         return code;
//     }
// 
//     @FunctionalInterface
//     public interface CheckedRunnable {
// 
//         void run() throws Exception;
//     }
// 
//     public static boolean ifCause(Throwable th, Consumer<SqlException> action, CheckedRunnable others) throws Exception {
//         int depth = 1;
//         while (depth++ < MAX_CAUSE_DEPTH && th != null && !(th instanceof SqlException)) {
//             th = th.getCause();
//         }
//         if (th instanceof SqlException) {
//             action.accept((SqlException) th);
//             return true;
//         } else {
//             others.run();
//             return false;
//         }
//     }
// }


// Path: src/test/java/com/github/pgasync/PipelineTest.java
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.hamcrest.CoreMatchers.*;
import static org.junit.Assert.assertThat;
import com.pgasync.Connectible;
import com.pgasync.Connection;
import com.pgasync.SqlException;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import java.util.Deque;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.SynchronousQueue;
import java.util.stream.IntStream;
import static java.lang.System.currentTimeMillis;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

    public void connectionPoolPipelinesQueries() throws InterruptedException {
        int count = 5;
        double sleep = 0.5;
        Deque<Long> results = new LinkedBlockingDeque<>();
        long startWrite = currentTimeMillis();
        for (int i = 0; i < count; ++i) {
            pool.completeQuery("select " + i + ", pg_sleep(" + sleep + ")")
                    .thenAccept(r -> results.add(currentTimeMillis()))
                    .exceptionally(th -> {
                        throw new AssertionError("failed", th);
                    });
        }
        long writeTime = currentTimeMillis() - startWrite;

        long remoteWaitTimeSeconds = (long) (sleep * count);
        SECONDS.sleep(2 + remoteWaitTimeSeconds);
        long readTime = results.getLast() - results.getFirst();

        assertThat(results.size(), is(count));
        assertThat(MILLISECONDS.toSeconds(writeTime), is(0L));
        assertThat(MILLISECONDS.toSeconds(readTime + 999) >= remoteWaitTimeSeconds, is(true));
    }

    private Connection getConnection() throws InterruptedException {
        SynchronousQueue<Connection> connQueue = new SynchronousQueue<>();
        pool.getConnection()
                .thenAccept(connQueue::offer);
        return c = connQueue.take();
    }

    @Test(expected = SqlException.class)