Skip to content

Commit b2e0e86

Browse files
gavinkinggavinking
committed
[hibernate#2156] add withConnection()
1 parent 27acfb2 commit b2e0e86

File tree

11 files changed

+129
-0
lines changed

11 files changed

+129
-0
lines changed

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/common/ConnectionConsumer.java

+24
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
/* Hibernate, Relational Persistence for Idiomatic Java
2+
*
3+
* SPDX-License-Identifier: Apache-2.0
4+
* Copyright: Red Hat Inc. and Hibernate Authors
5+
*/
6+
package org.hibernate.reactive.common;
7+
8+
import org.hibernate.Incubating;
9+
10+
import java.util.concurrent.CompletionStage;
11+
12+
/**
13+
* An operation which makes direct use of a database connection.
14+
*
15+
* @param <C> the connection type, usually
16+
* {@link io.vertx.sqlclient.SqlConnection}
17+
* @param <R> the result type of the operation, or {@link Void}
18+
*
19+
* @author Gavin King
20+
*/
21+
@Incubating @FunctionalInterface
22+
public interface ConnectionConsumer<C, R> {
23+
CompletionStage<R> accept(C connection);
24+
}

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

+25
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.hibernate.query.criteria.HibernateCriteriaBuilder;
2929
import org.hibernate.query.criteria.JpaCriteriaInsert;
3030
import org.hibernate.reactive.common.AffectedEntities;
31+
import org.hibernate.reactive.common.ConnectionConsumer;
3132
import org.hibernate.reactive.common.Identifier;
3233
import org.hibernate.reactive.common.ResultSetMapping;
3334
import org.hibernate.reactive.logging.impl.Log;
@@ -1525,6 +1526,18 @@ default Session setCacheRetrieveMode(CacheRetrieveMode cacheRetrieveMode) {
15251526
* The {@link SessionFactory} which created this session.
15261527
*/
15271528
SessionFactory getFactory();
1529+
1530+
/**
1531+
* Execute the given operation using the connection underlying the reactive session.
1532+
*
1533+
* @param consumer the operation to be executed
1534+
* @return the result of the operation via a {@link Uni}
1535+
*
1536+
* @param <C> the connection type, usually
1537+
* {@link io.vertx.sqlclient.SqlConnection}
1538+
* @param <R> the result type of the operation, or {@link Void}
1539+
*/
1540+
<C,R> Uni<R> withConnection(ConnectionConsumer<C,R> consumer);
15281541
}
15291542

15301543
/**
@@ -2104,6 +2117,18 @@ default Uni<Void> refresh(Object entity, LockModeType lockModeType) {
21042117
* The {@link SessionFactory} which created this session.
21052118
*/
21062119
SessionFactory getFactory();
2120+
2121+
/**
2122+
* Execute the given operation using the connection underlying the reactive session.
2123+
*
2124+
* @param consumer the operation to be executed
2125+
* @return the result of the operation via a {@link Uni}
2126+
*
2127+
* @param <C> the connection type, usually
2128+
* {@link io.vertx.sqlclient.SqlConnection}
2129+
* @param <R> the result type of the operation, or {@link Void}
2130+
*/
2131+
<C,R> Uni<R> withConnection(ConnectionConsumer<C,R> consumer);
21072132
}
21082133

21092134
/**

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/mutiny/impl/MutinySessionImpl.java

+5
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.hibernate.graph.RootGraph;
2525
import org.hibernate.query.criteria.JpaCriteriaInsert;
2626
import org.hibernate.reactive.common.AffectedEntities;
27+
import org.hibernate.reactive.common.ConnectionConsumer;
2728
import org.hibernate.reactive.common.Identifier;
2829
import org.hibernate.reactive.common.ResultSetMapping;
2930
import org.hibernate.reactive.engine.spi.ReactiveSharedSessionContractImplementor;
@@ -606,4 +607,8 @@ else if ( ReactiveSharedSessionContractImplementor.class.isAssignableFrom( clazz
606607
throw new PersistenceException( "Cannot unwrap type " + clazz );
607608
}
608609

610+
@Override
611+
public <C, T> Uni<T> withConnection(ConnectionConsumer<C, T> consumer) {
612+
return uni( () -> delegate.getReactiveConnection().withConnection( consumer ) );
613+
}
609614
}

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/mutiny/impl/MutinyStatelessSessionImpl.java

+6
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
import jakarta.persistence.criteria.CriteriaUpdate;
1313
import org.hibernate.LockMode;
1414
import org.hibernate.graph.spi.RootGraphImplementor;
15+
import org.hibernate.reactive.common.ConnectionConsumer;
1516
import org.hibernate.reactive.common.ResultSetMapping;
1617
import org.hibernate.reactive.mutiny.Mutiny;
1718
import org.hibernate.reactive.mutiny.Mutiny.Query;
@@ -370,4 +371,9 @@ public <T> EntityGraph<T> createEntityGraph(Class<T> rootType) {
370371
public <T> EntityGraph<T> createEntityGraph(Class<T> rootType, String graphName) {
371372
return delegate.createEntityGraph( rootType, graphName );
372373
}
374+
375+
@Override
376+
public <C, T> Uni<T> withConnection(ConnectionConsumer<C, T> consumer) {
377+
return uni( () -> delegate.getReactiveConnection().withConnection( consumer ) );
378+
}
373379
}

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/BatchingConnection.java

+6
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
import org.hibernate.reactive.adaptor.impl.ResultSetAdaptor;
1515

1616
import io.vertx.sqlclient.spi.DatabaseMetadata;
17+
import org.hibernate.reactive.common.ConnectionConsumer;
1718

1819
import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
1920

@@ -214,4 +215,9 @@ public CompletionStage<Void> rollbackTransaction() {
214215
public CompletionStage<Void> close() {
215216
return delegate.close();
216217
}
218+
219+
@Override
220+
public <C, R> CompletionStage<R> withConnection(ConnectionConsumer<C, R> consumer) {
221+
return delegate.withConnection( consumer );
222+
}
217223
}

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/ReactiveConnection.java

+3
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
import org.hibernate.Incubating;
1414

1515
import io.vertx.sqlclient.spi.DatabaseMetadata;
16+
import org.hibernate.reactive.common.ConnectionConsumer;
1617

1718
/**
1819
* Abstracts over reactive database connections, defining
@@ -80,4 +81,6 @@ interface Result extends Iterator<Object[]> {
8081
CompletionStage<Void> executeBatch();
8182

8283
CompletionStage<Void> close();
84+
85+
<C,R> CompletionStage<R> withConnection(ConnectionConsumer<C,R> consumer);
8386
}

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/pool/impl/SqlClientConnection.java

+6
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
import org.hibernate.engine.jdbc.spi.SqlStatementLogger;
1919
import org.hibernate.reactive.adaptor.impl.JdbcNull;
2020
import org.hibernate.reactive.adaptor.impl.ResultSetAdaptor;
21+
import org.hibernate.reactive.common.ConnectionConsumer;
2122
import org.hibernate.reactive.logging.impl.Log;
2223
import org.hibernate.reactive.logging.impl.LoggerFactory;
2324
import org.hibernate.reactive.pool.BatchingConnection;
@@ -325,6 +326,11 @@ public CompletionStage<Void> close() {
325326
.toCompletionStage();
326327
}
327328

329+
@Override @SuppressWarnings("unchecked")
330+
public <C,R> CompletionStage<R> withConnection(ConnectionConsumer<C,R> consumer) {
331+
return consumer.accept( (C) client() );
332+
}
333+
328334
@SuppressWarnings("unchecked")
329335
private static <T> T getLastInsertedId(RowSet<Row> rows, Class<T> idClass, String idColumnName) {
330336
final Long mySqlId = rows.property( MYSQL_LAST_INSERTED_ID );

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/Stage.java

+25
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
import org.hibernate.query.criteria.HibernateCriteriaBuilder;
3030
import org.hibernate.query.criteria.JpaCriteriaInsert;
3131
import org.hibernate.reactive.common.AffectedEntities;
32+
import org.hibernate.reactive.common.ConnectionConsumer;
3233
import org.hibernate.reactive.common.Identifier;
3334
import org.hibernate.reactive.common.ResultSetMapping;
3435
import org.hibernate.reactive.logging.impl.Log;
@@ -1557,6 +1558,18 @@ default Session setCacheRetrieveMode(CacheRetrieveMode cacheRetrieveMode) {
15571558
* The {@link SessionFactory} which created this session.
15581559
*/
15591560
SessionFactory getFactory();
1561+
1562+
/**
1563+
* Execute the given operation using the connection underlying the reactive session.
1564+
*
1565+
* @param consumer the operation to be executed
1566+
* @return the result of the operation via a {@link CompletionStage}
1567+
*
1568+
* @param <C> the connection type, usually
1569+
* {@link io.vertx.sqlclient.SqlConnection}
1570+
* @param <R> the result type of the operation, or {@link Void}
1571+
*/
1572+
<C,R> CompletionStage<R> withConnection(ConnectionConsumer<C,R> consumer);
15601573
}
15611574

15621575
/**
@@ -2155,6 +2168,18 @@ default CompletionStage<Void> refresh(Object entity, LockModeType lockModeType)
21552168
* The {@link SessionFactory} which created this session.
21562169
*/
21572170
SessionFactory getFactory();
2171+
2172+
/**
2173+
* Execute the given operation using the connection underlying the reactive session.
2174+
*
2175+
* @param consumer the operation to be executed
2176+
* @return the result of the operation via a {@link CompletionStage}
2177+
*
2178+
* @param <C> the connection type, usually
2179+
* {@link io.vertx.sqlclient.SqlConnection}
2180+
* @param <R> the result type of the operation, or {@link Void}
2181+
*/
2182+
<C,R> CompletionStage<R> withConnection(ConnectionConsumer<C,R> consumer);
21582183
}
21592184

21602185
/**

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/impl/StageSessionImpl.java

+6
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.hibernate.jpa.internal.util.LockModeTypeHelper;
2525
import org.hibernate.query.criteria.JpaCriteriaInsert;
2626
import org.hibernate.reactive.common.AffectedEntities;
27+
import org.hibernate.reactive.common.ConnectionConsumer;
2728
import org.hibernate.reactive.common.Identifier;
2829
import org.hibernate.reactive.common.ResultSetMapping;
2930
import org.hibernate.reactive.engine.ReactiveActionQueue;
@@ -609,4 +610,9 @@ else if ( ReactiveSharedSessionContractImplementor.class.isAssignableFrom( clazz
609610
}
610611
throw new PersistenceException( "Cannot unwrap type " + clazz );
611612
}
613+
614+
@Override
615+
public <C, T> CompletionStage<T> withConnection(ConnectionConsumer<C, T> consumer) {
616+
return delegate.getReactiveConnection().withConnection( consumer );
617+
}
612618
}

Diff for: ‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/stage/impl/StageStatelessSessionImpl.java

+6
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import org.hibernate.LockMode;
99
import org.hibernate.graph.spi.RootGraphImplementor;
1010
import org.hibernate.query.criteria.JpaCriteriaInsert;
11+
import org.hibernate.reactive.common.ConnectionConsumer;
1112
import org.hibernate.reactive.common.ResultSetMapping;
1213
import org.hibernate.reactive.pool.ReactiveConnection;
1314
import org.hibernate.reactive.session.ReactiveStatelessSession;
@@ -373,4 +374,9 @@ public <T> EntityGraph<T> createEntityGraph(Class<T> rootType) {
373374
public <T> EntityGraph<T> createEntityGraph(Class<T> rootType, String graphName) {
374375
return delegate.createEntityGraph( rootType, graphName );
375376
}
377+
378+
@Override
379+
public <C, T> CompletionStage<T> withConnection(ConnectionConsumer<C, T> consumer) {
380+
return delegate.getReactiveConnection().withConnection( consumer );
381+
}
376382
}

Diff for: ‎hibernate-reactive-core/src/test/java/org/hibernate/reactive/MutinySessionTest.java

+17
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import java.util.Objects;
1111
import java.util.concurrent.atomic.AtomicInteger;
1212

13+
import io.vertx.sqlclient.SqlConnection;
1314
import org.hibernate.LockMode;
1415
import org.hibernate.reactive.mutiny.Mutiny;
1516

@@ -668,6 +669,22 @@ public void testForceFlushWithDelete(VertxTestContext context) {
668669
);
669670
}
670671

672+
@Test
673+
public void reactiveWithConnection(VertxTestContext context) {
674+
test(
675+
context,
676+
getMutinySessionFactory()
677+
.withTransaction( (s,t) -> s.withConnection((SqlConnection c)
678+
-> c.query("select name from Pig").execute().toCompletionStage() ) )
679+
.invoke( res -> res.forEach( row -> {
680+
assertEquals(1, row.size() );
681+
assertEquals("Aloi", row.getString("name") );
682+
}) )
683+
);
684+
}
685+
686+
687+
671688
private void assertThatPigsAreEqual(GuineaPig expected, GuineaPig actual) {
672689
assertNotNull( actual );
673690
assertEquals( expected.getId(), actual.getId() );

0 commit comments

Comments
 (0)