Suspend SqlConnectionPool during CRaC snapshotting

pom.xml

@@ -60,6 +60,7 @@
     <!-- We skip sources jar generation as we do it with the assembly plugin to have greater control over the content -->
     <source.skip>true</source.skip>
     <testcontainers.version>1.17.6</testcontainers.version>
+    <org.crac.version>0.1.3</org.crac.version>
   </properties>
 
   <dependencyManagement>

vertx-sql-client/pom.xml

@@ -56,6 +56,11 @@
     </dependency>
 
     <!-- Others -->
+    <dependency>
+      <groupId>org.crac</groupId>
+      <artifactId>crac</artifactId>
+      <version>${org.crac.version}</version>
+    </dependency>
     <dependency>
       <groupId>io.vertx</groupId>
       <artifactId>vertx-core</artifactId>

vertx-sql-client/src/main/java/io/vertx/sqlclient/impl/pool/SqlConnectionPool.java

@@ -32,18 +32,22 @@
 import io.vertx.sqlclient.spi.DatabaseMetadata;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
+import org.crac.Core;
+import org.crac.Resource;
+
 /**
  * Todo :
  *
  * - for per statement pooling, have several physical connection and use the less busy one to avoid head of line blocking effect
  *
  * @author <a href="mailto:[email protected]">Julien Viet</a>
  */
-public class SqlConnectionPool {
+public class SqlConnectionPool implements Resource {
 
   private final Function<Context, Future<SqlConnection>> connectionProvider;
   private final VertxInternal vertx;
@@ -101,6 +105,7 @@ public EventLoopContext apply(ContextInternal contextInternal) {
     } else {
       pool.contextProvider(ctx -> ctx.owner().createEventLoopContext(ctx.nettyEventLoop(), null, Thread.currentThread().getContextClassLoader()));
     }
+    Core.getGlobalContext().register(this);
   }
 
   private final PoolConnector<PooledConnection> connector = new PoolConnector<PooledConnection>() {
@@ -258,6 +263,27 @@ public Future<Void> close() {
     return promise.future();
   }
 
+  @Override
+  public void beforeCheckpoint(org.crac.Context<? extends Resource> ctx) throws Exception {
+    CompletableFuture<Void> cf = new CompletableFuture<>();
+    pool.suspend(result -> {
+      if (result.failed()) {
+        cf.completeExceptionally(result.cause());
+      }
+      CompositeFuture.join(result.result().stream().map(
+        f -> f.compose(pooled -> Future.<Void>future(p -> pooled.conn.close(pooled, p)))
+      ).collect(Collectors.toList())).mapEmpty()
+        .onSuccess(ignored -> cf.complete(null))
+        .onFailure(cf::completeExceptionally);
+    });
+    cf.get();
+  }
+
+  @Override
+  public void afterRestore(org.crac.Context<? extends Resource> ctx) {
+    pool.resume();
+  }
+
   public class PooledConnection implements Connection, Connection.Holder {
 
     private final ConnectionFactory factory;