The pool notifies the connect handler after the connection has been created and inserted in the pool and before the first time it is used. This is implemented for connection acquisitions but not for scheduled commands.

The connect handler is now called after the connection is created and before it is inserted in the pool. As consequence it works universally for connection acquisition and scheduled commands. fixes eclipse-vertx#1134

Author: vietj

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

@@ -425,21 +425,35 @@ public void testNoConnectionLeaks(TestContext ctx) {
  }
 
  @Test
- public void testConnectionHook(TestContext ctx) {
- Async async = ctx.async();
+ public void testConnectionHook1(TestContext ctx) {
+ Async async = ctx.async(2);
  Handler<SqlConnection> hook = f -> {
  vertx.setTimer(1000, id -> {
- f.close();
+ f.close().onComplete(ctx.asyncAssertSuccess(v -> async.countDown()));
  });
  };
  PgPool pool = createPool(options, new PoolOptions().setMaxSize(1)).connectHandler(hook);
  pool.getConnection(ctx.asyncAssertSuccess(conn -> {
  conn.query("SELECT id, randomnumber from WORLD").execute(ctx.asyncAssertSuccess(v2 -> {
- async.complete();
+ async.countDown();
  }));
  }));
  }
 
+ @Test
+ public void testConnectionHook2(TestContext ctx) {
+ Async async = ctx.async(2);
+ Handler<SqlConnection> hook = f -> {
+ vertx.setTimer(1000, id -> {
+ f.close().onComplete(ctx.asyncAssertSuccess(v -> async.countDown()));
+ });
+ };
+ PgPool pool = createPool(options, new PoolOptions().setMaxSize(1)).connectHandler(hook);
+ pool.query("SELECT id, randomnumber from WORLD").execute(ctx.asyncAssertSuccess(v2 -> {
+ async.countDown();
+ }));
+ }
+
  @Test
  public void testConnectionClosedInHook(TestContext ctx) {
  Async async = ctx.async(2);

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

@@ -109,7 +109,13 @@ public void connect(EventLoopContext context, PoolConnector.Listener listener, H
  if (conn.isValid()) {
  PooledConnection pooled = new PooledConnection(res.factory(), conn, listener);
  conn.init(pooled);
- handler.handle(Future.succeededFuture(new ConnectResult<>(pooled, pipeliningLimit, 0)));
+ Handler<PooledConnection> connectionHandler = hook.get();
+ if (connectionHandler != null) {
+ pooled.poolResultHandler = handler;
+ connectionHandler.handle(pooled);
+ } else {
+ handler.handle(Future.succeededFuture(new ConnectResult<>(pooled, pipeliningLimit, 0)));
+ }
  } else {
  handler.handle(Future.failedFuture(ConnectionBase.CLOSED_EXCEPTION));
  }
@@ -168,6 +174,7 @@ public <R> Future<R> execute(ContextInternal context, CommandBase<R> cmd) {
  public void acquire(ContextInternal context, long timeout, Handler<AsyncResult<PooledConnection>> handler) {
  class PoolRequest implements PoolWaiter.Listener<PooledConnection>, Handler<AsyncResult<Lease<PooledConnection>>> {
  private long timerID = -1L;
+
  @Override
  public void handle(AsyncResult<Lease<PooledConnection>> ar) {
  if (timerID != -1L) {
@@ -191,19 +198,10 @@ public void handle(AsyncResult<Lease<PooledConnection>> ar) {
  handler.handle(Future.failedFuture(ar.cause()));
  }
  }
+
  private void handle(Lease<PooledConnection> lease) {
  PooledConnection pooled = lease.get();
  pooled.lease = lease;
- if (!pooled.initialized) {
- Handler<PooledConnection> connectionHandler = hook.get();
- if (connectionHandler != null) {
- pooled.continuation = handler;
- connectionHandler.handle(pooled);
- return;
- } else {
- pooled.initialized = true;
- }
- }
  handler.handle(Future.succeededFuture(pooled));
  }
 
@@ -245,10 +243,9 @@ public class PooledConnection implements Connection, Connection.Holder {
  private final Connection conn;
  private final PoolConnector.Listener listener;
  private Holder holder;
+ private Handler<AsyncResult<ConnectResult<PooledConnection>>> poolResultHandler;
  private Lease<PooledConnection> lease;
  public long expirationTimestamp;
- private boolean initialized;
- private Handler<AsyncResult<PooledConnection>> continuation;
 
  PooledConnection(ConnectionFactory factory, Connection conn, PoolConnector.Listener listener) {
  this.factory = factory;
@@ -317,12 +314,11 @@ private void doClose(Holder holder, Promise<Void> promise) {
  promise.fail(msg);
  } else {
  this.holder = null;
- if (!initialized) {
- initialized = true;
- Handler<AsyncResult<PooledConnection>> c = continuation;
- continuation = null;
+ Handler<AsyncResult<ConnectResult<PooledConnection>>> resultHandler = poolResultHandler;
+ if (resultHandler != null) {
+ poolResultHandler = null;
  promise.complete();
- c.handle(Future.succeededFuture(this));
+ resultHandler.handle(Future.succeededFuture(new ConnectResult<>(this, pipeliningLimit, 0)));
  return;
  }
  if (beforeRecycle == null) {
@@ -346,11 +342,10 @@ public void handleClosed() {
  if (holder != null) {
  holder.handleClosed();
  }
-
- Handler<AsyncResult<PooledConnection>> c = this.continuation;
- if (c != null) {
- this.continuation = null;
- c.handle(Future.failedFuture(ConnectionBase.CLOSED_EXCEPTION));
+ Handler<AsyncResult<ConnectResult<PooledConnection>>> resultHandler = poolResultHandler;
+ if (resultHandler != null) {
+ poolResultHandler = null;
+ resultHandler.handle(Future.failedFuture(ConnectionBase.CLOSED_EXCEPTION));
  }
  listener.onRemove();
  }