Implement Postgres request cancellation

Author: vietj

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionFactory.java

@@ -97,9 +97,8 @@ protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventL
     });
   }
 
-  public void cancelRequest(SocketAddress server, int processId, int secretKey, Handler<AsyncResult<Void>> handler) {
-    // NOT GOOD
-    doConnect(server, vertx.createEventLoopContext(), (PgConnectOptions) options).onComplete(ar -> {
+  public void cancelRequest(PgConnectOptions options, int processId, int secretKey, Handler<AsyncResult<Void>> handler) {
+    doConnect(options.getSocketAddress(), vertx.createEventLoopContext(), options).onComplete(ar -> {
       if (ar.succeeded()) {
         PgSocketConnection conn = (PgSocketConnection) ar.result();
         conn.sendCancelRequestMessage(processId, secretKey, handler);
@@ -182,6 +181,8 @@ private PgSocketConnection newSocketConnection(EventLoopContext context, NetSock
     Predicate<String> preparedStatementCacheSqlFilter = options.getPreparedStatementCacheSqlFilter();
     int pipeliningLimit = options.getPipeliningLimit();
     boolean useLayer7Proxy = options.getUseLayer7Proxy();
-    return new PgSocketConnection(socket, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, useLayer7Proxy, context);
+    PgSocketConnection conn = new PgSocketConnection(socket, cachePreparedStatements, preparedStatementCacheMaxSize, preparedStatementCacheSqlFilter, pipeliningLimit, useLayer7Proxy, context);
+    conn.options = options;
+    return conn;
   }
 }

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java

@@ -133,7 +133,8 @@ public Future<Void> cancelRequest() {
   public PgConnection cancelRequest(Handler<AsyncResult<Void>> handler) {
     Context current = Vertx.currentContext();
     if (current == context) {
-      ((PgConnectionFactory)factory).cancelRequest(conn.server(), this.processId(), this.secretKey(), handler);
+      PgSocketConnection unwrap = (PgSocketConnection) conn.unwrap();
+      ((PgConnectionFactory)factory).cancelRequest(unwrap.options, this.processId(), this.secretKey(), handler);
     } else {
       context.runOnContext(v -> cancelRequest(handler));
     }

vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgSocketConnection.java

@@ -26,6 +26,7 @@
 import io.vertx.core.buffer.Buffer;
 import io.vertx.core.impl.EventLoopContext;
 import io.vertx.core.net.impl.NetSocketInternal;
+import io.vertx.pgclient.PgConnectOptions;
 import io.vertx.pgclient.PgException;
 import io.vertx.pgclient.impl.codec.NoticeResponse;
 import io.vertx.pgclient.impl.codec.PgCodec;
@@ -54,6 +55,7 @@ public class PgSocketConnection extends SocketConnectionBase {
   public int processId;
   public int secretKey;
   public PgDatabaseMetadata dbMetaData;
+  PgConnectOptions options;
 
   public PgSocketConnection(NetSocketInternal socket,
                             boolean cachePreparedStatements,

vertx-pg-client/src/test/java/io/vertx/pgclient/PgConnectionTest.java

@@ -102,7 +102,6 @@ public void testQueueQueries(TestContext ctx) {
     }));
   }
 
-  @Ignore("FIXME")
   @Test
   public void testCancelRequest(TestContext ctx) {
     Async async = ctx.async(2);

vertx-pg-client/src/test/java/io/vertx/pgclient/PgPoolTestBase.java

@@ -24,7 +24,6 @@
 import io.vertx.sqlclient.*;
 import org.junit.After;
 import org.junit.Before;
-import org.junit.Ignore;
 import org.junit.Test;
 
 import java.util.concurrent.atomic.AtomicReference;
@@ -183,7 +182,6 @@ public void testReconnect(TestContext ctx) {
     }));
   }
 
-  @Ignore("FIXME")
   @Test
   public void testCancelRequest(TestContext ctx) {
     Async async = ctx.async();

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/SqlConnectionBase.java

@@ -22,8 +22,6 @@
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.sqlclient.PrepareOptions;
 import io.vertx.sqlclient.PreparedStatement;
-import io.vertx.sqlclient.SqlClient;
-import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.Transaction;
 import io.vertx.sqlclient.impl.command.CommandBase;
 import io.vertx.sqlclient.impl.command.PrepareStatementCommand;