vert-x3#108: Implemented connection retries in the AsyncConnectionPool

ernestojpg · ernestojpg · commit d39fbc99c143 · 2018-02-21T12:12:20.000+01:00
Signed-off-by: Ernesto J. Perez &lt;ernestojpg@gmail.com&gt;
diff --git a/src/main/java/io/vertx/ext/asyncsql/impl/BaseSQLClient.java b/src/main/java/io/vertx/ext/asyncsql/impl/BaseSQLClient.java
@@ -49,15 +49,11 @@ public abstract class BaseSQLClient {
   protected final Logger log = LoggerFactory.getLogger(this.getClass());
   protected final Vertx vertx;
 
-  protected int maxPoolSize;
-  protected int transactionTimeout;
-  protected String registerAddress;
+  protected final JsonObject globalConfig;
 
-  public BaseSQLClient(Vertx vertx, JsonObject config) {
+  public BaseSQLClient(Vertx vertx, JsonObject globalConfig) {
     this.vertx = vertx;
-    this.maxPoolSize = config.getInteger("maxPoolSize", 10);
-    this.transactionTimeout = config.getInteger("transactionTimeout", 500);
-    this.registerAddress = config.getString("address");
+    this.globalConfig = globalConfig;
   }
 
   protected abstract AsyncConnectionPool pool();
@@ -96,7 +92,7 @@ public void close() {
     close(null);
   }
 
-  protected Configuration getConfiguration(
+  protected Configuration getConnectionConfiguration(
     String defaultHost,
     int defaultPort,
     String defaultDatabase,
diff --git a/src/main/java/io/vertx/ext/asyncsql/impl/MYSQLClientImpl.java b/src/main/java/io/vertx/ext/asyncsql/impl/MYSQLClientImpl.java
@@ -34,10 +34,9 @@ public class MYSQLClientImpl extends BaseSQLClient {
 
   private final MysqlAsyncConnectionPool pool;
 
-  public MYSQLClientImpl(Vertx vertx,
-                         JsonObject config) {
-    super(vertx, config);
-    pool = new MysqlAsyncConnectionPool(vertx, maxPoolSize, getConfiguration(
+  public MYSQLClientImpl(Vertx vertx, JsonObject globalConfig) {
+    super(vertx, globalConfig);
+    pool = new MysqlAsyncConnectionPool(vertx, globalConfig, getConnectionConfiguration(
         MySQLClient.DEFAULT_HOST,
         MySQLClient.DEFAULT_PORT,
         MySQLClient.DEFAULT_DATABASE,
@@ -46,7 +45,7 @@ public MYSQLClientImpl(Vertx vertx,
         MySQLClient.DEFAULT_CHARSET,
         MySQLClient.DEFAULT_CONNECT_TIMEOUT,
         MySQLClient.DEFAULT_TEST_TIMEOUT,
-        config));
+        globalConfig));
   }
 
   @Override
diff --git a/src/main/java/io/vertx/ext/asyncsql/impl/PostgreSQLClientImpl.java b/src/main/java/io/vertx/ext/asyncsql/impl/PostgreSQLClientImpl.java
@@ -17,9 +17,6 @@
 package io.vertx.ext.asyncsql.impl;
 
 import com.github.mauricio.async.db.Connection;
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Future;
-import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.asyncsql.PostgreSQLClient;
@@ -37,9 +34,9 @@ public class PostgreSQLClientImpl extends BaseSQLClient {
 
   private final PostgresqlAsyncConnectionPool pool;
 
-  public PostgreSQLClientImpl(Vertx vertx, JsonObject config) {
-    super(vertx, config);
-    pool = new PostgresqlAsyncConnectionPool(vertx, maxPoolSize, getConfiguration(
+  public PostgreSQLClientImpl(Vertx vertx, JsonObject globalConfig) {
+    super(vertx, globalConfig);
+    pool = new PostgresqlAsyncConnectionPool(vertx, globalConfig, getConnectionConfiguration(
         PostgreSQLClient.DEFAULT_HOST,
         PostgreSQLClient.DEFAULT_PORT,
         PostgreSQLClient.DEFAULT_DATABASE,
@@ -48,7 +45,7 @@ public PostgreSQLClientImpl(Vertx vertx, JsonObject config) {
         PostgreSQLClient.DEFAULT_CHARSET,
         PostgreSQLClient.DEFAULT_CONNECT_TIMEOUT,
         PostgreSQLClient.DEFAULT_TEST_TIMEOUT,
-        config));
+        globalConfig));
   }
 
   @Override
diff --git a/src/main/java/io/vertx/ext/asyncsql/impl/pool/AsyncConnectionPool.java b/src/main/java/io/vertx/ext/asyncsql/impl/pool/AsyncConnectionPool.java
@@ -22,6 +22,7 @@
 import io.vertx.core.Future;
 import io.vertx.core.Handler;
 import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
 import io.vertx.ext.asyncsql.impl.ScalaUtils;
@@ -37,40 +38,64 @@
  */
 public abstract class AsyncConnectionPool {
 
-  private final int maxPoolSize;
+  public static final int DEFAULT_MAX_POOL_SIZE = 10;
+  public static final int DEFAULT_MAX_CONNECTION_RETRIES = 0;       // No connection retries by default
+  public static final int DEFAULT_CONNECTION_RETRY_DELAY = 5_000;   // 5 seconds between retries by default
 
   private static final Logger logger = LoggerFactory.getLogger(AsyncConnectionPool.class);
-  protected final Configuration configuration;
+
+  private final int maxPoolSize;
+  private final int maxConnectionRetries;
+  private final int connectionRetryDelay;
+
+  protected final Configuration connectionConfig;
   protected final Vertx vertx;
 
   private int poolSize = 0;
   private final Deque<Connection> availableConnections = new ArrayDeque<>();
   private final Deque<Handler<AsyncResult<Connection>>> waiters = new ArrayDeque<>();
 
-  public AsyncConnectionPool(Vertx vertx, int maxPoolSize, Configuration configuration) {
+  public AsyncConnectionPool(Vertx vertx, JsonObject globalConfig, Configuration connectionConfig) {
     this.vertx = vertx;
-    this.maxPoolSize = maxPoolSize;
-    this.configuration = configuration;
+    this.maxPoolSize = globalConfig.getInteger("maxPoolSize", DEFAULT_MAX_POOL_SIZE);
+    this.maxConnectionRetries = globalConfig.getInteger("maxConnectionRetries", DEFAULT_MAX_CONNECTION_RETRIES);
+    this.connectionRetryDelay = globalConfig.getInteger("connectionRetryDelay", DEFAULT_CONNECTION_RETRY_DELAY);
+    this.connectionConfig = connectionConfig;
   }
 
   protected abstract Connection create();
 
   private synchronized void createConnection(Handler<AsyncResult<Connection>> handler) {
     poolSize += 1;
+    createAndConnect(new Handler<AsyncResult<Connection>>() {
+      int retries = 0;
+      @Override
+      public void handle(AsyncResult<Connection> connectionResult) {
+        if (connectionResult.succeeded()) {
+          handler.handle(connectionResult);
+        } else if (maxConnectionRetries<0 || retries<maxConnectionRetries) {
+          retries++;
+          logger.debug("Error creating connection. Waiting " + connectionRetryDelay + " ms for the retry " +
+            retries + (maxConnectionRetries >= 0 ? " of " + maxConnectionRetries : ""));
+          vertx.setTimer(connectionRetryDelay, timerId ->
+            createAndConnect(this) // Try to connect again using this handler
+          );
+        } else {
+          poolSize -= 1;
+          notifyWaitersAboutAvailableConnection();
+          handler.handle(connectionResult);
+        }
+      }
+    });
+  }
+
+  private synchronized void createAndConnect(Handler<AsyncResult<Connection>> handler) {
     try {
-      Connection connection = create();
-      connection
-          .connect()
-          .onComplete(ScalaUtils.toFunction1(asyncResult -> {
-            if (asyncResult.failed()) {
-              poolSize -= 1;
-              notifyWaitersAboutAvailableConnection();
-            }
-            handler.handle(asyncResult);
-          }), VertxEventLoopExecutionContext.create(vertx));
+      create()
+        .connect()
+        .onComplete(ScalaUtils.toFunction1(handler), VertxEventLoopExecutionContext.create(vertx));
     } catch (Throwable e) {
       logger.info("creating a connection went wrong", e);
-      poolSize -= 1;
       handler.handle(Future.failedFuture(e));
     }
   }
diff --git a/src/main/java/io/vertx/ext/asyncsql/impl/pool/MysqlAsyncConnectionPool.java b/src/main/java/io/vertx/ext/asyncsql/impl/pool/MysqlAsyncConnectionPool.java
@@ -21,6 +21,7 @@
 import com.github.mauricio.async.db.mysql.MySQLConnection;
 import com.github.mauricio.async.db.mysql.util.CharsetMapper;
 import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
 import io.vertx.ext.asyncsql.impl.VertxEventLoopExecutionContext;
 
 /**
@@ -30,13 +31,13 @@
  */
 public class MysqlAsyncConnectionPool extends AsyncConnectionPool {
 
-  public MysqlAsyncConnectionPool(Vertx vertx, int maxPoolSize, Configuration configuration) {
-    super(vertx, maxPoolSize, configuration);
+  public MysqlAsyncConnectionPool(Vertx vertx, JsonObject globalConfig, Configuration connectionConfig) {
+    super(vertx, globalConfig, connectionConfig);
   }
 
   @Override
   protected Connection create() {
-    return new MySQLConnection(configuration, CharsetMapper.Instance(),
+    return new MySQLConnection(connectionConfig, CharsetMapper.Instance(),
         vertx.nettyEventLoopGroup().next(),
         VertxEventLoopExecutionContext.create(vertx)
     );
diff --git a/src/main/java/io/vertx/ext/asyncsql/impl/pool/PostgresqlAsyncConnectionPool.java b/src/main/java/io/vertx/ext/asyncsql/impl/pool/PostgresqlAsyncConnectionPool.java
@@ -22,6 +22,7 @@
 import com.github.mauricio.async.db.postgresql.column.PostgreSQLColumnDecoderRegistry;
 import com.github.mauricio.async.db.postgresql.column.PostgreSQLColumnEncoderRegistry;
 import io.vertx.core.Vertx;
+import io.vertx.core.json.JsonObject;
 import io.vertx.ext.asyncsql.impl.VertxEventLoopExecutionContext;
 
 /**
@@ -31,14 +32,14 @@
  */
 public class PostgresqlAsyncConnectionPool extends AsyncConnectionPool {
 
-  public PostgresqlAsyncConnectionPool(Vertx vertx, int maxPoolSize, Configuration configuration) {
-    super(vertx, maxPoolSize, configuration);
+  public PostgresqlAsyncConnectionPool(Vertx vertx, JsonObject globalConfig, Configuration connectionConfig) {
+    super(vertx, globalConfig, connectionConfig);
   }
 
   @Override
   protected Connection create() {
     return new PostgreSQLConnection(
-        configuration,
+        connectionConfig,
         PostgreSQLColumnEncoderRegistry.Instance(),
         PostgreSQLColumnDecoderRegistry.Instance(),
         vertx.nettyEventLoopGroup().next(),
