[hibernate#712] Make ReactiveConnection#close reactive

They made it reactive with the upgrade to
Vert.x SQL client 4.

We also have to update the tests becaus now
they have to wait for the session to be close before
moving to the next chained operation.

Author: DavideD

hibernate-reactive-core/src/main/java/org/hibernate/reactive/mutiny/Mutiny.java

@@ -18,7 +18,6 @@
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.proxy.HibernateProxy;
 import org.hibernate.reactive.common.AffectedEntities;
-import org.hibernate.reactive.common.AutoCloseable;
 import org.hibernate.reactive.common.Identifier;
 import org.hibernate.reactive.common.ResultSetMapping;
 import org.hibernate.reactive.session.ReactiveSession;
@@ -304,7 +303,7 @@ interface Query<R> {
 	 *
 	 * @see org.hibernate.Session
 	 */
-	interface Session extends AutoCloseable {
+	interface Session {
 
 		/**
 		 * Asynchronously return the persistent instance of the given entity
@@ -1073,7 +1072,7 @@ <R> Query<R> createNativeQuery(String queryString, ResultSetMapping<R> resultSet
 		 * Close the reactive session and release the underlying database
 		 * connection.
 		 */
-		void close();
+		Uni<Void> close();
 
 		/**
 		 * @return false if {@link #close()} has been called
@@ -1111,7 +1110,7 @@ <R> Query<R> createNativeQuery(String queryString, ResultSetMapping<R> resultSet
 	 *
 	 * @see org.hibernate.StatelessSession
 	 */
-	interface StatelessSession extends AutoCloseable {
+	interface StatelessSession {
 
 		/**
 		 * Retrieve a row.
@@ -1382,15 +1381,13 @@ interface StatelessSession extends AutoCloseable {
 		/**
 		 * @return false if {@link #close()} has been called
 		 */
-		@Override
 		boolean isOpen();
 
 		/**
 		 * Close the reactive session and release the underlying database
 		 * connection.
 		 */
-		@Override
-		void close();
+		Uni<Void> close();
 	}
 
 	/**
@@ -1568,6 +1565,7 @@ interface SessionFactory extends AutoCloseable {
 		/**
 		 * Destroy the session factory and clean up its connection pool.
 		 */
+		@Override
 		void close();
 
 		/**

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

@@ -25,6 +25,7 @@
 import javax.persistence.criteria.CriteriaUpdate;
 import javax.persistence.metamodel.Attribute;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -468,8 +469,12 @@ public boolean isMarkedForRollback() {
 	}
 
 	@Override
-	public void close() {
-		delegate.close();
+	public Uni<Void> close() {
+		return uni( () -> {
+			CompletableFuture<Void> closing = new CompletableFuture<>();
+			delegate.close( closing );
+			return closing;
+		} );
 	}
 
 	@Override

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

@@ -17,6 +17,7 @@
 import javax.persistence.criteria.CriteriaDelete;
 import javax.persistence.criteria.CriteriaQuery;
 import javax.persistence.criteria.CriteriaUpdate;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 import java.util.function.Supplier;
@@ -215,8 +216,12 @@ public boolean isMarkedForRollback() {
     }
 
     @Override
-    public void close() {
-        delegate.close();
+    public Uni<Void> close() {
+        return uni( () -> {
+            CompletableFuture<Void> closing = new CompletableFuture<>();
+            delegate.close( closing );
+            return closing;
+        } );
     }
 
     @Override

hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/BatchingConnection.java

@@ -176,7 +176,7 @@ public CompletionStage<Void> rollbackTransaction() {
         return delegate.rollbackTransaction();
     }
 
-    public void close() {
-        delegate.close();
+    public CompletionStage<Void> close() {
+        return delegate.close();
     }
 }

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

@@ -61,6 +61,5 @@ interface Result extends Iterator<Object[]> {
 
 	CompletionStage<Void> executeBatch();
 
-	void close();
-
+	CompletionStage<Void> close();
 }

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

@@ -13,6 +13,7 @@
 
 import org.hibernate.reactive.pool.ReactiveConnection;
 import org.hibernate.reactive.pool.ReactiveConnectionPool;
+import org.hibernate.reactive.util.impl.CompletionStages;
 
 import static org.hibernate.reactive.common.InternalStateAssertions.assertUseOnEventLoop;
 
@@ -144,12 +145,16 @@ public CompletionStage<Void> executeBatch() {
 	}
 
 	@Override
-	public void close() {
+	public CompletionStage<Void> close() {
+		CompletionStage<Void> stage = CompletionStages.voidFuture();
 		if ( connection != null ) {
-			connection.close();
-			connection = null;
+			stage = stage.thenCompose( v -> connection.close() );
+
 		}
-		closed = true;
+		return stage.thenAccept( v -> {
+			connection = null;
+			closed = true;
+		} );
 	}
 
 }

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

@@ -217,8 +217,8 @@ public CompletionStage<Void> rollbackTransaction() {
 	}
 
 	@Override
-	public void close() {
-		connection.close();
+	public CompletionStage<Void> close() {
+		return connection.close().toCompletionStage();
 	}
 
 	/**

hibernate-reactive-core/src/main/java/org/hibernate/reactive/session/ReactiveSession.java

@@ -24,6 +24,7 @@
 import java.io.Serializable;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 /**
@@ -147,5 +148,7 @@ <T> CompletionStage<T> reactiveFind(Class<T> entityClass, Object id,
 
 	boolean isDirty();
 	boolean isOpen();
-	void close();
+
+	// Different approach so that we can overload the method in SessionImpl
+	void close(CompletableFuture<Void> closing);
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/session/ReactiveStatelessSession.java

@@ -9,6 +9,7 @@
 import org.hibernate.LockMode;
 
 import javax.persistence.EntityGraph;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 /**
@@ -60,6 +61,6 @@ public interface ReactiveStatelessSession extends ReactiveQueryExecutor {
 
     boolean isOpen();
 
-    void close();
+    void close(CompletableFuture<Void> closing);
 
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/session/impl/ReactiveSessionImpl.java

@@ -103,6 +103,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.function.BiFunction;
@@ -1516,10 +1517,24 @@ public ReactiveConnection getReactiveConnection() {
 
 	@Override
 	public void close() throws HibernateException {
+		throw new UnsupportedOperationException( "Non reactive close method called. Use close(CompletableFuture<Void> closing) instead." );
+	}
+
+	@Override
+	public void close(CompletableFuture<Void> closing) {
+		CompletionStage<Void> stage = voidFuture();
 		if ( reactiveConnection != null ) {
-			reactiveConnection.close();
+			stage = stage.thenCompose( v -> reactiveConnection.close() );
 		}
-		super.close();
+		stage.thenAccept( v -> super.close() )
+				.whenComplete( (v, t) -> {
+					if ( t != null ) {
+						closing.completeExceptionally( t );
+					}
+					else {
+						closing.complete( null );
+					}
+				} );
 	}
 
 	@Override @SuppressWarnings("unchecked")

hibernate-reactive-core/src/main/java/org/hibernate/reactive/session/impl/ReactiveStatelessSessionImpl.java

@@ -55,6 +55,7 @@
 import javax.persistence.Tuple;
 import java.io.Serializable;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 
 import static org.hibernate.reactive.id.impl.IdentifierGeneration.assignIdIfNecessary;
@@ -704,7 +705,20 @@ public <T> ReactiveQuery<T> createReactiveCriteriaQuery(String jpaqlString,
 
     @Override
     public void close() {
-        reactiveConnection.close();
-        super.close();
+        throw new UnsupportedOperationException( "Non reactive close method called. Use close(CompletableFuture<Void> closing) instead." );
+    }
+
+    @Override
+    public void close(CompletableFuture<Void> closing) {
+        reactiveConnection.close()
+                .thenAccept( v -> super.close() )
+                .whenComplete( (unused, throwable) -> {
+                    if ( throwable != null ) {
+                        closing.completeExceptionally( throwable );
+                    }
+                    else {
+                        closing.complete( null );
+                    }
+                } );
     }
 }

hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/Stage.java

@@ -17,7 +17,6 @@
 import org.hibernate.engine.spi.SharedSessionContractImplementor;
 import org.hibernate.proxy.HibernateProxy;
 import org.hibernate.reactive.common.AffectedEntities;
-import org.hibernate.reactive.common.AutoCloseable;
 import org.hibernate.reactive.common.Identifier;
 import org.hibernate.reactive.common.ResultSetMapping;
 import org.hibernate.reactive.session.ReactiveSession;
@@ -304,7 +303,7 @@ interface Query<R> {
 	 *
 	 * @see org.hibernate.Session
 	 */
-	interface Session extends AutoCloseable {
+	interface Session {
 
 		/**
 		 * Asynchronously return the persistent instance of the given entity
@@ -1075,7 +1074,7 @@ <R> Query<R> createNativeQuery(String queryString,
 		 * Close the reactive session and release the underlying database
 		 * connection.
 		 */
-		void close();
+		CompletionStage<Void> close();
 
 		/**
 		 * @return false if {@link #close()} has been called
@@ -1113,7 +1112,7 @@ <R> Query<R> createNativeQuery(String queryString,
 	 *
 	 * @see org.hibernate.StatelessSession
 	 */
-	interface StatelessSession extends AutoCloseable {
+	interface StatelessSession {
 
 		/**
 		 * Retrieve a row.
@@ -1384,15 +1383,13 @@ interface StatelessSession extends AutoCloseable {
 		/**
 		 * @return false if {@link #close()} has been called
 		 */
-		@Override
 		boolean isOpen();
 
 		/**
 		 * Close the reactive session and release the underlying database
 		 * connection.
 		 */
-		@Override
-		void close();
+		CompletionStage<Void> close();
 	}
 
 	/**
@@ -1570,6 +1567,7 @@ interface SessionFactory extends AutoCloseable {
 		/**
 		 * Destroy the session factory and clean up its connection pool.
 		 */
+		@Override
 		void close();
 
 		/**

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

@@ -25,6 +25,7 @@
 import javax.persistence.criteria.CriteriaUpdate;
 import javax.persistence.metamodel.Attribute;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 
@@ -477,8 +478,12 @@ public boolean isMarkedForRollback() {
 	}
 
 	@Override
-	public void close() {
-		delegate.close();
+	public CompletionStage<Void> close() {
+		return stage( v -> {
+			CompletableFuture<Void> closing = new CompletableFuture<>();
+			delegate.close( closing );
+			return closing;
+		} );
 	}
 
 	@Override

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

@@ -17,6 +17,7 @@
 import javax.persistence.criteria.CriteriaDelete;
 import javax.persistence.criteria.CriteriaQuery;
 import javax.persistence.criteria.CriteriaUpdate;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Function;
 
@@ -177,9 +178,13 @@ public <T> CompletionStage<T> withTransaction(Function<Stage.Transaction, Comple
     }
 
     @Override
-    public void close() {
-        delegate.close();
-    }
+    public CompletionStage<Void> close() {
+        return stage( v -> {
+            CompletableFuture<Void> closing = new CompletableFuture<>();
+            delegate.close( closing );
+            return closing;
+        } );
+   }
 
     @Override
     public boolean isOpen() {