diff --git a/vertx-mssql-client/src/test/java/io/vertx/mssqlclient/tck/MSSQLMetricsTest.java b/vertx-mssql-client/src/test/java/io/vertx/mssqlclient/tck/MSSQLMetricsTest.java index 4d2bd3c8e2..4a7eb17870 100644 --- a/vertx-mssql-client/src/test/java/io/vertx/mssqlclient/tck/MSSQLMetricsTest.java +++ b/vertx-mssql-client/src/test/java/io/vertx/mssqlclient/tck/MSSQLMetricsTest.java @@ -12,9 +12,11 @@ package io.vertx.mssqlclient.tck; import io.vertx.mssqlclient.MSSQLBuilder; +import io.vertx.mssqlclient.MSSQLConnectOptions; import io.vertx.mssqlclient.junit.MSSQLRule; import io.vertx.sqlclient.ClientBuilder; import io.vertx.sqlclient.Pool; +import io.vertx.sqlclient.SqlConnectOptions; import io.vertx.sqlclient.tck.MetricsTestBase; import org.junit.ClassRule; @@ -23,9 +25,14 @@ public class MSSQLMetricsTest extends MetricsTestBase { @ClassRule public static MSSQLRule rule = MSSQLRule.SHARED_INSTANCE; + @Override + protected SqlConnectOptions connectOptions() { + return new MSSQLConnectOptions(rule.options()); + } + @Override protected ClientBuilder poolBuilder() { - return MSSQLBuilder.pool().connectingTo(rule.options()); + return MSSQLBuilder.pool().connectingTo(connectOptions()); } @Override diff --git a/vertx-oracle-client/src/test/java/io/vertx/oracleclient/test/tck/OracleMetricsTest.java b/vertx-oracle-client/src/test/java/io/vertx/oracleclient/test/tck/OracleMetricsTest.java index 389385203c..94d53fdecf 100644 --- a/vertx-oracle-client/src/test/java/io/vertx/oracleclient/test/tck/OracleMetricsTest.java +++ b/vertx-oracle-client/src/test/java/io/vertx/oracleclient/test/tck/OracleMetricsTest.java @@ -13,9 +13,11 @@ import io.vertx.ext.unit.TestContext; import io.vertx.oracleclient.OracleBuilder; +import io.vertx.oracleclient.OracleConnectOptions; import io.vertx.oracleclient.test.junit.OracleRule; import io.vertx.sqlclient.ClientBuilder; import io.vertx.sqlclient.Pool; +import io.vertx.sqlclient.SqlConnectOptions; import io.vertx.sqlclient.tck.MetricsTestBase; import org.junit.ClassRule; import org.junit.Ignore; @@ -26,9 +28,14 @@ public class OracleMetricsTest extends MetricsTestBase { @ClassRule public static OracleRule rule = OracleRule.SHARED_INSTANCE; + @Override + protected SqlConnectOptions connectOptions() { + return new OracleConnectOptions(rule.options()); + } + @Override protected ClientBuilder poolBuilder() { - return OracleBuilder.pool().connectingTo(rule.options()); + return OracleBuilder.pool().connectingTo(connectOptions()); } @Override @@ -49,4 +56,25 @@ public void testPreparedBatchQuery(TestContext ctx) { public void testPrepareAndBatchQuery(TestContext ctx) { super.testPrepareAndBatchQuery(ctx); } + + @Test + @Ignore("Implementation of the test does not work with Oracle") + @Override + public void testQueuing(TestContext ctx) throws Exception { + super.testQueuing(ctx); + } + + @Test + @Ignore("Implementation of the test does not work with Oracle") + @Override + public void testQueuingTimeout(TestContext ctx) throws Exception { + super.testQueuingTimeout(ctx); + } + + @Test + @Ignore("Implementation of the test does not work with Oracle") + @Override + public void testConnectionLost(TestContext ctx) throws Exception { + super.testConnectionLost(ctx); + } } diff --git a/vertx-pg-client/src/test/java/io/vertx/pgclient/PgMetricsTest.java b/vertx-pg-client/src/test/java/io/vertx/pgclient/PgMetricsTest.java index 4d009cdc5d..17c799a670 100644 --- a/vertx-pg-client/src/test/java/io/vertx/pgclient/PgMetricsTest.java +++ b/vertx-pg-client/src/test/java/io/vertx/pgclient/PgMetricsTest.java @@ -14,6 +14,7 @@ import io.vertx.pgclient.junit.ContainerPgRule; import io.vertx.sqlclient.ClientBuilder; import io.vertx.sqlclient.Pool; +import io.vertx.sqlclient.SqlConnectOptions; import io.vertx.sqlclient.tck.MetricsTestBase; import org.junit.ClassRule; @@ -22,9 +23,14 @@ public class PgMetricsTest extends MetricsTestBase { @ClassRule public static ContainerPgRule rule = new ContainerPgRule(); + @Override + protected SqlConnectOptions connectOptions() { + return new PgConnectOptions(rule.options()); + } + @Override protected ClientBuilder poolBuilder() { - return PgBuilder.pool().connectingTo(rule.options()); + return PgBuilder.pool().connectingTo(connectOptions()); } @Override 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 index bf2a2ce626..ad3157ebf5 100644 --- 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 @@ -186,6 +186,16 @@ private Object dequeueAndBeginUse(Object metric) { return NO_METRICS; } + private void dequeueAndReject(Object metric) { + if (metrics != null && metric != NO_METRICS) { + try { + metrics.rejected(metric); + } catch (Exception e) { + // Log + } + } + } + private Object endUse(Object metric) { if (metrics != null && metric != NO_METRICS) { try { @@ -271,6 +281,7 @@ public void onEnqueue(PoolWaiter waiter) { pool.cancel(waiter, ar -> { if (ar.succeeded()) { if (ar.result()) { + dequeueAndReject(queueMetric); handler.handle(Future.failedFuture("Timeout")); } } else { diff --git a/vertx-sql-client/src/test/java/io/vertx/sqlclient/tck/MetricsTestBase.java b/vertx-sql-client/src/test/java/io/vertx/sqlclient/tck/MetricsTestBase.java index 312647b2d9..fdd8fa174e 100644 --- a/vertx-sql-client/src/test/java/io/vertx/sqlclient/tck/MetricsTestBase.java +++ b/vertx-sql-client/src/test/java/io/vertx/sqlclient/tck/MetricsTestBase.java @@ -36,6 +36,9 @@ import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; + @RunWith(VertxUnitRunner.class) public abstract class MetricsTestBase { @@ -81,6 +84,8 @@ protected Pool getPool() { return pool; } + protected abstract SqlConnectOptions connectOptions(); + protected abstract ClientBuilder poolBuilder(); protected Pool createPool(Vertx vertx) { @@ -122,6 +127,15 @@ public void close() { @Test public void testQueuing(TestContext ctx) throws Exception { + testQueuing(ctx, false); + } + + @Test + public void testQueuingTimeout(TestContext ctx) throws Exception { + testQueuing(ctx, true); + } + + private void testQueuing(TestContext ctx, boolean timeout) throws Exception { AtomicInteger queueSize = new AtomicInteger(); AtomicInteger inUse = new AtomicInteger(); List enqueueMetrics = Collections.synchronizedList(new ArrayList<>()); @@ -156,28 +170,106 @@ public void end(Object inUseMetric, boolean succeeded) { endMetrics.add(inUseMetric); } }; - Pool pool = createPool(vertx, new PoolOptions().setMaxSize(1).setName("the-pool")); - SqlConnection conn = pool.getConnection().toCompletionStage().toCompletableFuture().get(20, TimeUnit.SECONDS); + PoolOptions poolOptions = new PoolOptions().setMaxSize(1).setName("the-pool"); + if (timeout) { + poolOptions.setConnectionTimeout(2).setConnectionTimeoutUnit(SECONDS); + } + Pool pool = createPool(vertx, poolOptions); + SqlConnection conn = Future.await(pool.getConnection(), 20, SECONDS); int num = 16; List> futures = new ArrayList<>(); for (int i = 0;i < num;i++) { - futures.add(pool.query("SELECT * FROM immutable WHERE id=1").execute()); - } - long now = System.currentTimeMillis(); - while (queueSize.get() != num) { - ctx.assertTrue(System.currentTimeMillis() - now < 20_000); - Thread.sleep(100); + futures.add(pool.withConnection(sqlConn -> sqlConn.query("SELECT * FROM immutable WHERE id=1").execute())); } - conn.close(); - Future.join(futures).toCompletionStage().toCompletableFuture().get(20, TimeUnit.SECONDS); + awaitQueueSize(ctx, queueSize, timeout ? 0 : num); + Future.await(conn.close(), 20, SECONDS); + Future.await(Future.join(futures).otherwiseEmpty(), 20, SECONDS); ctx.assertEquals(0, queueSize.get()); ctx.assertEquals(0, inUse.get()); - ctx.assertEquals(enqueueMetrics, dequeueMetrics); + if (timeout) { + ctx.assertTrue(enqueueMetrics.containsAll(dequeueMetrics) && dequeueMetrics.containsAll(enqueueMetrics)); + } else { + ctx.assertEquals(enqueueMetrics, dequeueMetrics); + } ctx.assertEquals(beginMetrics, endMetrics); ctx.assertEquals("sql", poolType); ctx.assertEquals("the-pool", poolName); } + private void awaitQueueSize(TestContext ctx, AtomicInteger queueSize, int num) throws InterruptedException { + long now = System.currentTimeMillis(); + for (; ; ) { + if (queueSize.get() != num) { + if (System.currentTimeMillis() - now >= 20_000) { + ctx.fail("Timeout waiting for queue size " + queueSize.get() + " to be equal to " + num); + } else { + MILLISECONDS.sleep(500); + } + } else { + break; + } + } + } + + @Test + public void testConnectionLost(TestContext ctx) throws Exception { + SqlConnectOptions connectOptions = connectOptions(); + ProxyServer proxy = ProxyServer.create(vertx, connectOptions.getPort(), connectOptions.getHost()); + AtomicReference firstConnection = new AtomicReference<>(); + proxy.proxyHandler(proxiedConn -> { + if (firstConnection.compareAndSet(null, proxiedConn)) { + proxiedConn.connect(); + } + }); + // Start proxy + Async listenLatch = ctx.async(); + proxy.listen(8080, "localhost", ctx.asyncAssertSuccess(res -> listenLatch.complete())); + listenLatch.awaitSuccess(20_000); + + + AtomicInteger queueSize = new AtomicInteger(); + poolMetrics = new PoolMetrics() { + @Override + public Object submitted() { + queueSize.incrementAndGet(); + return null; + } + + @Override + public Object begin(Object o) { + queueSize.decrementAndGet(); + return null; + } + + @Override + public void rejected(Object o) { + queueSize.decrementAndGet(); + } + }; + PoolOptions poolOptions = new PoolOptions() + .setConnectionTimeout(500) + .setConnectionTimeoutUnit(MILLISECONDS) + .setMaxSize(1) + .setName("the-pool"); + Pool pool = poolBuilder() + .with(poolOptions) + .using(vertx) + .connectingTo(connectOptions.setHost("localhost").setPort(8080)) + .build(); + SqlConnection conn = Future.await(pool.getConnection(), 20, SECONDS); + int num = 16; + Async async = ctx.async(num + 1); + for (int i = 0; i < num; i++) { + pool.withConnection(sqlConn -> sqlConn.query("SELECT * FROM immutable WHERE id=1").execute()) + .onComplete(ctx.asyncAssertFailure(t -> async.countDown())); + } + conn.closeHandler(v -> async.countDown()); + awaitQueueSize(ctx, queueSize, 16); + firstConnection.get().clientSocket().close(); + async.await(20_000); + ctx.assertEquals(0, queueSize.get()); + } + @Test public void testSimpleQuery(TestContext ctx) { Function> fn = conn -> conn.query("SELECT * FROM immutable WHERE id=1").execute();