@Test public void testRetriesAndSuccess(TestContext context) { final int MAX_RETRIES = 5; final Async async = context.async(); final AsyncConnectionPoolMock pool = new AsyncConnectionPoolMock( new JsonObject() .put("maxConnectionRetries", MAX_RETRIES) .put("connectionRetryDelay", 100L), new Supplier<Connection>() { int count=0; @Override public Connection get() { count++; if (count < MAX_RETRIES + 1) return getFailingConnection(); else return getGoodConnection(); } }); pool.take(result -> { context.assertTrue(result.succeeded()); context.assertEquals(MAX_RETRIES + 1, pool.connectionAttempts); context.assertEquals(1, pool.createdConnections); Mockito.verify(vertx, Mockito.times(MAX_RETRIES)).setTimer(Mockito.eq(100L), Mockito.any()); async.complete(); }); }
public synchronized void take(Handler<AsyncResult<Connection>> handler) { Connection connection = availableConnections.poll(); if (connection == null) { createOrWaitForAvailableConnection(handler); } else { cancelTimer(connection); if (connection.isConnected()) { if (connectionConfig != null && connectionConfig.getConnectionTestTimeout() > 0) { AtomicBoolean testCompleted = new AtomicBoolean(false); long timer = vertx.setTimer(connectionConfig.getConnectionTestTimeout(), ignored -> { if (testCompleted.compareAndSet(false, true)) { logger.info("connection test timeout"); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } }); connection.sendQuery("SELECT 1 AS alive") .whenCompleteAsync((ignored, error) -> { if (error != null) { logger.info("connection test failed", error); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } else { if (testCompleted.compareAndSet(false, true)) { if (this.connectionConfig.getConnectionTestTimeout() > 0) { vertx.cancelTimer(timer); } handler.handle(Future.succeededFuture(connection)); } } }, ConversionUtils.vertxToExecutor(vertx)); } else { handler.handle(Future.succeededFuture(connection)); } } else { poolSize -= 1; take(handler); } } }
AsyncConnectionPool { public synchronized void take(Handler<AsyncResult<Connection>> handler) { Connection connection = availableConnections.poll(); if (connection == null) { createOrWaitForAvailableConnection(handler); } else { cancelTimer(connection); if (connection.isConnected()) { if (connectionConfig != null && connectionConfig.getConnectionTestTimeout() > 0) { AtomicBoolean testCompleted = new AtomicBoolean(false); long timer = vertx.setTimer(connectionConfig.getConnectionTestTimeout(), ignored -> { if (testCompleted.compareAndSet(false, true)) { logger.info("connection test timeout"); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } }); connection.sendQuery("SELECT 1 AS alive") .whenCompleteAsync((ignored, error) -> { if (error != null) { logger.info("connection test failed", error); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } else { if (testCompleted.compareAndSet(false, true)) { if (this.connectionConfig.getConnectionTestTimeout() > 0) { vertx.cancelTimer(timer); } handler.handle(Future.succeededFuture(connection)); } } }, ConversionUtils.vertxToExecutor(vertx)); } else { handler.handle(Future.succeededFuture(connection)); } } else { poolSize -= 1; take(handler); } } } }
AsyncConnectionPool { public synchronized void take(Handler<AsyncResult<Connection>> handler) { Connection connection = availableConnections.poll(); if (connection == null) { createOrWaitForAvailableConnection(handler); } else { cancelTimer(connection); if (connection.isConnected()) { if (connectionConfig != null && connectionConfig.getConnectionTestTimeout() > 0) { AtomicBoolean testCompleted = new AtomicBoolean(false); long timer = vertx.setTimer(connectionConfig.getConnectionTestTimeout(), ignored -> { if (testCompleted.compareAndSet(false, true)) { logger.info("connection test timeout"); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } }); connection.sendQuery("SELECT 1 AS alive") .whenCompleteAsync((ignored, error) -> { if (error != null) { logger.info("connection test failed", error); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } else { if (testCompleted.compareAndSet(false, true)) { if (this.connectionConfig.getConnectionTestTimeout() > 0) { vertx.cancelTimer(timer); } handler.handle(Future.succeededFuture(connection)); } } }, ConversionUtils.vertxToExecutor(vertx)); } else { handler.handle(Future.succeededFuture(connection)); } } else { poolSize -= 1; take(handler); } } } AsyncConnectionPool(Vertx vertx, JsonObject globalConfig, ConnectionPoolConfiguration connectionConfig); }
AsyncConnectionPool { public synchronized void take(Handler<AsyncResult<Connection>> handler) { Connection connection = availableConnections.poll(); if (connection == null) { createOrWaitForAvailableConnection(handler); } else { cancelTimer(connection); if (connection.isConnected()) { if (connectionConfig != null && connectionConfig.getConnectionTestTimeout() > 0) { AtomicBoolean testCompleted = new AtomicBoolean(false); long timer = vertx.setTimer(connectionConfig.getConnectionTestTimeout(), ignored -> { if (testCompleted.compareAndSet(false, true)) { logger.info("connection test timeout"); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } }); connection.sendQuery("SELECT 1 AS alive") .whenCompleteAsync((ignored, error) -> { if (error != null) { logger.info("connection test failed", error); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } else { if (testCompleted.compareAndSet(false, true)) { if (this.connectionConfig.getConnectionTestTimeout() > 0) { vertx.cancelTimer(timer); } handler.handle(Future.succeededFuture(connection)); } } }, ConversionUtils.vertxToExecutor(vertx)); } else { handler.handle(Future.succeededFuture(connection)); } } else { poolSize -= 1; take(handler); } } } AsyncConnectionPool(Vertx vertx, JsonObject globalConfig, ConnectionPoolConfiguration connectionConfig); synchronized int getPoolSize(); synchronized void take(Handler<AsyncResult<Connection>> handler); synchronized void expire(Connection connection); synchronized void giveBack(Connection connection); synchronized void close(); synchronized void close(Handler<AsyncResult<Void>> handler); }
AsyncConnectionPool { public synchronized void take(Handler<AsyncResult<Connection>> handler) { Connection connection = availableConnections.poll(); if (connection == null) { createOrWaitForAvailableConnection(handler); } else { cancelTimer(connection); if (connection.isConnected()) { if (connectionConfig != null && connectionConfig.getConnectionTestTimeout() > 0) { AtomicBoolean testCompleted = new AtomicBoolean(false); long timer = vertx.setTimer(connectionConfig.getConnectionTestTimeout(), ignored -> { if (testCompleted.compareAndSet(false, true)) { logger.info("connection test timeout"); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } }); connection.sendQuery("SELECT 1 AS alive") .whenCompleteAsync((ignored, error) -> { if (error != null) { logger.info("connection test failed", error); connection.disconnect(); synchronized (this) { poolSize -= 1; } take(handler); } else { if (testCompleted.compareAndSet(false, true)) { if (this.connectionConfig.getConnectionTestTimeout() > 0) { vertx.cancelTimer(timer); } handler.handle(Future.succeededFuture(connection)); } } }, ConversionUtils.vertxToExecutor(vertx)); } else { handler.handle(Future.succeededFuture(connection)); } } else { poolSize -= 1; take(handler); } } } AsyncConnectionPool(Vertx vertx, JsonObject globalConfig, ConnectionPoolConfiguration connectionConfig); synchronized int getPoolSize(); synchronized void take(Handler<AsyncResult<Connection>> handler); synchronized void expire(Connection connection); synchronized void giveBack(Connection connection); synchronized void close(); synchronized void close(Handler<AsyncResult<Void>> handler); static final int DEFAULT_MAX_POOL_SIZE; static final int DEFAULT_MAX_CONNECTION_RETRIES; static final int DEFAULT_CONNECTION_RETRY_DELAY; static final int DEFAULT_CONNECTION_RELEASE_DELAY; }