Merge pull request #1305 from eclipse-vertx/connect-options-supplier

Connect options supplier support

Author: vietj

vertx-db2-client/src/main/java/examples/SqlClientExamples.java

@@ -407,15 +407,11 @@ public static void poolSharing3(Vertx vertx, DB2ConnectOptions database, int max
       .setEventLoopSize(4));
   }
 
-  public void dynamicPoolConfig(Vertx vertx, DB2Pool pool) {
-    pool.connectionProvider(ctx -> {
-      Future<DB2ConnectOptions> fut = retrieveOptions();
-      return fut.compose(connectOptions -> {
-        // Do not forget to close later
-        ConnectionFactory factory = DB2Driver.INSTANCE.createConnectionFactory(vertx, connectOptions);
-        return factory.connect(ctx);
-      });
-    });
+  public void dynamicPoolConfig(Vertx vertx, PoolOptions poolOptions) {
+    DB2Pool pool = DB2Pool.pool(vertx, () -> {
+      Future<DB2ConnectOptions> connectOptions = retrieveOptions();
+      return connectOptions;
+    }, poolOptions);
   }
 
   private Future<DB2ConnectOptions> retrieveOptions() {

vertx-db2-client/src/main/java/io/vertx/db2client/DB2Pool.java

@@ -16,6 +16,7 @@
 package io.vertx.db2client;
 
 import io.vertx.codegen.annotations.Fluent;
+import io.vertx.codegen.annotations.GenIgnore;
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
@@ -31,6 +32,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static io.vertx.db2client.DB2ConnectOptions.fromUri;
 
@@ -109,6 +111,28 @@ static DB2Pool pool(Vertx vertx, List<DB2ConnectOptions> databases, PoolOptions
     return (DB2Pool) DB2Driver.INSTANCE.createPool(vertx, databases, options);
   }
 
+  /**
+   * Create a connection pool to the DB2 {@code databases}. The supplier is called
+   * to provide the options when a new connection is created by the pool.
+   *
+   * @param databases the databases supplier
+   * @param poolOptions the options for creating the pool
+   * @return the connection pool
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  static DB2Pool pool(Supplier<Future<DB2ConnectOptions>> databases, PoolOptions poolOptions) {
+    return pool(null, databases, poolOptions);
+  }
+
+
+  /**
+   * Like {@link #pool(Supplier, PoolOptions)} with a specific {@link Vertx} instance.
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  static DB2Pool pool(Vertx vertx, Supplier<Future<DB2ConnectOptions>> databases, PoolOptions poolOptions) {
+    return (DB2Pool) DB2Driver.INSTANCE.createPool(vertx, databases, poolOptions);
+  }
+
   /**
    * Like {@link #client(String, PoolOptions)} with default options.
    */

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionFactory.java

@@ -21,53 +21,57 @@
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.EventLoopContext;
 import io.vertx.core.impl.VertxInternal;
-import io.vertx.core.net.NetClientOptions;
+import io.vertx.core.net.NetClient;
 import io.vertx.core.net.SocketAddress;
 import io.vertx.core.net.impl.NetSocketInternal;
+import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.SqlConnection;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.ConnectionFactoryBase;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 
-public class DB2ConnectionFactory extends ConnectionFactoryBase {
+import java.util.Map;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
 
-  private int pipeliningLimit;
+public class DB2ConnectionFactory extends ConnectionFactoryBase {
 
-  public DB2ConnectionFactory(VertxInternal vertx, DB2ConnectOptions options) {
+  public DB2ConnectionFactory(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> options) {
     super(vertx, options);
   }
 
   @Override
-  protected void initializeConfiguration(SqlConnectOptions connectOptions) {
-    DB2ConnectOptions options = (DB2ConnectOptions) connectOptions;
-    this.pipeliningLimit = options.getPipeliningLimit();
-  }
-
-  @Override
-  protected void configureNetClientOptions(NetClientOptions netClientOptions) {
-    // currently no-op
-  }
-
-  @Override
-  protected Future<Connection> doConnectInternal(SocketAddress server, String username, String password, String database, EventLoopContext context) {
+  protected Future<Connection> doConnectInternal(SqlConnectOptions options, EventLoopContext context) {
+    DB2ConnectOptions db2Options = DB2ConnectOptions.wrap(options);
+    SocketAddress server = options.getSocketAddress();
+    boolean cachePreparedStatements = options.getCachePreparedStatements();
+    int preparedStatementCacheSize = options.getPreparedStatementCacheMaxSize();
+    Predicate<String> preparedStatementCacheSqlFilter = options.getPreparedStatementCacheSqlFilter();
+    String username = options.getUser();
+    String password = options.getPassword();
+    String database = options.getDatabase();
+    Map<String, String> properties = options.getProperties();
+    int pipeliningLimit = db2Options.getPipeliningLimit();
+    NetClient netClient = netClient(options);
     return netClient.connect(server).flatMap(so -> {
-      DB2SocketConnection conn = new DB2SocketConnection((NetSocketInternal) so, cachePreparedStatements,
+      VertxMetrics vertxMetrics = vertx.metricsSPI();
+      ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(db2Options.getSocketAddress(), "sql", db2Options.getMetricsName()) : null;
+      DB2SocketConnection conn = new DB2SocketConnection((NetSocketInternal) so, metrics, db2Options, cachePreparedStatements,
         preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
       conn.init();
       return Future.future(p -> conn.sendStartupMessage(username, password, database, properties, p));
     });
   }
 
   @Override
-  public Future<SqlConnection> connect(Context context) {
+  public Future<SqlConnection> connect(Context context, SqlConnectOptions options) {
     ContextInternal contextInternal = (ContextInternal) context;
-    QueryTracer tracer = contextInternal.tracer() == null ? null : new QueryTracer(contextInternal.tracer(), options);
     Promise<SqlConnection> promise = contextInternal.promise();
-    connect(asEventLoopContext(contextInternal))
+    connect(asEventLoopContext(contextInternal), options)
       .map(conn -> {
-        DB2ConnectionImpl db2Connection = new DB2ConnectionImpl(contextInternal, this, conn, tracer, null);
+        DB2ConnectionImpl db2Connection = new DB2ConnectionImpl(contextInternal, this, conn);
         conn.init(db2Connection);
         return (SqlConnection)db2Connection;
       }).onComplete(promise);

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2ConnectionImpl.java

@@ -20,14 +20,13 @@
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.ContextInternal;
-import io.vertx.core.spi.metrics.ClientMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.db2client.DB2Connection;
 import io.vertx.db2client.impl.command.PingCommand;
 import io.vertx.db2client.spi.DB2Driver;
 import io.vertx.sqlclient.impl.Connection;
+import io.vertx.sqlclient.impl.SingletonSupplier;
 import io.vertx.sqlclient.impl.SqlConnectionBase;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 
 public class DB2ConnectionImpl extends SqlConnectionBase<DB2ConnectionImpl> implements DB2Connection {
@@ -36,16 +35,16 @@ public static Future<DB2Connection> connect(Vertx vertx, DB2ConnectOptions optio
     ContextInternal ctx = (ContextInternal) vertx.getOrCreateContext();
     DB2ConnectionFactory client;
     try {
-      client = new DB2ConnectionFactory(ctx.owner(), options);
+      client = new DB2ConnectionFactory(ctx.owner(), SingletonSupplier.wrap(options));
     } catch (Exception e) {
       return ctx.failedFuture(e);
     }
     ctx.addCloseHook(client);
     return (Future) client.connect(ctx);
   }
 
-  public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
-    super(context, factory, conn, DB2Driver.INSTANCE, tracer, metrics);
+  public DB2ConnectionImpl(ContextInternal context, ConnectionFactory factory, Connection conn) {
+    super(context, factory, conn, DB2Driver.INSTANCE);
   }
 
   @Override

vertx-db2-client/src/main/java/io/vertx/db2client/impl/DB2SocketConnection.java

@@ -22,12 +22,14 @@
 import io.vertx.core.AsyncResult;
 import io.vertx.core.Handler;
 import io.vertx.core.Promise;
-import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.EventLoopContext;
 import io.vertx.core.net.impl.NetSocketInternal;
+import io.vertx.core.spi.metrics.ClientMetrics;
+import io.vertx.db2client.DB2ConnectOptions;
 import io.vertx.db2client.impl.codec.DB2Codec;
 import io.vertx.db2client.impl.command.InitialHandshakeCommand;
 import io.vertx.db2client.impl.drda.ConnectionMetaData;
+import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.QueryResultHandler;
 import io.vertx.sqlclient.impl.SocketConnectionBase;
@@ -40,17 +42,21 @@
 
 public class DB2SocketConnection extends SocketConnectionBase {
 
+  private final DB2ConnectOptions connectOptions;
   private DB2Codec codec;
   private Handler<Void> closeHandler;
   public final ConnectionMetaData connMetadata = new ConnectionMetaData();
 
   public DB2SocketConnection(NetSocketInternal socket,
+      ClientMetrics clientMetrics,
+      DB2ConnectOptions connectOptions,
       boolean cachePreparedStatements,
       int preparedStatementCacheSize,
       Predicate<String> preparedStatementCacheSqlFilter,
       int pipeliningLimit,
       EventLoopContext context) {
-    super(socket, cachePreparedStatements, preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
+    super(socket, clientMetrics, cachePreparedStatements, preparedStatementCacheSize, preparedStatementCacheSqlFilter, pipeliningLimit, context);
+    this.connectOptions = connectOptions;
   }
 
   // TODO RETURN FUTURE ???
@@ -63,6 +69,11 @@ void sendStartupMessage(String username,
     schedule(context, cmd).onComplete(completionHandler);
   }
 
+  @Override
+  protected SqlConnectOptions connectOptions() {
+    return connectOptions;
+  }
+
   @Override
   public void init() {
     codec = new DB2Codec(this);

vertx-db2-client/src/main/java/io/vertx/db2client/spi/DB2Driver.java

@@ -15,27 +15,25 @@
  */
 package io.vertx.db2client.spi;
 
+import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.core.impl.CloseFuture;
 import io.vertx.core.impl.ContextInternal;
 import io.vertx.core.impl.VertxInternal;
 import io.vertx.core.json.JsonObject;
-import io.vertx.core.spi.metrics.ClientMetrics;
-import io.vertx.core.spi.metrics.VertxMetrics;
 import io.vertx.db2client.DB2ConnectOptions;
-import io.vertx.db2client.DB2Pool;
 import io.vertx.db2client.impl.*;
+import io.vertx.sqlclient.Pool;
 import io.vertx.sqlclient.PoolOptions;
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.impl.Connection;
 import io.vertx.sqlclient.impl.PoolImpl;
+import io.vertx.sqlclient.impl.SingletonSupplier;
 import io.vertx.sqlclient.impl.SqlConnectionInternal;
-import io.vertx.sqlclient.impl.tracing.QueryTracer;
 import io.vertx.sqlclient.spi.ConnectionFactory;
 import io.vertx.sqlclient.spi.Driver;
 
-import java.util.List;
-import java.util.stream.Collectors;
+import java.util.function.Supplier;
 
 public class DB2Driver implements Driver {
 
@@ -44,7 +42,7 @@ public class DB2Driver implements Driver {
   public static final DB2Driver INSTANCE = new DB2Driver();
 
   @Override
-  public DB2Pool newPool(Vertx vertx, List<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
+  public Pool newPool(Vertx vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
     VertxInternal vx = (VertxInternal) vertx;
     PoolImpl pool;
     if (options.isShared()) {
@@ -55,16 +53,10 @@ public DB2Pool newPool(Vertx vertx, List<? extends SqlConnectOptions> databases,
     return new DB2PoolImpl(vx, closeFuture, pool);
   }
 
-  private PoolImpl newPoolImpl(VertxInternal vertx, List<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
-    DB2ConnectOptions baseConnectOptions = DB2ConnectOptions.wrap(databases.get(0));
-    QueryTracer tracer = vertx.tracer() == null ? null : new QueryTracer(vertx.tracer(), baseConnectOptions);
-    VertxMetrics vertxMetrics = vertx.metricsSPI();
-    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(baseConnectOptions.getSocketAddress(), "sql", baseConnectOptions.getMetricsName()) : null;
+  private PoolImpl newPoolImpl(VertxInternal vertx, Supplier<? extends Future<? extends SqlConnectOptions>> databases, PoolOptions options, CloseFuture closeFuture) {
     boolean pipelinedPool = options instanceof Db2PoolOptions && ((Db2PoolOptions) options).isPipelined();
-    int pipeliningLimit = pipelinedPool ? baseConnectOptions.getPipeliningLimit() : 1;
-    PoolImpl pool = new PoolImpl(vertx, this, tracer, metrics, pipeliningLimit, options, null, null, closeFuture);
-    List<ConnectionFactory> lst = databases.stream().map(o -> createConnectionFactory(vertx, o)).collect(Collectors.toList());
-    ConnectionFactory factory = ConnectionFactory.roundRobinSelector(lst);
+    PoolImpl pool = new PoolImpl(vertx, this, pipelinedPool, options, null, null, closeFuture);
+    ConnectionFactory factory = createConnectionFactory(vertx, databases);
     pool.connectionProvider(factory::connect);
     pool.init();
     closeFuture.add(factory);
@@ -84,11 +76,16 @@ public boolean acceptsOptions(SqlConnectOptions options) {
 
   @Override
   public ConnectionFactory createConnectionFactory(Vertx vertx, SqlConnectOptions database) {
-    return new DB2ConnectionFactory((VertxInternal) vertx, DB2ConnectOptions.wrap(database));
+    return new DB2ConnectionFactory((VertxInternal) vertx, SingletonSupplier.wrap(database));
   }
 
   @Override
-  public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection conn, QueryTracer tracer, ClientMetrics metrics) {
-    return new DB2ConnectionImpl(context, factory, conn, tracer, metrics);
+  public ConnectionFactory createConnectionFactory(Vertx vertx, Supplier<? extends Future<? extends SqlConnectOptions>> database) {
+    return new DB2ConnectionFactory((VertxInternal) vertx, database);
+  }
+
+  @Override
+  public SqlConnectionInternal wrapConnection(ContextInternal context, ConnectionFactory factory, Connection conn) {
+    return new DB2ConnectionImpl(context, factory, conn);
   }
 }

vertx-mssql-client/src/main/java/examples/SqlClientExamples.java

@@ -416,15 +416,11 @@ public static void poolSharing3(Vertx vertx, MSSQLConnectOptions database, int m
       .setEventLoopSize(4));
   }
 
-  public void dynamicPoolConfig(Vertx vertx, MSSQLPool pool) {
-    pool.connectionProvider(ctx -> {
-      Future<MSSQLConnectOptions> fut = retrieveOptions();
-      return fut.compose(connectOptions -> {
-        // Do not forget to close later
-        ConnectionFactory factory = MSSQLDriver.INSTANCE.createConnectionFactory(vertx, connectOptions);
-        return factory.connect(ctx);
-      });
-    });
+  public void dynamicPoolConfig(Vertx vertx, PoolOptions poolOptions) {
+    MSSQLPool pool = MSSQLPool.pool(vertx, () -> {
+      Future<MSSQLConnectOptions> connectOptions = retrieveOptions();
+      return connectOptions;
+    }, poolOptions);
   }
 
   private Future<MSSQLConnectOptions> retrieveOptions() {

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/MSSQLPool.java

@@ -12,17 +12,19 @@
 package io.vertx.mssqlclient;
 
 import io.vertx.codegen.annotations.Fluent;
+import io.vertx.codegen.annotations.GenIgnore;
 import io.vertx.codegen.annotations.VertxGen;
 import io.vertx.core.Context;
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.mssqlclient.spi.MSSQLDriver;
 import io.vertx.sqlclient.*;
+import io.vertx.sqlclient.impl.SingletonSupplier;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.function.Function;
+import java.util.function.Supplier;
 
 import static io.vertx.mssqlclient.MSSQLConnectOptions.fromUri;
 
@@ -75,7 +77,7 @@ static MSSQLPool pool(MSSQLConnectOptions database, PoolOptions options) {
    * Like {@link #pool(MSSQLConnectOptions, PoolOptions)} with a specific {@link Vertx} instance.
    */
   static MSSQLPool pool(Vertx vertx, MSSQLConnectOptions database, PoolOptions options) {
-    return pool(vertx, Collections.singletonList(database), options);
+    return pool(vertx, SingletonSupplier.wrap(database), options);
   }
 
   /**
@@ -97,6 +99,27 @@ static MSSQLPool pool(Vertx vertx, List<MSSQLConnectOptions> databases, PoolOpti
     return (MSSQLPool) MSSQLDriver.INSTANCE.createPool(vertx, databases, options);
   }
 
+  /**
+   * Create a connection pool to the SQL Server {@code databases}. The supplier is called
+   * to provide the options when a new connection is created by the pool.
+   *
+   * @param databases the databases supplier
+   * @param options the options for creating the pool
+   * @return the connection pool
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  static MSSQLPool pool(Supplier<Future<MSSQLConnectOptions>> databases, PoolOptions options) {
+    return pool(null, databases, options);
+  }
+
+  /**
+   * Like {@link #pool(Supplier, PoolOptions)} with a specific {@link Vertx} instance.
+   */
+  @GenIgnore(GenIgnore.PERMITTED_TYPE)
+  static MSSQLPool pool(Vertx vertx, Supplier<Future<MSSQLConnectOptions>> databases, PoolOptions options) {
+    return (MSSQLPool) MSSQLDriver.INSTANCE.createPool(vertx, databases, options);
+  }
+
   @Override
   MSSQLPool connectHandler(Handler<SqlConnection> handler);