Implement reference counted pools - fixes #1094

Author: vietj

vertx-db2-client/src/main/asciidoc/index.adoc

@@ -109,6 +109,10 @@ The {@link io.vertx.db2client.DB2Pool} allows you to create a pool or a pooled c
 - pool operations are not pipelined, only connection client are pipelined
 - pooled client operations are pipelined
 
+== Pool sharing
+
+include::pool_sharing.adoc[]
+
 == Configuration
 
 There are several alternatives for you to configure the client.

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

@@ -19,6 +19,8 @@
 import java.util.Arrays;
 import java.util.List;
 
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.DeploymentOptions;
 import io.vertx.core.Vertx;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.db2client.DB2ConnectOptions;
@@ -357,4 +359,30 @@ public void poolConfig02(DB2Pool pool, String sql) {
       });
     });
   }
+
+  public void poolSharing1(Vertx vertx, DB2ConnectOptions database, int maxSize) {
+    Pool pool = DB2Pool.pool(database, new PoolOptions().setMaxSize(maxSize));
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      @Override
+      public void start() throws Exception {
+        // Use the pool
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
+
+  public void poolSharing2(Vertx vertx, DB2ConnectOptions database, int maxSize) {
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      Pool pool;
+      @Override
+      public void start() {
+        // Get or create a shared pool
+        // this actually creates a lease to the pool
+        // when the verticle is undeployed, the lease will be released automaticaly
+        pool = DB2Pool.pool(database, new PoolOptions()
+          .setMaxSize(maxSize)
+          .setShared(true)
+          .setName("my-pool"));
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
 }

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

@@ -15,13 +15,15 @@
  */
 package io.vertx.db2client.impl;
 
+import io.vertx.core.impl.CloseFuture;
+import io.vertx.core.impl.VertxInternal;
 import io.vertx.db2client.DB2Pool;
 import io.vertx.sqlclient.Pool;
 import io.vertx.sqlclient.impl.PoolBase;
 
 public class DB2PoolImpl extends PoolBase<DB2PoolImpl> implements DB2Pool {
 
-  public DB2PoolImpl(Pool delegate) {
-    super(delegate);
+  public DB2PoolImpl(VertxInternal vertx, CloseFuture closeFuture, Pool delegate) {
+    super(vertx, closeFuture, delegate);
   }
 }

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

@@ -41,24 +41,36 @@
 
 public class DB2Driver implements Driver {
 
+  private static final String SHARED_CLIENT_KEY = "__vertx.shared.db2client";
+
   public static final DB2Driver INSTANCE = new DB2Driver();
 
   @Override
   public DB2Pool newPool(Vertx vertx, List<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
     VertxInternal vx = (VertxInternal) vertx;
+    PoolImpl pool;
+    if (options.isShared()) {
+      pool = vx.createSharedClient(SHARED_CLIENT_KEY, options.getName(), closeFuture, cf -> newPoolImpl(vx, databases, options, cf));
+    } else {
+      pool = newPoolImpl(vx, databases, options, closeFuture);
+    }
+    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 = vx.tracer() == null ? null : new QueryTracer(vx.tracer(), baseConnectOptions);
-    VertxMetrics vertxMetrics = vx.metricsSPI();
+    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;
     boolean pipelinedPool = options instanceof Db2PoolOptions && ((Db2PoolOptions) options).isPipelined();
     int pipeliningLimit = pipelinedPool ? baseConnectOptions.getPipeliningLimit() : 1;
-    ClientMetrics metrics = vertxMetrics != null ? vertxMetrics.createClientMetrics(baseConnectOptions.getSocketAddress(), "sql", baseConnectOptions.getMetricsName()) : null;
-    PoolImpl pool = new PoolImpl(vx, this, baseConnectOptions, null, tracer, metrics, pipeliningLimit, options, closeFuture);
-    pool.init();
+    PoolImpl pool = new PoolImpl(vertx, this, baseConnectOptions, null, tracer, metrics, pipeliningLimit, options, closeFuture);
     List<ConnectionFactory> lst = databases.stream().map(o -> createConnectionFactory(vertx, o)).collect(Collectors.toList());
     ConnectionFactory factory = ConnectionFactory.roundRobinSelector(lst);
     pool.connectionProvider(factory::connect);
+    pool.init();
     closeFuture.add(factory);
-    return new DB2PoolImpl(pool);
+    return pool;
   }
 
   @Override

vertx-mssql-client/src/main/asciidoc/index.adoc

@@ -279,6 +279,11 @@ For example, {@link io.vertx.core.net.PemTrustOptions} can be used if a PEM file
 
 For further details about SSL support in Vert.x, please refer to the https://vertx.io/docs/vertx-core/java/#ssl[Vert.x Core documentation].
 
+== Pool sharing
+
+include::pool_sharing.adoc[]
+
 == Advanced pool configuration
 
 include::pool_config.adoc[]
+

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

@@ -16,6 +16,8 @@
  */
 package examples;
 
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.DeploymentOptions;
 import io.vertx.core.Vertx;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.docgen.Source;
@@ -356,4 +358,30 @@ public void poolConfig02(MSSQLPool pool, String sql) {
       });
     });
   }
+
+  public void poolSharing1(Vertx vertx, MSSQLConnectOptions database, int maxSize) {
+    Pool pool = MSSQLPool.pool(database, new PoolOptions().setMaxSize(maxSize));
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      @Override
+      public void start() throws Exception {
+        // Use the pool
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
+
+  public void poolSharing2(Vertx vertx, MSSQLConnectOptions database, int maxSize) {
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      Pool pool;
+      @Override
+      public void start() {
+        // Get or create a shared pool
+        // this actually creates a lease to the pool
+        // when the verticle is undeployed, the lease will be released automaticaly
+        pool = MSSQLPool.pool(database, new PoolOptions()
+          .setMaxSize(maxSize)
+          .setShared(true)
+          .setName("my-pool"));
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
 }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/impl/MSSQLPoolImpl.java

@@ -11,13 +11,15 @@
 
 package io.vertx.mssqlclient.impl;
 
+import io.vertx.core.impl.CloseFuture;
+import io.vertx.core.impl.VertxInternal;
 import io.vertx.mssqlclient.MSSQLPool;
 import io.vertx.sqlclient.Pool;
 import io.vertx.sqlclient.impl.PoolBase;
 
 public class MSSQLPoolImpl extends PoolBase<MSSQLPoolImpl> implements MSSQLPool {
 
-  public MSSQLPoolImpl(Pool delegate) {
-    super(delegate);
+  public MSSQLPoolImpl(VertxInternal vertx, CloseFuture closeFuture, Pool delegate) {
+    super(vertx, closeFuture, delegate);
   }
 }

vertx-mssql-client/src/main/java/io/vertx/mssqlclient/spi/MSSQLDriver.java

@@ -40,22 +40,34 @@
 
 public class MSSQLDriver implements Driver {
 
+  private static final String SHARED_CLIENT_KEY = "__vertx.shared.mssqlclient";
+
   public static final MSSQLDriver INSTANCE = new MSSQLDriver();
 
   @Override
   public MSSQLPool newPool(Vertx vertx, List<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
     VertxInternal vx = (VertxInternal) vertx;
+    PoolImpl pool;
+    if (options.isShared()) {
+      pool = vx.createSharedClient(SHARED_CLIENT_KEY, options.getName(), closeFuture, cf -> newPoolImpl(vx, databases, options, cf));
+    } else {
+      pool = newPoolImpl(vx, databases, options, closeFuture);
+    }
+    return new MSSQLPoolImpl(vx, closeFuture, pool);
+  }
+
+  private PoolImpl newPoolImpl(VertxInternal vertx, List<? extends SqlConnectOptions> databases, PoolOptions options, CloseFuture closeFuture) {
     MSSQLConnectOptions baseConnectOptions = MSSQLConnectOptions.wrap(databases.get(0));
-    QueryTracer tracer = vx.tracer() == null ? null : new QueryTracer(vx.tracer(), baseConnectOptions);
-    VertxMetrics vertxMetrics = vx.metricsSPI();
+    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;
-    PoolImpl pool = new PoolImpl(vx, this, baseConnectOptions, null, tracer, metrics, 1, options, closeFuture);
-    pool.init();
+    PoolImpl pool = new PoolImpl(vertx, this, baseConnectOptions, null, tracer, metrics, 1, options, closeFuture);
     List<ConnectionFactory> lst = databases.stream().map(o -> createConnectionFactory(vertx, o)).collect(Collectors.toList());
     ConnectionFactory factory = ConnectionFactory.roundRobinSelector(lst);
     pool.connectionProvider(factory::connect);
+    pool.init();
     closeFuture.add(factory);
-    return new MSSQLPoolImpl(pool);
+    return pool;
   }
 
   @Override

vertx-mysql-client/src/main/asciidoc/index.adoc

@@ -92,6 +92,10 @@ You can easily get one from the pool:
 
 Once you are done with the connection you must close it to release it to the pool, so it can be reused.
 
+== Pool sharing
+
+include::pool_sharing.adoc[]
+
 === Unix Domain Socket
 
 Sometimes for simplicity, security or performance reasons, it is required to connect via a https://dev.mysql.com/doc/refman/8.0/en/server-system-variables.html#sysvar_socket[Unix Domain Socket].

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

@@ -16,6 +16,8 @@
  */
 package examples;
 
+import io.vertx.core.AbstractVerticle;
+import io.vertx.core.DeploymentOptions;
 import io.vertx.core.Vertx;
 import io.vertx.core.tracing.TracingPolicy;
 import io.vertx.docgen.Source;
@@ -341,4 +343,30 @@ public void poolConfig02(MySQLPool pool, String sql) {
       });
     });
   }
+
+  public void poolSharing1(Vertx vertx, MySQLConnectOptions database, int maxSize) {
+    Pool pool = MySQLPool.pool(database, new PoolOptions().setMaxSize(maxSize));
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      @Override
+      public void start() throws Exception {
+        // Use the pool
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
+
+  public void poolSharing2(Vertx vertx, MySQLConnectOptions database, int maxSize) {
+    vertx.deployVerticle(() -> new AbstractVerticle() {
+      Pool pool;
+      @Override
+      public void start() {
+        // Get or create a shared pool
+        // this actually creates a lease to the pool
+        // when the verticle is undeployed, the lease will be released automaticaly
+        pool = MySQLPool.pool(database, new PoolOptions()
+          .setMaxSize(maxSize)
+          .setShared(true)
+          .setName("my-pool"));
+      }
+    }, new DeploymentOptions().setInstances(4));
+  }
 }