Skip to content

Commit 0f7de03

Browse files
DavideDDavideD
committed
[#2518] Add createSession and createStatelessSession
In Mutiny and Stage classes
1 parent f8fd105 commit 0f7de03

File tree

6 files changed

+366
-11
lines changed

6 files changed

+366
-11
lines changed

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

Lines changed: 58 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,11 @@
55
*/
66
package org.hibernate.reactive.mutiny;
77

8+
import java.lang.invoke.MethodHandles;
9+
import java.util.List;
10+
import java.util.function.BiFunction;
11+
import java.util.function.Function;
12+
813
import org.hibernate.Cache;
914
import org.hibernate.CacheMode;
1015
import org.hibernate.Filter;
@@ -44,10 +49,6 @@
4449
import jakarta.persistence.criteria.CriteriaUpdate;
4550
import jakarta.persistence.metamodel.Attribute;
4651
import jakarta.persistence.metamodel.Metamodel;
47-
import java.lang.invoke.MethodHandles;
48-
import java.util.List;
49-
import java.util.function.BiFunction;
50-
import java.util.function.Function;
5152

5253
import static org.hibernate.engine.internal.ManagedTypeHelper.asPersistentAttributeInterceptable;
5354
import static org.hibernate.engine.internal.ManagedTypeHelper.isPersistentAttributeInterceptable;
@@ -1971,6 +1972,59 @@ interface Transaction {
19711972
*/
19721973
interface SessionFactory extends AutoCloseable {
19731974

1975+
/**
1976+
* Obtain a new {@link Session reactive session}, the main
1977+
* interaction point between the user's program and Hibernate
1978+
* Reactive.
1979+
* <p>
1980+
* The underlying database connection is obtained lazily
1981+
* when the returned {@link Session} needs to access the
1982+
* database.
1983+
* <p>
1984+
* The client must close the session using {@link Session#close()}.
1985+
*/
1986+
@Incubating
1987+
Session createSession();
1988+
1989+
/**
1990+
* Obtain a new {@link Session reactive session}.
1991+
* <p>
1992+
* The underlying database connection is obtained lazily
1993+
* when the returned {@link Session} needs to access the
1994+
* database.
1995+
* <p>
1996+
* The client must close the session using {@link Session#close()}.
1997+
* @param tenantId the id of the tenant
1998+
*/
1999+
@Incubating
2000+
Session createSession(String tenantId);
2001+
2002+
/**
2003+
* Obtain a new {@link StatelessSession reactive stateless session}.
2004+
* <p>
2005+
* The underlying database connection is obtained lazily
2006+
* when the returned {@link StatelessSession} needs to access the
2007+
* database.
2008+
* <p>
2009+
* The client must close the session using {@link Session#close()}.
2010+
*/
2011+
@Incubating
2012+
StatelessSession createStatelessSession();
2013+
2014+
/**
2015+
* Obtain a new {@link StatelessSession reactive stateless session}.
2016+
* <p>
2017+
* The underlying database connection is obtained lazily
2018+
* when the returned {@link StatelessSession} needs to access the
2019+
* database.
2020+
* <p>
2021+
* The client must close the session using {@link Session#close()}.
2022+
*
2023+
* @param tenantId the id of the tenant
2024+
*/
2025+
@Incubating
2026+
StatelessSession createStatelessSession(String tenantId);
2027+
19742028
/**
19752029
* Obtain a new {@link Session reactive session} {@link Uni}, the main
19762030
* interaction point between the user's program and Hibernate

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

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@
2828
import org.hibernate.reactive.mutiny.Mutiny;
2929
import org.hibernate.reactive.pool.ReactiveConnection;
3030
import org.hibernate.reactive.pool.ReactiveConnectionPool;
31+
import org.hibernate.reactive.session.ReactiveSession;
32+
import org.hibernate.reactive.session.ReactiveStatelessSession;
3133
import org.hibernate.reactive.session.impl.ReactiveSessionImpl;
3234
import org.hibernate.reactive.session.impl.ReactiveStatelessSessionImpl;
3335
import org.hibernate.service.ServiceRegistry;
@@ -88,6 +90,32 @@ public Context getContext() {
8890
return context;
8991
}
9092

93+
@Override
94+
public Mutiny.Session createSession() {
95+
return createSession( getTenantIdentifier( options() ) );
96+
}
97+
98+
@Override
99+
public Mutiny.Session createSession(String tenantId) {
100+
final SessionCreationOptions options = options();
101+
ReactiveConnectionPool pool = delegate.getServiceRegistry().getService( ReactiveConnectionPool.class );
102+
ReactiveSession sessionImpl = new ReactiveSessionImpl( delegate, options, pool.getProxyConnection( tenantId ) );
103+
return new MutinySessionImpl( sessionImpl, this );
104+
}
105+
106+
@Override
107+
public Mutiny.StatelessSession createStatelessSession() {
108+
return createStatelessSession( getTenantIdentifier( options() ) );
109+
}
110+
111+
@Override
112+
public Mutiny.StatelessSession createStatelessSession(String tenantId) {
113+
final SessionCreationOptions options = options();
114+
ReactiveConnectionPool pool = delegate.getServiceRegistry().getService( ReactiveConnectionPool.class );
115+
ReactiveStatelessSession sessionImpl = new ReactiveStatelessSessionImpl( delegate, options, pool.getProxyConnection( tenantId ) );
116+
return new MutinyStatelessSessionImpl( sessionImpl, this );
117+
}
118+
91119
@Override
92120
public Uni<Mutiny.Session> openSession() {
93121
SessionCreationOptions options = options();

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

Lines changed: 21 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,12 +37,25 @@
3737
*/
3838
@Incubating
3939
public interface ReactiveConnectionPool extends Service {
40+
/**
41+
* Obtain a lazily-initializing reactive connection. The
42+
* actual connection might be made when the returned
43+
* instance if {@link ReactiveConnection} is first used.
44+
*/
45+
ReactiveConnection getProxyConnection();
4046

4147
/**
4248
* Obtain a reactive connection, returning the connection
43-
* via a {@link CompletionStage}.
49+
* via a {@link CompletionStage} and overriding the default
50+
* {@link SqlExceptionHelper} for the pool.
4451
*/
45-
CompletionStage<ReactiveConnection> getConnection();
52+
ReactiveConnection getProxyConnection(SqlExceptionHelper sqlExceptionHelper);
53+
54+
/**
55+
* Obtain a reactive connection for the given tenant id,
56+
* returning the connection via a {@link CompletionStage}.
57+
*/
58+
ReactiveConnection getProxyConnection(String tenantId);
4659

4760
/**
4861
* Obtain a reactive connection, returning the connection
@@ -57,6 +70,12 @@ public interface ReactiveConnectionPool extends Service {
5770
*/
5871
CompletionStage<ReactiveConnection> getConnection(String tenantId);
5972

73+
/**
74+
* Obtain a reactive connection, returning the connection
75+
* via a {@link CompletionStage}.
76+
*/
77+
CompletionStage<ReactiveConnection> getConnection();
78+
6079
/**
6180
* Obtain a reactive connection for the given tenant id,
6281
* returning the connection via a {@link CompletionStage}

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

Lines changed: 180 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,10 +7,12 @@
77

88
import java.sql.ResultSet;
99
import java.sql.SQLException;
10+
import java.util.List;
1011
import java.util.Objects;
1112
import java.util.concurrent.CompletableFuture;
1213
import java.util.concurrent.CompletionStage;
1314
import java.util.function.Consumer;
15+
import java.util.function.Supplier;
1416

1517
import org.hibernate.engine.jdbc.internal.FormatStyle;
1618
import org.hibernate.engine.jdbc.spi.SqlExceptionHelper;
@@ -26,8 +28,11 @@
2628
import io.vertx.sqlclient.RowSet;
2729
import io.vertx.sqlclient.SqlConnection;
2830
import io.vertx.sqlclient.Tuple;
31+
import io.vertx.sqlclient.spi.DatabaseMetadata;
2932

33+
import static org.hibernate.reactive.util.impl.CompletionStages.completedFuture;
3034
import static org.hibernate.reactive.util.impl.CompletionStages.rethrow;
35+
import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
3136

3237
/**
3338
* A pool of reactive connections backed by a supplier of
@@ -72,13 +77,31 @@ public abstract class SqlClientPool implements ReactiveConnectionPool {
7277
* subclasses which support multitenancy.
7378
*
7479
* @param tenantId the id of the tenant
80+
*
7581
* @throws UnsupportedOperationException if multitenancy is not supported
7682
* @see ReactiveConnectionPool#getConnection(String)
7783
*/
7884
protected Pool getTenantPool(String tenantId) {
7985
throw new UnsupportedOperationException( "multitenancy not supported by built-in SqlClientPool" );
8086
}
8187

88+
@Override
89+
public ReactiveConnection getProxyConnection() {
90+
return new ProxyConnection( this::getConnection );
91+
}
92+
93+
@Override
94+
public ReactiveConnection getProxyConnection(String tenantId) {
95+
return tenantId == null
96+
? new ProxyConnection( this::getConnection )
97+
: new ProxyConnection( () -> getConnection( tenantId ) );
98+
}
99+
100+
@Override
101+
public ReactiveConnection getProxyConnection(SqlExceptionHelper sqlExceptionHelper) {
102+
return new ProxyConnection( () -> getConnection( sqlExceptionHelper ) );
103+
}
104+
82105
@Override
83106
public CompletionStage<ReactiveConnection> getConnection() {
84107
return getConnectionFromPool( getPool() );
@@ -143,10 +166,13 @@ private <T> T convertException(T rows, String sql, Throwable sqlException) {
143166
if ( sqlException == null ) {
144167
return rows;
145168
}
146-
if ( sqlException instanceof DatabaseException ) {
147-
DatabaseException de = (DatabaseException) sqlException;
169+
if ( sqlException instanceof DatabaseException de ) {
148170
sqlException = getSqlExceptionHelper()
149-
.convert( new SQLException( de.getMessage(), de.getSqlState(), de.getErrorCode() ), "error executing SQL statement", sql );
171+
.convert(
172+
new SQLException( de.getMessage(), de.getSqlState(), de.getErrorCode() ),
173+
"error executing SQL statement",
174+
sql
175+
);
150176
}
151177
return rethrow( sqlException );
152178
}
@@ -186,4 +212,155 @@ private SqlClientConnection newConnection(SqlConnection connection) {
186212
private SqlClientConnection newConnection(SqlConnection connection, SqlExceptionHelper sqlExceptionHelper) {
187213
return new SqlClientConnection( connection, getPool(), getSqlStatementLogger(), sqlExceptionHelper );
188214
}
215+
216+
private static class ProxyConnection implements ReactiveConnection {
217+
private final Supplier<CompletionStage<ReactiveConnection>> connectionSupplier;
218+
private Integer batchSize;
219+
private ReactiveConnection connection;
220+
221+
public ProxyConnection(Supplier<CompletionStage<ReactiveConnection>> connectionSupplier) {
222+
this.connectionSupplier = connectionSupplier;
223+
}
224+
225+
/**
226+
* @return the existing {@link ReactiveConnection}, or open a new one
227+
*/
228+
CompletionStage<ReactiveConnection> connection() {
229+
if ( connection == null ) {
230+
return connectionSupplier.get()
231+
.thenApply( conn -> {
232+
if ( batchSize != null ) {
233+
conn.withBatchSize( batchSize );
234+
}
235+
connection = conn;
236+
return connection;
237+
} );
238+
}
239+
return completedFuture( connection );
240+
}
241+
242+
@Override
243+
public boolean isTransactionInProgress() {
244+
return connection != null && connection.isTransactionInProgress();
245+
}
246+
247+
@Override
248+
public DatabaseMetadata getDatabaseMetadata() {
249+
Objects.requireNonNull( connection, "Database metadata not available until the connection is opened" );
250+
return connection.getDatabaseMetadata();
251+
}
252+
253+
@Override
254+
public CompletionStage<Void> execute(String sql) {
255+
return connection().thenCompose( conn -> conn.execute( sql ) );
256+
}
257+
258+
@Override
259+
public CompletionStage<Void> executeOutsideTransaction(String sql) {
260+
return connection().thenCompose( conn -> conn.executeOutsideTransaction( sql ) );
261+
}
262+
263+
@Override
264+
public CompletionStage<Void> executeUnprepared(String sql) {
265+
return connection().thenCompose( conn -> conn.executeUnprepared( sql ) );
266+
}
267+
268+
@Override
269+
public CompletionStage<Integer> update(String sql) {
270+
return connection().thenCompose( conn -> conn.update( sql ) );
271+
}
272+
273+
@Override
274+
public CompletionStage<Integer> update(String sql, Object[] paramValues) {
275+
return connection().thenCompose( conn -> conn.update( sql, paramValues ) );
276+
}
277+
278+
@Override
279+
public CompletionStage<Void> update(String sql, Object[] paramValues, boolean allowBatching, Expectation expectation) {
280+
return connection().thenCompose( conn -> conn.update( sql, paramValues, allowBatching, expectation ) );
281+
}
282+
283+
@Override
284+
public CompletionStage<int[]> update(String sql, List<Object[]> paramValues) {
285+
return connection().thenCompose( conn -> conn.update( sql, paramValues ) );
286+
}
287+
288+
@Override
289+
public CompletionStage<Result> select(String sql) {
290+
return connection().thenCompose( conn -> conn.select( sql ) );
291+
}
292+
293+
@Override
294+
public CompletionStage<Result> select(String sql, Object[] paramValues) {
295+
return connection().thenCompose( conn -> conn.select( sql ) );
296+
}
297+
298+
@Override
299+
public CompletionStage<ResultSet> selectJdbc(String sql, Object[] paramValues) {
300+
return connection().thenCompose( conn -> conn.selectJdbc( sql, paramValues ) );
301+
}
302+
303+
@Override
304+
public <T> CompletionStage<T> insertAndSelectIdentifier(
305+
String sql,
306+
Object[] paramValues,
307+
Class<T> idClass,
308+
String idColumnName) {
309+
return connection().thenCompose( conn -> conn
310+
.insertAndSelectIdentifier( sql, paramValues, idClass, idColumnName ) );
311+
}
312+
313+
@Override
314+
public CompletionStage<ResultSet> insertAndSelectIdentifierAsResultSet(
315+
String sql,
316+
Object[] paramValues,
317+
Class<?> idClass,
318+
String idColumnName) {
319+
return connection().thenCompose( conn -> conn
320+
.insertAndSelectIdentifierAsResultSet( sql, paramValues, idClass, idColumnName ) );
321+
}
322+
323+
@Override
324+
public <T> CompletionStage<T> selectIdentifier(String sql, Object[] paramValues, Class<T> idClass) {
325+
return connection().thenCompose( conn -> conn.selectIdentifier( sql, paramValues, idClass ) );
326+
}
327+
328+
@Override
329+
public CompletionStage<Void> beginTransaction() {
330+
return connection().thenCompose( ReactiveConnection::beginTransaction );
331+
}
332+
333+
@Override
334+
public CompletionStage<Void> commitTransaction() {
335+
return connection().thenCompose( ReactiveConnection::commitTransaction );
336+
}
337+
338+
@Override
339+
public CompletionStage<Void> rollbackTransaction() {
340+
return connection().thenCompose( ReactiveConnection::rollbackTransaction );
341+
}
342+
343+
@Override
344+
public ReactiveConnection withBatchSize(int batchSize) {
345+
if ( connection == null ) {
346+
this.batchSize = batchSize;
347+
}
348+
else {
349+
connection = connection.withBatchSize( batchSize );
350+
}
351+
return this;
352+
}
353+
354+
@Override
355+
public CompletionStage<Void> executeBatch() {
356+
return connection().thenCompose( ReactiveConnection::executeBatch );
357+
}
358+
359+
@Override
360+
public CompletionStage<Void> close() {
361+
return connection != null
362+
? connection.close().thenAccept( v -> connection = null )
363+
: voidFuture();
364+
}
365+
}
189366
}

0 commit comments

Comments
 (0)