[hibernate#950] Return Uni or CompletionStage for all openSession methods

Author: Sanne

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

@@ -23,6 +23,7 @@
 import org.hibernate.CacheMode;
 import org.hibernate.Filter;
 import org.hibernate.FlushMode;
+import org.hibernate.HibernateException;
 import org.hibernate.Incubating;
 import org.hibernate.LockMode;
 import org.hibernate.collection.internal.AbstractPersistentCollection;
@@ -1515,62 +1516,15 @@ interface Transaction {
 	 */
 	interface SessionFactory extends AutoCloseable {
 
-		/**
-		 * Obtain a new {@link Session reactive session}, the main
-		 * interaction point between the user's program and Hibernate
-		 * Reactive.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link Session} needs to access the database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link Session#close()}.
-		 *
-		 * @see #withSession(Function)
-		 */
-		Session openSession();
+		Uni<Session> openSession();
 
-		/**
-		 * Obtain a new {@link Session reactive session} for a
-		 * specified tenant.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link Session} needs to access the database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link Session#close()}.
-		 *
-		 * @param tenantId the id of the tenant
-		 *
-		 * @see #withSession(Function)
-		 */
-		Session openSession(String tenantId);
+		Uni<Mutiny.Session> openSession(String tenantId);
 
-		/**
-		 * Obtain a {@link StatelessSession reactive stateless session}.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link StatelessSession} needs to access the
-		 * database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link StatelessSession#close()}.
-		 */
-		StatelessSession openStatelessSession();
+		Session getCurrentContextualSession();
 
-		/**
-		 * Obtain a {@link StatelessSession reactive stateless session}.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link StatelessSession} needs to access the
-		 * database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link StatelessSession#close()}.
-		 *
-		 * @param tenantId the id of the tenant
-		 */
-		StatelessSession openStatelessSession(String tenantId);
+		Uni<StatelessSession> openStatelessSession() throws HibernateException;
+
+		Uni<StatelessSession> openStatelessSession(String tenantId);
 
 		/**
 		 * Perform work using a {@link Session reactive session}.

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

@@ -76,32 +76,24 @@ public Context getContext() {
 		return context;
 	}
 
-	@Override
-	public Mutiny.Session openSession() {
-		SessionCreationOptions options = options();
-		return new MutinySessionImpl(
-				new ReactiveSessionImpl( delegate, options, proxyConnection( options.getTenantIdentifier() ) ),
-				this
-		);
+	public Mutiny.Session getCurrentContextualSession() {
+		Mutiny.Session current = context.get( contextKeyForSession );
+		if ( current != null && current.isOpen() ) {
+			return current;
+		}
+		return null;
 	}
 
 	@Override
-	public Mutiny.Session openSession(String tenantId) {
-		Objects.requireNonNull( tenantId, "parameter 'tenantId' is required" );
-		return new MutinySessionImpl(
-				new ReactiveSessionImpl( delegate, options( tenantId ), proxyConnection( tenantId ) ),
-				this
-		);
-	}
-
-	Uni<Mutiny.Session> newSession() throws HibernateException {
+	public Uni<Mutiny.Session> openSession() throws HibernateException {
 		SessionCreationOptions options = options();
 		return uni( () -> connection( options.getTenantIdentifier() ) )
 				.chain( reactiveConnection -> create( reactiveConnection, () -> new ReactiveSessionImpl( delegate, options, reactiveConnection ) ) )
 				.map( s -> new MutinySessionImpl(s, this) );
 	}
 
-	Uni<Mutiny.Session> newSession(String tenantId) throws HibernateException {
+	@Override
+	public Uni<Mutiny.Session> openSession(String tenantId) throws HibernateException {
 		return uni( () -> connection( tenantId ) )
 				.chain( reactiveConnection -> create( reactiveConnection, () -> new ReactiveSessionImpl( delegate, options( tenantId ), reactiveConnection ) ) )
 				.map( s -> new MutinySessionImpl(s, this) );
@@ -116,29 +108,15 @@ private <S> Uni<S> create(ReactiveConnection connection, Supplier<S> supplier) {
 	}
 
 	@Override
-	public Mutiny.StatelessSession openStatelessSession() {
-		SessionCreationOptions options = options();
-		return new MutinyStatelessSessionImpl(
-				new ReactiveStatelessSessionImpl( delegate, options, proxyConnection( options.getTenantIdentifier() ) ),
-				this
-		);
-	}
-
-	public Mutiny.StatelessSession openStatelessSession(String tenantId) {
-		return new MutinyStatelessSessionImpl(
-				new ReactiveStatelessSessionImpl( delegate, options( tenantId ), proxyConnection( tenantId ) ),
-				this
-		);
-	}
-
-	Uni<Mutiny.StatelessSession> newStatelessSession() throws HibernateException {
+	public Uni<Mutiny.StatelessSession> openStatelessSession() throws HibernateException {
 		SessionCreationOptions options = options();
 		return uni( () -> connection( options.getTenantIdentifier() ) )
 				.chain( reactiveConnection -> create( reactiveConnection, () -> new ReactiveStatelessSessionImpl( delegate, options, reactiveConnection ) ) )
 				.map( s -> new MutinyStatelessSessionImpl(s, this) );
 	}
 
-	Uni<Mutiny.StatelessSession> newStatelessSession(String tenantId) {
+	@Override
+	public Uni<Mutiny.StatelessSession> openStatelessSession(String tenantId) {
 		return uni( () -> connection( tenantId ) )
 				.chain( reactiveConnection -> create( reactiveConnection, () -> new ReactiveStatelessSessionImpl( delegate, options( tenantId ), reactiveConnection ) ) )
 				.map( s -> new MutinyStatelessSessionImpl( s, this ) );
@@ -160,33 +138,26 @@ private CompletionStage<ReactiveConnection> connection(String tenantId) {
 				: connectionPool.getConnection( tenantId );
 	}
 
-	private ReactiveConnection proxyConnection(String tenantId) {
-		assertUseOnEventLoop();
-		return tenantId==null
-				? connectionPool.getProxyConnection()
-				: connectionPool.getProxyConnection( tenantId );
-	}
-
 	@Override
 	public <T> Uni<T> withSession(Function<Mutiny.Session, Uni<T>> work) {
 		Objects.requireNonNull( work, "parameter 'work' is required" );
 		Mutiny.Session current = context.get( contextKeyForSession );
 		if ( current != null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newSession(), work, contextKeyForSession );
+		return withSession( openSession(), work, contextKeyForSession );
 	}
 
 	@Override
 	public <T> Uni<T> withSession(String tenantId, Function<Mutiny.Session, Uni<T>> work) {
 		Objects.requireNonNull( tenantId, "parameter 'tenantId' is required" );
 		Objects.requireNonNull( work, "parameter 'work' is required" );
-		Context.Key<Mutiny.Session> key = new MultitenantKey( contextKeyForSession, tenantId );
+		Context.Key<Mutiny.Session> key = new MultitenantKey<>( contextKeyForSession, tenantId );
 		Mutiny.Session current = context.get(key);
 		if ( current!=null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newSession( tenantId ), work, key );
+		return withSession( openSession( tenantId ), work, key );
 	}
 
 	@Override
@@ -196,7 +167,7 @@ public <T> Uni<T> withStatelessSession(Function<Mutiny.StatelessSession, Uni<T>>
 		if ( current != null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newStatelessSession(), work, contextKeyForStatelessSession );
+		return withSession( openStatelessSession(), work, contextKeyForStatelessSession );
 	}
 
 	@Override
@@ -208,7 +179,7 @@ public <T> Uni<T> withStatelessSession(String tenantId, Function<Mutiny.Stateles
 		if ( current != null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newStatelessSession(tenantId), work, key );
+		return withSession( openStatelessSession( tenantId), work, key );
 	}
 
 	private<S extends Mutiny.Closeable, T> Uni<T> withSession(

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

@@ -1516,62 +1516,13 @@ interface Transaction {
 	 */
 	interface SessionFactory extends AutoCloseable {
 
-		/**
-		 * Obtain a new {@link Session reactive session}, the main
-		 * interaction point between the user's program and Hibernate
-		 * Reactive.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link Session} needs to access the database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link Session#close()}.
-		 *
-		 * @see #withSession(Function)
-		 */
-		Session openSession();
+		CompletionStage<Session> openSession();
 
-		/**
-		 * Obtain a new {@link Session reactive session} for a
-		 * specified tenant.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link Session} needs to access the database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link Session#close()}.
-		 *
-		 * @param tenantId the id of the tenant
-		 *
-		 * @see #withSession(Function)
-		 */
-		Session openSession(String tenantId);
+		CompletionStage<Session> openSession(String tenantId);
 
-		/**
-		 * Obtain a {@link StatelessSession reactive stateless session}.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link StatelessSession} needs to access the
-		 * database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link StatelessSession#close()}.
-		 */
-		StatelessSession openStatelessSession();
+		CompletionStage<StatelessSession> openStatelessSession();
 
-		/**
-		 * Obtain a {@link StatelessSession reactive stateless session}.
-		 * <p>
-		 * The underlying database connection is obtained lazily when
-		 * the returned {@link StatelessSession} needs to access the
-		 * database.
-		 * <p>
-		 * The client must explicitly close the session by calling
-		 * {@link StatelessSession#close()}.
-		 *
-		 * @param tenantId the id of the tenant
-		 */
-		StatelessSession openStatelessSession(String tenantId);
+		CompletionStage<StatelessSession> openStatelessSession(String tenantId);
 
 		/**
 		 * Perform work using a {@link Session reactive session}.

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

@@ -68,59 +68,30 @@ public Context getContext() {
 	}
 
 	@Override
-	public Stage.Session openSession() {
-		SessionCreationOptions options = options();
-		return new StageSessionImpl(
-				new ReactiveSessionImpl( delegate, options, proxyConnection( options.getTenantIdentifier() ) ),
-				this
-		);
-	}
-
-	@Override
-	public Stage.Session openSession(String tenantId) {
-		return new StageSessionImpl(
-				new ReactiveSessionImpl( delegate, options( tenantId ), proxyConnection( tenantId ) ),
-				this
-		);
-	}
-
-	public Stage.StatelessSession openStatelessSession() {
-		SessionCreationOptions options = options();
-		return new StageStatelessSessionImpl(
-				new ReactiveStatelessSessionImpl( delegate, options, proxyConnection( options.getTenantIdentifier() ) ),
-				this
-		);
-	}
-
-	@Override
-	public Stage.StatelessSession openStatelessSession(String tenantId) {
-		return new StageStatelessSessionImpl(
-				new ReactiveStatelessSessionImpl( delegate, options( tenantId ), proxyConnection( tenantId ) ),
-				this
-		);
-	}
-
-	CompletionStage<Stage.Session> newSession() {
+	public CompletionStage<Stage.Session> openSession() {
 		SessionCreationOptions options = options();
 		return stage( v -> connection( options.getTenantIdentifier() )
 				.thenCompose( connection -> create( connection, () -> new ReactiveSessionImpl( delegate, options, connection ) ) )
 				.thenApply( s -> new StageSessionImpl(s, this) ) );
 	}
 
-	CompletionStage<Stage.Session> newSession(String tenantId) {
+	@Override
+	public CompletionStage<Stage.Session> openSession(String tenantId) {
 		return stage( v -> connection( tenantId )
 				.thenCompose( connection -> create( connection, () -> new ReactiveSessionImpl( delegate, options( tenantId ), connection ) ) )
 				.thenApply( s -> new StageSessionImpl(s, this) ) );
 	}
 
-	CompletionStage<Stage.StatelessSession> newStatelessSession() {
+	@Override
+	public CompletionStage<Stage.StatelessSession> openStatelessSession() {
 		SessionCreationOptions options = options();
 		return stage( v -> connection( options.getTenantIdentifier() )
 				.thenCompose( connection -> create( connection, () -> new ReactiveStatelessSessionImpl( delegate, options, connection ) ) )
 				.thenApply( s -> new StageStatelessSessionImpl(s, this) ) );
 	}
 
-	CompletionStage<Stage.StatelessSession> newStatelessSession(String tenantId) {
+	@Override
+	public CompletionStage<Stage.StatelessSession> openStatelessSession(String tenantId) {
 		return stage( v -> connection( tenantId )
 				.thenCompose( connection -> create( connection, () -> new ReactiveStatelessSessionImpl( delegate, options( tenantId ), connection ) ) )
 				.thenApply( s -> new StageStatelessSessionImpl( s, this ) ) );
@@ -157,20 +128,14 @@ private CompletionStage<ReactiveConnection> connection(String tenantId) {
 				: connectionPool.getConnection( tenantId );
 	}
 
-	private ReactiveConnection proxyConnection(String tenantId) {
-		return tenantId==null
-				? connectionPool.getProxyConnection()
-				: connectionPool.getProxyConnection( tenantId );
-	}
-
 	@Override
 	public <T> CompletionStage<T> withSession(Function<Stage.Session, CompletionStage<T>> work) {
 		Objects.requireNonNull( work, "parameter 'work' is required" );
 		Stage.Session current = context.get( contextKeyForSession );
 		if ( current!=null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newSession(), work, contextKeyForSession );
+		return withSession( openSession(), work, contextKeyForSession );
 	}
 
 	@Override
@@ -182,7 +147,7 @@ public <T> CompletionStage<T> withSession(String tenantId, Function<Stage.Sessio
 		if ( current!=null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newSession( tenantId ), work, key );
+		return withSession( openSession( tenantId ), work, key );
 	}
 
 	@Override
@@ -192,7 +157,7 @@ public <T> CompletionStage<T> withStatelessSession(Function<Stage.StatelessSessi
 		if ( current!=null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newStatelessSession(), work, contextKeyForStatelessSession );
+		return withSession( openStatelessSession(), work, contextKeyForStatelessSession );
 	}
 
 	@Override
@@ -204,7 +169,7 @@ public <T> CompletionStage<T> withStatelessSession(String tenantId, Function<Sta
 		if ( current != null && current.isOpen() ) {
 			return work.apply( current );
 		}
-		return withSession( newStatelessSession(tenantId), work, key );
+		return withSession( openStatelessSession( tenantId), work, key );
 	}
 
 	private <S extends Stage.Closeable, T> CompletionStage<T> withSession(