Decouple ConnectionFactory from connection API.

vietj · vietj · commit 8acb97ffd0b8 · 2025-06-30T12:01:56.000+02:00
Motivation:

The ConnectionFactory SPI currently provides an SqlConnection, most callers actually care about the actual Connection SPI and unwrap it then cast it.

Instead we can have the ConnectionFactory only care about providing Connection and let callers wrap it with an SqlConnection of their choice.

As consequence this simplifies the pool implementation as they operation on Connection.
diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionFactory.java b/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionFactory.java
@@ -65,15 +65,10 @@ protected Future<Connection> doConnectInternal(DB2ConnectOptions options, Contex
   }
 
   @Override
-  public Future<SqlConnection> connect(Context context, DB2ConnectOptions options) {
+  public Future<Connection> connect(Context context, DB2ConnectOptions options) {
     ContextInternal contextInternal = (ContextInternal) context;
-    Promise<SqlConnection> promise = contextInternal.promise();
-    connect(asEventLoopContext(contextInternal), options)
-      .map(conn -> {
-        DB2ConnectionImpl db2Connection = new DB2ConnectionImpl(contextInternal, this, conn);
-        conn.init(db2Connection);
-        return (SqlConnection)db2Connection;
-      }).onComplete(promise);
+    Promise<Connection> promise = contextInternal.promise();
+    connect(asEventLoopContext(contextInternal), options).onComplete(promise);
     return promise.future();
   }
 }
diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionImpl.java b/vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionImpl.java
@@ -37,7 +37,12 @@ public static Future<DB2Connection> connect(Vertx vertx, DB2ConnectOptions optio
     } catch (Exception e) {
       return ctx.failedFuture(e);
     }
-    return prepareForClose(ctx, client.connect((Context)ctx, options)).map(DB2Connection::cast);
+    return client.connect((Context)ctx, options).map(conn -> {
+      DB2ConnectionImpl impl = new DB2ConnectionImpl(ctx, client, conn);
+      conn.init(impl);
+      prepareForClose(ctx, impl);
+      return impl;
+    });
   }
 
   public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {
diff --git a/vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java b/vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java
@@ -64,7 +64,8 @@ public Pool newPool(Vertx vertx, Supplier<Future<DB2ConnectOptions>> databases,
   private PoolImpl newPoolImpl(VertxInternal vertx, Handler<SqlConnection> connectHandler, Supplier<Future<DB2ConnectOptions>> databases, PoolOptions options, NetClientOptions transportOptions, CloseFuture closeFuture) {
     boolean pipelinedPool = options instanceof Db2PoolOptions && ((Db2PoolOptions) options).isPipelined();
     ConnectionFactory<DB2ConnectOptions> factory = createConnectionFactory(vertx, transportOptions);
-    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, null, null, context -> factory.connect(context, databases.get()), connectHandler, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, null, null,
+      factory, databases, connectHandler, closeFuture);
     pool.init();
     closeFuture.add(factory);
     return pool;
diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionFactory.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionFactory.java
@@ -106,16 +106,10 @@ private Future<Connection> login(MSSQLSocketConnection conn, MSSQLConnectOptions
   }
 
   @Override
-  public Future<SqlConnection> connect(Context context, MSSQLConnectOptions options) {
+  public Future<Connection> connect(Context context, MSSQLConnectOptions options) {
     ContextInternal ctx = (ContextInternal) context;
-    Promise<SqlConnection> promise = ctx.promise();
-    connect(asEventLoopContext(ctx), options)
-      .map(conn -> {
-        MSSQLConnectionImpl msConn = new MSSQLConnectionImpl(ctx, this, conn);
-        conn.init(msConn);
-        return (SqlConnection)msConn;
-      })
-      .onComplete(promise);
+    Promise<Connection> promise = ctx.promise();
+    connect(asEventLoopContext(ctx), options).onComplete(promise);
     return promise.future();
   }
 }
diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionImpl.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLConnectionImpl.java
@@ -36,7 +36,12 @@ public MSSQLConnectionImpl(ContextInternal context, ConnectionFactory factory, C
   public static Future<MSSQLConnection> connect(Vertx vertx, MSSQLConnectOptions options) {
     ContextInternal ctx = (ContextInternal) vertx.getOrCreateContext();
     MSSQLConnectionFactory client = new MSSQLConnectionFactory(ctx.owner());
-    return prepareForClose(ctx, client.connect((Context)ctx, options)).map(MSSQLConnection::cast);
+    return client.connect((Context)ctx, options).map(conn -> {
+      MSSQLConnectionImpl impl = new MSSQLConnectionImpl(ctx, client, conn);
+      conn.init(impl);
+      prepareForClose(ctx, impl);
+      return impl;
+    });
   }
 
   @Override
diff --git a/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java b/vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java
@@ -65,7 +65,8 @@ public Pool newPool(Vertx vertx, Supplier<Future<MSSQLConnectOptions>> databases
 
   private PoolImpl newPoolImpl(VertxInternal vertx, Handler<SqlConnection> connectHandler, Supplier<Future<MSSQLConnectOptions>> databases, PoolOptions poolOptions, NetClientOptions transportOptions, CloseFuture closeFuture) {
     ConnectionFactory<MSSQLConnectOptions> factory = createConnectionFactory(vertx, transportOptions);
-    PoolImpl pool = new PoolImpl(vertx, this, false, poolOptions, null, null, context -> factory.connect(context, databases.get()), connectHandler, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this, false, poolOptions, null, null,
+      factory, databases, connectHandler, closeFuture);
     pool.init();
     closeFuture.add(factory);
     return pool;
diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionFactory.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionFactory.java
@@ -142,15 +142,15 @@ private Future<Connection> doConnect(MySQLConnectOptions options, SslMode sslMod
   }
 
   @Override
-  public Future<SqlConnection> connect(Context context, MySQLConnectOptions options) {
+  public Future<Connection> connect(Context context, MySQLConnectOptions options) {
     ContextInternal contextInternal = (ContextInternal) context;
-    Promise<SqlConnection> promise = contextInternal.promise();
+    Promise<Connection> promise = contextInternal.promise();
     connect(asEventLoopContext(contextInternal), options)
-      .map(conn -> {
+      /*.map(conn -> {
         MySQLConnectionImpl mySQLConnection = new MySQLConnectionImpl(contextInternal, this, conn);
         conn.init(mySQLConnection);
         return (SqlConnection)mySQLConnection;
-      })
+      })*/
       .onComplete(promise);
     return promise.future();
   }
diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionImpl.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/impl/MySQLConnectionImpl.java
@@ -37,7 +37,12 @@ public static Future<MySQLConnection> connect(ContextInternal ctx, MySQLConnectO
     } catch (Exception e) {
       return ctx.failedFuture(e);
     }
-    return prepareForClose(ctx, client.connect((Context)ctx, options)).map(MySQLConnection::cast);
+    return client.connect((Context)ctx, options).map(conn -> {
+      MySQLConnectionImpl impl = new MySQLConnectionImpl(ctx, client, conn);
+      conn.init(impl);
+      prepareForClose(ctx, impl);
+      return impl;
+    });
   }
 
   public MySQLConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {
diff --git a/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java b/vertx-mysql-client/src/main/java/io/vertx/mysqlclient/spi/MySQLDriver.java
@@ -64,7 +64,8 @@ public Pool newPool(Vertx vertx, Supplier<Future<MySQLConnectOptions>> databases
   private PoolImpl newPoolImpl(VertxInternal vertx, Handler<SqlConnection> connectHandler, Supplier<Future<MySQLConnectOptions>> databases, PoolOptions poolOptions, NetClientOptions transportOptions, CloseFuture closeFuture) {
     boolean pipelinedPool = poolOptions instanceof MySQLPoolOptions && ((MySQLPoolOptions) poolOptions).isPipelined();
     ConnectionFactory<MySQLConnectOptions> factory = createConnectionFactory(vertx, transportOptions);
-    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, poolOptions, null, null, context -> factory.connect(context, databases.get()), connectHandler, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, poolOptions, null, null, factory,
+      databases, connectHandler, closeFuture);
     pool.init();
     closeFuture.add(factory);
     return pool;
diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionFactory.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionFactory.java
@@ -13,15 +13,14 @@
 import io.vertx.core.Completable;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
-import io.vertx.core.Promise;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.VertxInternal;
 import io.vertx.core.json.JsonObject;
 import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.oracleclient.OracleConnectOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
-import io.vertx.sqlclient.SqlConnection;
+import io.vertx.sqlclient.internal.Connection;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 import oracle.jdbc.OracleConnection;
 import oracle.jdbc.datasource.OracleDataSource;
@@ -59,7 +58,7 @@ private OracleDataSource getDatasource(SqlConnectOptions options) {
   }
 
   @Override
-  public Future<SqlConnection> connect(Context context, OracleConnectOptions options) {
+  public Future<Connection> connect(Context context, OracleConnectOptions options) {
     OracleDataSource datasource = getDatasource(options);
     VertxMetrics vertxMetrics = ((VertxInternal)context.owner()).metrics();
     ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(options.getSocketAddress(), "sql", options.getMetricsName()) : null;
@@ -68,9 +67,7 @@ public Future<SqlConnection> connect(Context context, OracleConnectOptions optio
       OracleConnection orac = datasource.createConnectionBuilder().build();
       OracleMetadata metadata = new OracleMetadata(orac.getMetaData());
       OracleJdbcConnection conn = new OracleJdbcConnection(ctx, metrics, options, orac, metadata);
-      OracleConnectionImpl msConn = new OracleConnectionImpl(ctx, this, conn);
-      conn.init(msConn);
-      return msConn;
+      return conn;
     });
   }
 }
diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionImpl.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/impl/OracleConnectionImpl.java
@@ -11,6 +11,7 @@
 
 package io.vertx.oracleclient.impl;
 
+import io.vertx.core.Context;
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.core.internal.ContextInternal;
@@ -30,6 +31,11 @@ public OracleConnectionImpl(ContextInternal context, ConnectionFactory factory,
   public static Future<OracleConnection> connect(Vertx vertx, OracleConnectOptions options) {
     ContextInternal ctx = (ContextInternal) vertx.getOrCreateContext();
     OracleConnectionFactory client = new OracleConnectionFactory(ctx.owner());
-    return prepareForClose(ctx, client.connect(ctx, options)).map(OracleConnection::cast);
+    return client.connect(ctx, options).map(conn -> {
+      OracleConnectionImpl impl = new OracleConnectionImpl(ctx, client, conn);
+      conn.init(impl);
+      prepareForClose(ctx, impl);
+      return impl;
+    });
   }
 }
diff --git a/vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java b/vertx-oracle-client/src/main/java/io/vertx/oracleclient/spi/OracleDriver.java
@@ -61,7 +61,8 @@ private PoolImpl newPoolImpl(VertxInternal vertx, Handler<SqlConnection> connect
     Function<Connection, Future<Void>> afterAcquire = conn -> ((OracleJdbcConnection) conn).afterAcquire();
     Function<Connection, Future<Void>> beforeRecycle = conn -> ((OracleJdbcConnection) conn).beforeRecycle();
     ConnectionFactory<OracleConnectOptions> factory = createConnectionFactory(vertx, null);
-    PoolImpl pool = new PoolImpl(vertx, this,  false, options, afterAcquire, beforeRecycle, context -> factory.connect(context, databases.get()), connectHandler, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this,  false, options, afterAcquire, beforeRecycle,
+      factory, databases, connectHandler, closeFuture);
     pool.init();
     closeFuture.add(factory);
     return pool;
diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionFactory.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionFactory.java
@@ -28,7 +28,6 @@
 import io.vertx.core.internal.net.NetSocketInternal;
 import io.vertx.pgclient.PgConnectOptions;
 import io.vertx.pgclient.SslMode;
-import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.internal.Connection;
 import io.vertx.sqlclient.impl.ConnectionFactoryBase;
 
@@ -153,19 +152,13 @@ private Future<Connection> doConnect(ConnectOptions connectOptions, ContextInter
   }
 
   @Override
-  public Future<SqlConnection> connect(Context context, PgConnectOptions options) {
+  public Future<Connection> connect(Context context, PgConnectOptions options) {
     ContextInternal contextInternal = (ContextInternal) context;
     if (options.isUsingDomainSocket() && !vertx.isNativeTransportEnabled()) {
       return contextInternal.failedFuture(new IllegalArgumentException(NATIVE_TRANSPORT_REQUIRED));
     }
-    PromiseInternal<SqlConnection> promise = contextInternal.promise();
-    connect(asEventLoopContext(contextInternal), options)
-      .map(conn -> {
-        PgConnectionImpl pgConn = new PgConnectionImpl(this, contextInternal, conn);
-        conn.init(pgConn);
-        return (SqlConnection)pgConn;
-      })
-      .onComplete(promise);
+    PromiseInternal<Connection> promise = contextInternal.promise();
+    connect(asEventLoopContext(contextInternal), options).onComplete(promise);
     return promise.future();
   }
 
diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/impl/PgConnectionImpl.java
@@ -42,7 +42,12 @@ public static Future<PgConnection> connect(ContextInternal context, PgConnectOpt
     } catch (Exception e) {
       return context.failedFuture(e);
     }
-    return prepareForClose(context, client.connect((Context)context, options)).map(PgConnection::cast);
+    return client.connect((Context)context, options).map(conn -> {
+      PgConnectionImpl impl = new PgConnectionImpl(client, context, conn);
+      conn.init(impl);
+      prepareForClose(context, impl);
+      return impl;
+    });
   }
 
   private volatile Handler<PgNotification> notificationHandler;
diff --git a/vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java b/vertx-pg-client/src/main/java/io/vertx/pgclient/spi/PgDriver.java
@@ -47,7 +47,8 @@ public Pool newPool(Vertx vertx, Supplier<Future<PgConnectOptions>> databases, P
   private PoolImpl newPoolImpl(VertxInternal vertx, Handler<SqlConnection> connectHandler, Supplier<Future<PgConnectOptions>> databases, PoolOptions poolOptions, NetClientOptions transportOptions, CloseFuture closeFuture) {
     boolean pipelinedPool = poolOptions instanceof PgPoolOptions && ((PgPoolOptions) poolOptions).isPipelined();
     ConnectionFactory<PgConnectOptions> factory = createConnectionFactory(vertx, transportOptions);
-    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, poolOptions, null, null, context -> factory.connect(context, databases.get()), connectHandler, closeFuture);
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, poolOptions, null, null,
+      factory, databases, connectHandler, closeFuture);
     pool.init();
     closeFuture.add(factory);
     return pool;
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java
@@ -23,6 +23,7 @@
 import io.vertx.core.internal.pool.*;
 import io.vertx.core.internal.ContextInternal;
 import io.vertx.core.internal.VertxInternal;
+import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.internal.Connection;
 import io.vertx.sqlclient.internal.SqlConnectionBase;
@@ -34,6 +35,7 @@
 
 import java.util.List;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 /**
@@ -47,7 +49,8 @@ public class SqlConnectionPool {
 
   private static final Object NO_METRICS = new Object();
 
-  private final Function<Context, Future<SqlConnection>> connectionProvider;
+  private final Function<Context, Future<Connection>> connectionProvider;
+  private final ConnectionFactory<?> connectionFactory;
   private final VertxInternal vertx;
   private final PoolMetrics metrics;
   private final ConnectionPool<PooledConnection> pool;
@@ -59,7 +62,9 @@ public class SqlConnectionPool {
   private final long maxLifetime;
   private final int maxSize;
 
-  public SqlConnectionPool(Function<Context, Future<SqlConnection>> connectionProvider,
+  // TODO : use connection provider with Connection instead of SqlConnection
+  public <O extends SqlConnectOptions> SqlConnectionPool(Supplier<Future<O>> optionsProvider,
+                           ConnectionFactory<O> connectionFactory,
                            PoolMetrics metrics,
                            Handler<PooledConnection> hook,
                            Function<Connection, Future<Void>> afterAcquire,
@@ -78,14 +83,15 @@ public SqlConnectionPool(Function<Context, Future<SqlConnection>> connectionProv
       throw new IllegalArgumentException("afterAcquire and beforeRecycle hooks must be both not null");
     }
     this.pool = ConnectionPool.pool(connector, new int[]{maxSize}, maxWaitQueueSize);
+    this.connectionFactory = connectionFactory;
     this.metrics = metrics;
     this.vertx = vertx;
     this.pipelined = pipelined;
     this.idleTimeout = idleTimeout;
     this.maxLifetime = maxLifetime;
     this.maxSize = maxSize;
     this.hook = hook;
-    this.connectionProvider = connectionProvider;
+    this.connectionProvider = context -> connectionFactory.connect(context, optionsProvider.get());
     this.afterAcquire = afterAcquire;
     this.beforeRecycle = beforeRecycle;
 
@@ -114,12 +120,10 @@ public ContextInternal apply(ContextInternal contextInternal) {
   private final PoolConnector<PooledConnection> connector = new PoolConnector<>() {
     @Override
     public Future<ConnectResult<PooledConnection>> connect(ContextInternal context, Listener listener) {
-      Future<SqlConnection> future = connectionProvider.apply(context);
-      return future.compose(res -> {
-        SqlConnectionBase connBase = (SqlConnectionBase) res;
-        Connection conn = connBase.unwrap();
+      Future<Connection> future = connectionProvider.apply(context);
+      return future.compose(conn -> {
         if (conn.isValid()) {
-          PooledConnection pooled = new PooledConnection(connBase.factory(), conn, listener);
+          PooledConnection pooled = new PooledConnection(connectionFactory, conn, listener);
           conn.init(pooled);
           if (hook != null) {
             Promise<ConnectResult<PooledConnection>> p = Promise.promise();
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/SqlConnectionBase.java
@@ -230,10 +230,13 @@ private void doClose(Completable<Void> promise) {
   protected static Future<SqlConnection> prepareForClose(ContextInternal ctx, Future<SqlConnection> future) {
     return future.andThen(ar -> {
       if (ar.succeeded()) {
-        SqlConnectionBase<?> base = (SqlConnectionBase<?>) ar.result();
-        base.closeFactoryAfterUsage = true;
-        ctx.addCloseHook(base);
+        prepareForClose(ctx, (SqlConnectionBase<?>) ar.result());
       }
     });
   }
+
+  protected static void prepareForClose(ContextInternal ctx, SqlConnectionBase<?> base) {
+    base.closeFactoryAfterUsage = true;
+    ctx.addCloseHook(base);
+  }
 }
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/pool/PoolImpl.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/internal/pool/PoolImpl.java
@@ -32,9 +32,11 @@
 import io.vertx.sqlclient.internal.SqlClientBase;
 import io.vertx.sqlclient.internal.SqlConnectionInternal;
 import io.vertx.sqlclient.internal.command.CommandBase;
+import io.vertx.sqlclient.spi.ConnectionFactory;
 import io.vertx.sqlclient.spi.Driver;
 
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
@@ -55,13 +57,14 @@ public class PoolImpl extends SqlClientBase implements Pool, Closeable {
   private final Handler<SqlConnection> connectionInitializer;
   private long timerID;
 
-  public PoolImpl(VertxInternal vertx,
+  public <O extends SqlConnectOptions> PoolImpl(VertxInternal vertx,
                   Driver driver,
                   boolean pipelined,
                   PoolOptions poolOptions,
                   Function<Connection, Future<Void>> afterAcquire,
                   Function<Connection, Future<Void>> beforeRecycle,
-                  Function<Context, Future<SqlConnection>> connectionProvider,
+                  ConnectionFactory<O> connectionFactory,
+                  Supplier<Future<O>> connectionProvider,
                   Handler<SqlConnection> connectionInitializer,
                   CloseFuture closeFuture) {
     super(driver);
@@ -83,7 +86,9 @@ public PoolImpl(VertxInternal vertx,
     this.timerID = -1L;
     this.pipelined = pipelined;
     this.vertx = vertx;
-    this.pool = new SqlConnectionPool(connectionProvider, poolMetrics, hook, afterAcquire, beforeRecycle, vertx, idleTimeout, maxLifetime, poolOptions.getMaxSize(), pipelined, poolOptions.getMaxWaitQueueSize(), poolOptions.getEventLoopSize());
+    this.pool = new SqlConnectionPool(connectionProvider, connectionFactory, poolMetrics, hook, afterAcquire,
+      beforeRecycle, vertx, idleTimeout, maxLifetime, poolOptions.getMaxSize(), pipelined,
+      poolOptions.getMaxWaitQueueSize(), poolOptions.getEventLoopSize());
     this.closeFuture = closeFuture;
     this.connectionInitializer = connectionInitializer;
   }
diff --git a/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/ConnectionFactory.java b/vertx-sql-client/src/main/java/io/vertx/sqlclient/spi/ConnectionFactory.java

Original file line number	Diff line number	Diff line change
`@@ -65,15 +65,10 @@ protected Future<Connection> doConnectInternal(DB2ConnectOptions options, Contex`
`65`	`65`	`}`
`66`	`66`
`67`	`67`	`@Override`
`68`		`- public Future<SqlConnection> connect(Context context, DB2ConnectOptions options) {`
	`68`	`+ public Future<Connection> connect(Context context, DB2ConnectOptions options) {`
`69`	`69`	`ContextInternal contextInternal = (ContextInternal) context;`
`70`		`- Promise<SqlConnection> promise = contextInternal.promise();`
`71`		`- connect(asEventLoopContext(contextInternal), options)`
`72`		`- .map(conn -> {`
`73`		`- DB2ConnectionImpl db2Connection = new DB2ConnectionImpl(contextInternal, this, conn);`
`74`		`- conn.init(db2Connection);`
`75`		`- return (SqlConnection)db2Connection;`
`76`		`- }).onComplete(promise);`
	`70`	`+ Promise<Connection> promise = contextInternal.promise();`
	`71`	`+ connect(asEventLoopContext(contextInternal), options).onComplete(promise);`
`77`	`72`	`return promise.future();`
`78`	`73`	`}`
`79`	`74`	`}`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,12 @@ public static Future<DB2Connection> connect(Vertx vertx, DB2ConnectOptions optio`
`37`	`37`	`} catch (Exception e) {`
`38`	`38`	`return ctx.failedFuture(e);`
`39`	`39`	`}`
`40`		`- return prepareForClose(ctx, client.connect((Context)ctx, options)).map(DB2Connection::cast);`
	`40`	`+ return client.connect((Context)ctx, options).map(conn -> {`
	`41`	`+ DB2ConnectionImpl impl = new DB2ConnectionImpl(ctx, client, conn);`
	`42`	`+ conn.init(impl);`
	`43`	`+ prepareForClose(ctx, impl);`
	`44`	`+ return impl;`
	`45`	`+ });`
`41`	`46`	`}`
`42`	`47`
`43`	`48`	`public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {`
Original file line number	Diff line number	Diff line change
`@@ -106,16 +106,10 @@ private Future<Connection> login(MSSQLSocketConnection conn, MSSQLConnectOptions`
`106`	`106`	`}`
`107`	`107`
`108`	`108`	`@Override`
`109`		`- public Future<SqlConnection> connect(Context context, MSSQLConnectOptions options) {`
	`109`	`+ public Future<Connection> connect(Context context, MSSQLConnectOptions options) {`
`110`	`110`	`ContextInternal ctx = (ContextInternal) context;`
`111`		`- Promise<SqlConnection> promise = ctx.promise();`
`112`		`- connect(asEventLoopContext(ctx), options)`
`113`		`- .map(conn -> {`
`114`		`- MSSQLConnectionImpl msConn = new MSSQLConnectionImpl(ctx, this, conn);`
`115`		`- conn.init(msConn);`
`116`		`- return (SqlConnection)msConn;`
`117`		`- })`
`118`		`- .onComplete(promise);`
	`111`	`+ Promise<Connection> promise = ctx.promise();`
	`112`	`+ connect(asEventLoopContext(ctx), options).onComplete(promise);`
`119`	`113`	`return promise.future();`
`120`	`114`	`}`
`121`	`115`	`}`