Provide a way for users to customize PgPool creation

Resolves: quarkusio#29348

extensions/reactive-pg-client/deployment/src/main/java/io/quarkus/reactive/pg/client/deployment/ReactivePgClientProcessor.java

@@ -2,11 +2,18 @@
 
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
+import java.util.function.Predicate;
 
 import javax.enterprise.context.ApplicationScoped;
 
+import org.jboss.jandex.DotName;
+import org.jboss.jandex.Type;
+
 import io.quarkus.arc.deployment.SyntheticBeanBuildItem;
 import io.quarkus.arc.deployment.SyntheticBeanBuildItem.ExtendedBeanConfigurator;
+import io.quarkus.arc.deployment.UnremovableBeanBuildItem;
+import io.quarkus.arc.deployment.ValidationPhaseBuildItem;
 import io.quarkus.arc.processor.DotNames;
 import io.quarkus.datasource.common.runtime.DataSourceUtil;
 import io.quarkus.datasource.common.runtime.DatabaseKind;
@@ -34,6 +41,7 @@
 import io.quarkus.reactive.datasource.runtime.DataSourceReactiveBuildTimeConfig;
 import io.quarkus.reactive.datasource.runtime.DataSourcesReactiveBuildTimeConfig;
 import io.quarkus.reactive.datasource.runtime.DataSourcesReactiveRuntimeConfig;
+import io.quarkus.reactive.pg.client.PgPoolCreator;
 import io.quarkus.reactive.pg.client.runtime.DataSourcesReactivePostgreSQLConfig;
 import io.quarkus.reactive.pg.client.runtime.PgPoolRecorder;
 import io.quarkus.reactive.pg.client.runtime.PostgreSQLServiceBindingConverter;
@@ -121,6 +129,23 @@ void addHealthCheck(
                         dataSourcesBuildTimeConfig.healthEnabled));
     }
 
+    @BuildStep
+    void unremoveableBeans(BuildProducer<UnremovableBeanBuildItem> producer) {
+        producer.produce(UnremovableBeanBuildItem.beanTypes(PgPoolCreator.class));
+    }
+
+    @BuildStep
+    void validateBeans(ValidationPhaseBuildItem validationPhase,
+            BuildProducer<ValidationPhaseBuildItem.ValidationErrorBuildItem> errors) {
+        long pgPoolCreatorCount = validationPhase.getContext()
+                .beans().matchBeanTypes(new PgPoolCreatorBeanClassPredicate()).stream().count();
+        if (pgPoolCreatorCount > 1) {
+            errors.produce(new ValidationPhaseBuildItem.ValidationErrorBuildItem(
+                    new IllegalStateException(
+                            "There can be at most one bean of type '" + PgPoolCreator.class.getName() + "'")));
+        }
+    }
+
     @BuildStep
     void registerServiceBinding(Capabilities capabilities, BuildProducer<ServiceProviderBuildItem> serviceProvider,
             BuildProducer<DefaultDataSourceDbKindBuildItem> dbKind) {
@@ -243,4 +268,14 @@ private static void addQualifiers(ExtendedBeanConfigurator configurator, String
                     .done();
         }
     }
+
+    private static class PgPoolCreatorBeanClassPredicate implements Predicate<Set<Type>> {
+        private static final Type PG_POOL_CREATOR = Type.create(DotName.createSimple(PgPoolCreator.class.getName()),
+                Type.Kind.CLASS);
+
+        @Override
+        public boolean test(Set<Type> types) {
+            return types.contains(PG_POOL_CREATOR);
+        }
+    }
 }

extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/LocalhostPgPoolCreator.java

@@ -0,0 +1,14 @@
+package io.quarkus.reactive.pg.client;
+
+import javax.inject.Singleton;
+
+import io.vertx.pgclient.PgPool;
+
+@Singleton
+public class LocalhostPgPoolCreator implements PgPoolCreator {
+
+    @Override
+    public PgPool create(Input input) {
+        return PgPool.pool(input.vertx(), input.pgConnectOptions().setHost("localhost").setPort(5431), input.poolOptions());
+    }
+}

extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultiplePgPoolCreatorsTest.java

@@ -0,0 +1,40 @@
+package io.quarkus.reactive.pg.client;
+
+import static org.junit.jupiter.api.Assertions.fail;
+
+import javax.enterprise.inject.spi.DeploymentException;
+import javax.inject.Singleton;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+import io.vertx.pgclient.PgPool;
+
+public class MultiplePgPoolCreatorsTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClass(CustomCredentialsProvider.class)
+                    .addClass(CredentialsTestResource.class)
+                    .addClass(LocalhostPgPoolCreator.class)
+                    .addClass(AnotherPgPoolCreator.class)
+                    .addAsResource("application-credentials-with-erroneous-url.properties", "application.properties"))
+            .setExpectedException(DeploymentException.class);
+
+    @Test
+    public void test() {
+        fail("Should never have been called");
+    }
+
+    @Singleton
+    public static class AnotherPgPoolCreator implements PgPoolCreator {
+
+        @Override
+        public PgPool create(Input input) {
+            return PgPool.pool(input.vertx(), input.pgConnectOptions(), input.poolOptions());
+        }
+    }
+
+}

extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/PgPoolCreatorTest.java

@@ -0,0 +1,30 @@
+package io.quarkus.reactive.pg.client;
+
+import static io.restassured.RestAssured.given;
+
+import org.hamcrest.CoreMatchers;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusUnitTest;
+
+public class PgPoolCreatorTest {
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar
+                    .addClass(CustomCredentialsProvider.class)
+                    .addClass(CredentialsTestResource.class)
+                    .addClass(LocalhostPgPoolCreator.class)
+                    .addAsResource("application-credentials-with-erroneous-url.properties", "application.properties"));
+
+    @Test
+    public void testConnect() {
+        given()
+                .when().get("/test")
+                .then()
+                .statusCode(200)
+                .body(CoreMatchers.equalTo("OK"));
+    }
+
+}

extensions/reactive-pg-client/deployment/src/test/resources/application-credentials-with-erroneous-url.properties

@@ -0,0 +1,3 @@
+quarkus.datasource.db-kind=postgresql
+quarkus.datasource.credentials-provider=custom
+quarkus.datasource.reactive.url=vertx-reactive:postgresql://test:12345/hibernate_orm_test

extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/PgPoolCreator.java

@@ -0,0 +1,34 @@
+package io.quarkus.reactive.pg.client;
+
+import io.vertx.core.Vertx;
+import io.vertx.pgclient.PgConnectOptions;
+import io.vertx.pgclient.PgPool;
+import io.vertx.sqlclient.PoolOptions;
+
+/**
+ * This interface is an integration point that allows users to use {@link Vertx}, {@link PoolOptions} and
+ * {@link PgConnectOptions}
+ * configured automatically by Quarkus, but provide on a custom strategy for creating the final {@link PgPool}, instead of
+ * letting Quarkus create it (by calling {@link PgPool#pool(Vertx, PgConnectOptions, PoolOptions)}).
+ *
+ * Implementations of this class are meant to be used as CDI beans - if a single implementation marked as a CDI bean,
+ * it will be picked up by Quarkus and used to create the pool.
+ */
+public interface PgPoolCreator {
+
+    PgPool create(Input input);
+
+    interface Input {
+
+        Vertx vertx();
+
+        PoolOptions poolOptions();
+
+        PgConnectOptions pgConnectOptions();
+
+        /**
+         * The datasource name for which the pool is being created
+         */
+        String dataSourceName();
+    }
+}

extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/runtime/PgPoolRecorder.java

@@ -13,14 +13,18 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 
+import javax.enterprise.inject.Instance;
+
 import org.jboss.logging.Logger;
 
+import io.quarkus.arc.Arc;
 import io.quarkus.credentials.CredentialsProvider;
 import io.quarkus.credentials.runtime.CredentialsProviderFinder;
 import io.quarkus.datasource.runtime.DataSourceRuntimeConfig;
 import io.quarkus.datasource.runtime.DataSourcesRuntimeConfig;
 import io.quarkus.reactive.datasource.runtime.DataSourceReactiveRuntimeConfig;
 import io.quarkus.reactive.datasource.runtime.DataSourcesReactiveRuntimeConfig;
+import io.quarkus.reactive.pg.client.PgPoolCreator;
 import io.quarkus.runtime.RuntimeValue;
 import io.quarkus.runtime.ShutdownContext;
 import io.quarkus.runtime.annotations.Recorder;
@@ -46,6 +50,7 @@ public RuntimeValue<PgPool> configurePgPool(RuntimeValue<Vertx> vertx,
 
         PgPool pgPool = initialize(vertx.getValue(),
                 eventLoopCount.get(),
+                dataSourceName,
                 dataSourcesRuntimeConfig.getDataSourceRuntimeConfig(dataSourceName),
                 dataSourcesReactiveRuntimeConfig.getDataSourceReactiveRuntimeConfig(dataSourceName),
                 dataSourcesReactivePostgreSQLConfig.getDataSourceReactiveRuntimeConfig(dataSourceName));
@@ -60,6 +65,7 @@ public RuntimeValue<io.vertx.mutiny.pgclient.PgPool> mutinyPgPool(RuntimeValue<P
 
     private PgPool initialize(Vertx vertx,
             Integer eventLoopCount,
+            String dataSourceName,
             DataSourceRuntimeConfig dataSourceRuntimeConfig,
             DataSourceReactiveRuntimeConfig dataSourceReactiveRuntimeConfig,
             DataSourceReactivePostgreSQLConfig dataSourceReactivePostgreSQLConfig) {
@@ -71,7 +77,7 @@ private PgPool initialize(Vertx vertx,
             log.warn(
                     "Configuration element 'thread-local' on Reactive datasource connections is deprecated and will be ignored. The started pool will always be based on a per-thread separate pool now.");
         }
-        return PgPool.pool(vertx, pgConnectOptions, poolOptions);
+        return createPool(vertx, poolOptions, pgConnectOptions, dataSourceName);
     }
 
     private PoolOptions toPoolOptions(Integer eventLoopCount,
@@ -194,4 +200,49 @@ private PgConnectOptions toPgConnectOptions(DataSourceRuntimeConfig dataSourceRu
 
         return pgConnectOptions;
     }
+
+    private PgPool createPool(Vertx vertx, PoolOptions poolOptions, PgConnectOptions pgConnectOptions,
+            String dataSourceName) {
+        Instance<PgPoolCreator> instance = Arc.container().select(PgPoolCreator.class);
+        if (instance.isResolvable()) {
+            PgPoolCreator.Input input = new DefaultInput(vertx, poolOptions, pgConnectOptions, dataSourceName);
+            return instance.get().create(input);
+        }
+        return PgPool.pool(vertx, pgConnectOptions, poolOptions);
+    }
+
+    private static class DefaultInput implements PgPoolCreator.Input {
+        private final Vertx vertx;
+        private final PoolOptions poolOptions;
+        private final PgConnectOptions pgConnectOptions;
+        private final String dataSourceName;
+
+        public DefaultInput(Vertx vertx, PoolOptions poolOptions, PgConnectOptions pgConnectOptions,
+                String datasourceName) {
+            this.vertx = vertx;
+            this.poolOptions = poolOptions;
+            this.pgConnectOptions = pgConnectOptions;
+            this.dataSourceName = datasourceName;
+        }
+
+        @Override
+        public Vertx vertx() {
+            return vertx;
+        }
+
+        @Override
+        public PoolOptions poolOptions() {
+            return poolOptions;
+        }
+
+        @Override
+        public PgConnectOptions pgConnectOptions() {
+            return pgConnectOptions;
+        }
+
+        @Override
+        public String dataSourceName() {
+            return dataSourceName;
+        }
+    }
 }