proposed extension point for connection pools

hibernate-reactive-core/src/main/java/org/hibernate/reactive/boot/impl/ReactiveServiceRegistryInitializer.java

@@ -2,7 +2,7 @@
 
 import org.hibernate.boot.registry.StandardServiceRegistryBuilder;
 import org.hibernate.reactive.jpa.impl.ReactivePersisterClassResolverInitiator;
-import org.hibernate.reactive.service.initiator.ReactiveConnectionPoolInitiator;
+import org.hibernate.reactive.pool.impl.SqlClientPoolInitiator;
 import org.hibernate.reactive.service.initiator.DummyConnectionProviderInitiator;
 import org.hibernate.reactive.service.initiator.ReactiveJdbcEnvironmentInitiator;
 import org.hibernate.reactive.service.initiator.ReactiveQueryTranslatorFactoryInitiator;
@@ -20,9 +20,9 @@ public class ReactiveServiceRegistryInitializer implements ServiceContributor {
 	public void contribute(StandardServiceRegistryBuilder serviceRegistryBuilder) {
 		serviceRegistryBuilder.addInitiator( DummyConnectionProviderInitiator.INSTANCE );
 		serviceRegistryBuilder.addInitiator( ReactiveJdbcEnvironmentInitiator.INSTANCE );
-		serviceRegistryBuilder.addInitiator( ReactiveConnectionPoolInitiator.INSTANCE );
 		serviceRegistryBuilder.addInitiator( ReactiveTransactionCoordinatorBuilderInitiator.INSTANCE );
 		serviceRegistryBuilder.addInitiator( ReactivePersisterClassResolverInitiator.INSTANCE );
 		serviceRegistryBuilder.addInitiator( ReactiveQueryTranslatorFactoryInitiator.INSTANCE );
+		serviceRegistryBuilder.addInitiator( SqlClientPoolInitiator.INSTANCE );
 	}
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/impl/ReactiveSessionInternal.java

@@ -10,7 +10,7 @@
 import org.hibernate.event.internal.MergeContext;
 import org.hibernate.internal.util.collections.IdentitySet;
 import org.hibernate.reactive.engine.spi.ReactiveActionQueue;
-import org.hibernate.reactive.service.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.stage.Stage;
 
 /**

hibernate-reactive-core/src/main/java/org/hibernate/reactive/impl/ReactiveSessionInternalImpl.java

@@ -67,7 +67,7 @@
 import org.hibernate.reactive.mutiny.Mutiny;
 import org.hibernate.reactive.mutiny.impl.MutinySessionImpl;
 import org.hibernate.reactive.persister.entity.impl.ReactiveEntityPersister;
-import org.hibernate.reactive.service.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.stage.Stage;
 import org.hibernate.reactive.stage.impl.StageSessionImpl;
 import org.hibernate.reactive.util.impl.CompletionStages;

hibernate-reactive-core/src/main/java/org/hibernate/reactive/mutiny/impl/MutinySessionFactoryImpl.java

@@ -5,7 +5,7 @@
 import org.hibernate.internal.SessionFactoryImpl;
 import org.hibernate.reactive.impl.ReactiveSessionInternalImpl;
 import org.hibernate.reactive.mutiny.Mutiny;
-import org.hibernate.reactive.service.ReactiveConnectionPool;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
 
 import java.util.function.Function;
 

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/ReactiveAbstractEntityPersister.java

@@ -19,7 +19,7 @@
 import org.hibernate.reactive.adaptor.impl.PreparedStatementAdaptor;
 import org.hibernate.reactive.impl.ReactiveSessionInternal;
 import org.hibernate.reactive.loader.entity.impl.ReactiveDynamicBatchingEntityLoaderBuilder;
-import org.hibernate.reactive.service.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.sql.impl.Parameters;
 import org.hibernate.reactive.util.impl.CompletionStages;
 import org.hibernate.sql.Delete;

hibernate-reactive-core/src/main/java/org/hibernate/reactive/persister/entity/impl/TableReactiveIdentifierGenerator.java

@@ -21,7 +21,7 @@
 import org.hibernate.mapping.Table;
 import org.hibernate.persister.spi.PersisterCreationContext;
 import org.hibernate.reactive.impl.ReactiveSessionInternal;
-import org.hibernate.reactive.service.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.sql.impl.Parameters;
 
 import java.util.Collections;

hibernate-reactive-core/src/main/java/org/hibernate/reactive/service/ReactiveConnection.java → hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/ReactiveConnection.java

@@ -1,13 +1,15 @@
-package org.hibernate.reactive.service;
+package org.hibernate.reactive.pool;
 
 import java.sql.ResultSet;
 import java.util.Iterator;
 import java.util.concurrent.CompletionStage;
 
 /**
- * Abstracts over reactive connection pools.
+ * Abstracts over reactive database connections, defining
+ * operations that allow queries to be executed asynchronously
+ * via {@link CompletionStage}.
  *
- * @see SqlClientConnection for the Vert.x-based implementation
+ * @see ReactiveConnectionPool
  */
 public interface ReactiveConnection {
 

hibernate-reactive-core/src/main/java/org/hibernate/reactive/service/ReactiveConnectionPool.java → hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/ReactiveConnectionPool.java

@@ -1,4 +1,4 @@
-package org.hibernate.reactive.service;
+package org.hibernate.reactive.pool;
 
 import org.hibernate.service.Service;
 
@@ -8,7 +8,11 @@
  * A Hibernate {@link Service} that provides access to pooled
  * {@link ReactiveConnection reactive connections}.
  *
- * @see SqlClientPool
+ * A program may integrate a custom {@link ReactiveConnectionPool}
+ * with Hibernate Reactive by contributing a new service using a
+ * {@link org.hibernate.boot.registry.StandardServiceInitiator}
+ * or from code-based Hibernate configuration by calling
+ * {@link org.hibernate.boot.registry.StandardServiceRegistryBuilder#addService}.
  */
 public interface ReactiveConnectionPool extends Service {
 

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/Handlers.java

@@ -0,0 +1,33 @@
+package org.hibernate.reactive.pool.impl;
+
+import io.vertx.core.AsyncResult;
+import io.vertx.core.Handler;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Consumer;
+
+/**
+ * Utility operations for working with Vert.x handlers
+ */
+public class Handlers {
+
+	protected static <T> CompletionStage<T> toCompletionStage(
+			Consumer<Handler<AsyncResult<T>>> completionConsumer) {
+		CompletableFuture<T> cs = new CompletableFuture<>();
+//		try {
+			completionConsumer.accept( ar -> {
+				if ( ar.succeeded() ) {
+					cs.complete( ar.result() );
+				}
+				else {
+					cs.completeExceptionally( ar.cause() );
+				}
+			} );
+//		}
+//		catch (Exception e) {
+//			cs.completeExceptionally( e );
+//		}
+		return cs;
+	}
+}

hibernate-reactive-core/src/main/java/org/hibernate/reactive/service/SqlClientConnection.java → hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/SqlClientConnection.java

@@ -1,16 +1,13 @@
-package org.hibernate.reactive.service;
+package org.hibernate.reactive.pool.impl;
 
-import io.vertx.core.AsyncResult;
-import io.vertx.core.Handler;
 import io.vertx.sqlclient.*;
 import org.hibernate.reactive.adaptor.impl.ResultSetAdaptor;
+import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.util.impl.CompletionStages;
 
 import java.sql.ResultSet;
 import java.util.Objects;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
-import java.util.function.Consumer;
 
 import static io.vertx.core.Future.failedFuture;
 import static io.vertx.core.Future.succeededFuture;
@@ -27,23 +24,11 @@ public class SqlClientConnection implements ReactiveConnection {
 	private final SqlConnection connection;
 	private Transaction transaction;
 
-	private SqlClientConnection(SqlConnection connection, boolean showSQL) {
+	SqlClientConnection(SqlConnection connection, boolean showSQL) {
 		this.showSQL = showSQL;
 		this.connection = connection;
 	}
 
-	public static CompletionStage<ReactiveConnection> create(Pool pool, boolean showSQL) {
-		return toCompletionStage(
-				handler -> pool.getConnection(
-						ar -> handler.handle(
-								ar.succeeded()
-										? succeededFuture( new SqlClientConnection( ar.result(), showSQL ) )
-										: failedFuture( ar.cause() )
-						)
-				)
-		);
-	}
-
 	@Override
 	public CompletionStage<Integer> update(String sql, Object[] paramValues) {
 		return update( sql, Tuple.wrap( paramValues ) );
@@ -109,7 +94,7 @@ public CompletionStage<RowSet<Row>> preparedQuery(String sql, Tuple parameters)
 		if (showSQL) {
 			System.out.println(sql);
 		}
-		return toCompletionStage(
+		return Handlers.toCompletionStage(
 				handler -> client().preparedQuery( sql ).execute( parameters, handler )
 		);
 	}
@@ -119,7 +104,7 @@ public CompletionStage<RowSet<Row>> preparedQuery(String sql) {
 		if (showSQL) {
 			System.out.println(sql);
 		}
-		return toCompletionStage(
+		return Handlers.toCompletionStage(
 				handler -> client().preparedQuery( sql ).execute( handler )
 		);
 	}
@@ -138,7 +123,7 @@ public CompletionStage<Void> beginTransaction() {
 
 	@Override
 	public CompletionStage<Void> commitTransaction() {
-		return toCompletionStage(
+		return Handlers.toCompletionStage(
 				handler -> transaction.commit(
 						ar -> {
 							transaction = null;
@@ -151,7 +136,7 @@ public CompletionStage<Void> commitTransaction() {
 
 	@Override
 	public CompletionStage<Void> rollbackTransaction() {
-		return toCompletionStage(
+		return Handlers.toCompletionStage(
 				handler -> transaction.rollback(
 						ar -> {
 							transaction = null;
@@ -167,25 +152,6 @@ public void close() {
 		connection.close();
 	}
 
-	protected static <T> CompletionStage<T> toCompletionStage(
-			Consumer<Handler<AsyncResult<T>>> completionConsumer) {
-		CompletableFuture<T> cs = new CompletableFuture<>();
-//		try {
-			completionConsumer.accept( ar -> {
-				if ( ar.succeeded() ) {
-					cs.complete( ar.result() );
-				}
-				else {
-					cs.completeExceptionally( ar.cause() );
-				}
-			} );
-//		}
-//		catch (Exception e) {
-//			cs.completeExceptionally( e );
-//		}
-		return cs;
-	}
-
 	/**
 	 * Loads MySQLClient.LAST_INSERTED_ID via reflection to avoid a hard
 	 * dependency on the MySQL driver 

hibernate-reactive-core/src/main/java/org/hibernate/reactive/service/SqlClientPool.java → hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/SqlClientPool.java

@@ -1,4 +1,4 @@
-package org.hibernate.reactive.service;
+package org.hibernate.reactive.pool.impl;
 
 import java.net.URI;
 import java.util.Map;
@@ -10,6 +10,8 @@
 import org.hibernate.internal.util.config.ConfigurationException;
 import org.hibernate.internal.util.config.ConfigurationHelper;
 import org.hibernate.reactive.cfg.ReactiveSettings;
+import org.hibernate.reactive.pool.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
 import org.hibernate.reactive.util.impl.JdbcUrlParser;
 import org.hibernate.service.spi.Configurable;
 import org.hibernate.service.spi.Stoppable;
@@ -20,6 +22,9 @@
 import io.vertx.sqlclient.SqlConnectOptions;
 import io.vertx.sqlclient.spi.Driver;
 
+import static io.vertx.core.Future.failedFuture;
+import static io.vertx.core.Future.succeededFuture;
+
 /**
  * A pool of reactive connections backed by a Vert.x {@link Pool}.
  */
@@ -131,7 +136,15 @@ private Pool configurePool(Map configurationValues) {
 
 	@Override
 	public CompletionStage<ReactiveConnection> getConnection() {
-		return SqlClientConnection.create( pool, showSQL );
+		return Handlers.toCompletionStage(
+				handler -> pool.getConnection(
+						ar -> handler.handle(
+								ar.succeeded()
+										? succeededFuture( new SqlClientConnection( ar.result(), showSQL) )
+										: failedFuture( ar.cause() )
+						)
+				)
+		);
 	}
 
 	@Override

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/SqlClientPoolInitiator.java

@@ -0,0 +1,32 @@
+package org.hibernate.reactive.pool.impl;
+
+import org.hibernate.boot.registry.StandardServiceInitiator;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
+import org.hibernate.service.spi.ServiceRegistryImplementor;
+
+import java.util.Map;
+
+/**
+ * A Hibernate {@link StandardServiceInitiator service initiator} that
+ * integrates our {@link ReactiveConnectionPool}.
+ *
+ * @see ReactiveConnectionPool
+ * @see SqlClientPool
+ */
+public class SqlClientPoolInitiator implements StandardServiceInitiator<ReactiveConnectionPool> {
+
+	public static final SqlClientPoolInitiator INSTANCE = new SqlClientPoolInitiator();
+
+	private SqlClientPoolInitiator() {
+	}
+
+	@Override
+	public ReactiveConnectionPool initiateService(Map configurationValues, ServiceRegistryImplementor registry) {
+		return new SqlClientPool( configurationValues );
+	}
+
+	@Override
+	public Class<ReactiveConnectionPool> getServiceInitiated() {
+		return ReactiveConnectionPool.class;
+	}
+}

hibernate-reactive-core/src/main/java/org/hibernate/reactive/service/ReactiveGenerationTarget.java

@@ -1,5 +1,7 @@
 package org.hibernate.reactive.service;
 
+import org.hibernate.reactive.pool.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
 import org.hibernate.service.ServiceRegistry;
 import org.hibernate.tool.schema.internal.exec.GenerationTarget;
 

hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/impl/StageSessionFactoryImpl.java

@@ -3,7 +3,7 @@
 import org.hibernate.HibernateException;
 import org.hibernate.internal.SessionFactoryImpl;
 import org.hibernate.reactive.impl.ReactiveSessionInternalImpl;
-import org.hibernate.reactive.service.ReactiveConnectionPool;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
 import org.hibernate.reactive.stage.Stage;
 
 import java.util.concurrent.CompletionStage;

hibernate-reactive-core/src/test/java/org/hibernate/reactive/BaseMutinyTest.java

@@ -11,8 +11,8 @@
 import org.hibernate.cfg.Configuration;
 import org.hibernate.reactive.containers.PostgreSQLDatabase;
 import org.hibernate.reactive.mutiny.Mutiny;
-import org.hibernate.reactive.service.ReactiveConnection;
-import org.hibernate.reactive.service.ReactiveConnectionPool;
+import org.hibernate.reactive.pool.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
 import org.hibernate.reactive.stage.Stage;
 import org.junit.After;
 import org.junit.Before;

hibernate-reactive-core/src/test/java/org/hibernate/reactive/BaseReactiveTest.java

@@ -7,8 +7,8 @@
 import org.hibernate.cfg.AvailableSettings;
 import org.hibernate.cfg.Configuration;
 import org.hibernate.reactive.containers.PostgreSQLDatabase;
-import org.hibernate.reactive.service.ReactiveConnection;
-import org.hibernate.reactive.service.ReactiveConnectionPool;
+import org.hibernate.reactive.pool.ReactiveConnection;
+import org.hibernate.reactive.pool.ReactiveConnectionPool;
 import org.hibernate.reactive.stage.Stage;
 import org.junit.After;
 import org.junit.Before;