Skip to content

Commit 919aaa6

Browse files
dreab8dreab8
committed
[hibernate#1905] Reactive find with lock in Quarkus with reactive hibernate
1 parent 923452b commit 919aaa6

16 files changed

+324
-25
lines changed

‎hibernate-reactive-core/src/main/java/org/hibernate/engine/internal/ReactivePersistenceContextAdapter.java

+93
Original file line numberDiff line numberDiff line change
@@ -12,19 +12,27 @@
1212

1313
import org.hibernate.HibernateException;
1414
import org.hibernate.collection.spi.PersistentCollection;
15+
import org.hibernate.engine.spi.EntityHolder;
1516
import org.hibernate.engine.spi.EntityKey;
1617
import org.hibernate.engine.spi.PersistenceContext;
1718
import org.hibernate.engine.spi.SessionImplementor;
1819
import org.hibernate.engine.spi.SharedSessionContractImplementor;
20+
import org.hibernate.event.service.spi.EventListenerGroup;
21+
import org.hibernate.event.spi.PostLoadEvent;
22+
import org.hibernate.event.spi.PostLoadEventListener;
1923
import org.hibernate.persister.entity.EntityPersister;
24+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
2025
import org.hibernate.reactive.logging.impl.Log;
2126
import org.hibernate.reactive.persister.entity.impl.ReactiveEntityPersister;
2227
import org.hibernate.reactive.session.ReactiveSession;
28+
import org.hibernate.sql.exec.spi.Callback;
29+
import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingState;
2330

2431
import static java.lang.invoke.MethodHandles.lookup;
2532
import static org.hibernate.pretty.MessageHelper.infoString;
2633
import static org.hibernate.reactive.logging.impl.LoggerFactory.make;
2734
import static org.hibernate.reactive.util.impl.CompletionStages.completedFuture;
35+
import static org.hibernate.reactive.util.impl.CompletionStages.loop;
2836
import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
2937

3038
/**
@@ -130,4 +138,89 @@ public Object removeEntity(EntityKey key) {
130138
}
131139
return result;
132140
}
141+
142+
@Override
143+
public void postLoad(JdbcValuesSourceProcessingState processingState, Consumer<EntityHolder> holderConsumer) {
144+
throw LOG.nonReactiveMethodCall( "reactivePostLoad(JdbcValuesSourceProcessingState, Consumer<EntityHolder>) )" );
145+
}
146+
147+
/**
148+
* Reactive version of {@link StatefulPersistenceContext#postLoad(JdbcValuesSourceProcessingState, Consumer)}
149+
*
150+
*/
151+
public CompletionStage<Void> reactivePostLoad(JdbcValuesSourceProcessingState processingState, Consumer<EntityHolder> holderConsumer) {
152+
final ReactiveCallbackImpl callback = (ReactiveCallbackImpl) processingState.getExecutionContext().getCallback();
153+
154+
if ( processingState.getLoadingEntityHolders() != null ) {
155+
final EventListenerGroup<PostLoadEventListener> listenerGroup =
156+
getSession().getFactory().getEventListenerGroups().eventListenerGroup_POST_LOAD;
157+
final PostLoadEvent postLoadEvent = processingState.getPostLoadEvent();
158+
return loop(
159+
processingState.getLoadingEntityHolders(), entityHolder ->
160+
processLoadedEntityHolder(
161+
entityHolder,
162+
listenerGroup,
163+
postLoadEvent,
164+
callback,
165+
holderConsumer
166+
)
167+
).thenAccept( unused -> processingState.getLoadingEntityHolders().clear() );
168+
}
169+
if ( processingState.getReloadedEntityHolders() != null ) {
170+
return loop(
171+
processingState.getLoadingEntityHolders(), entityHolder ->
172+
processLoadedEntityHolder(
173+
entityHolder,
174+
null,
175+
null,
176+
callback,
177+
holderConsumer
178+
)
179+
).thenAccept( unused -> processingState.getLoadingEntityHolders().clear() );
180+
}
181+
return voidFuture();
182+
}
183+
184+
/**
185+
* Reactive version of {@link StatefulPersistenceContext#processLoadedEntityHolder(EntityHolder, EventListenerGroup, PostLoadEvent, Callback, Consumer)}
186+
*/
187+
private CompletionStage<Void> processLoadedEntityHolder(
188+
EntityHolder holder,
189+
EventListenerGroup<PostLoadEventListener> listenerGroup,
190+
PostLoadEvent postLoadEvent,
191+
ReactiveCallbackImpl callback,
192+
Consumer<EntityHolder> holderConsumer) {
193+
if ( holderConsumer != null ) {
194+
holderConsumer.accept( holder );
195+
}
196+
if ( holder.getEntity() == null ) {
197+
// It's possible that we tried to load an entity and found out it doesn't exist,
198+
// in which case we added an entry with a null proxy and entity.
199+
// Remove that empty entry on post load to avoid unwanted side effects
200+
getEntitiesByKey().remove( holder.getEntityKey() );
201+
}
202+
else {
203+
if ( postLoadEvent != null ) {
204+
postLoadEvent.reset();
205+
postLoadEvent.setEntity( holder.getEntity() )
206+
.setId( holder.getEntityKey().getIdentifier() )
207+
.setPersister( holder.getDescriptor() );
208+
listenerGroup.fireEventOnEachListener(
209+
postLoadEvent,
210+
PostLoadEventListener::onPostLoad
211+
);
212+
if ( callback != null ) {
213+
return callback.invokeReactiveLoadActions(
214+
holder.getEntity(),
215+
holder.getDescriptor(),
216+
getSession()
217+
).thenAccept( v ->
218+
holder.resetEntityInitialier()
219+
);
220+
}
221+
}
222+
223+
}
224+
return voidFuture();
225+
}
133226
}

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/engine/impl/ReactiveCallbackImpl.java

+70
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
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.engine.impl;
7+
8+
import java.util.ArrayList;
9+
import java.util.List;
10+
import java.util.concurrent.CompletionStage;
11+
12+
import org.hibernate.engine.spi.SharedSessionContractImplementor;
13+
import org.hibernate.loader.ast.spi.AfterLoadAction;
14+
import org.hibernate.metamodel.mapping.EntityMappingType;
15+
import org.hibernate.reactive.loader.ast.spi.ReactiveAfterLoadAction;
16+
import org.hibernate.reactive.logging.impl.Log;
17+
import org.hibernate.sql.exec.spi.Callback;
18+
19+
import static java.lang.invoke.MethodHandles.lookup;
20+
import static org.hibernate.reactive.logging.impl.LoggerFactory.make;
21+
import static org.hibernate.reactive.util.impl.CompletionStages.loop;
22+
23+
/**
24+
* Reactive equivalent of {@link org.hibernate.sql.exec.internal.CallbackImpl}
25+
*/
26+
public class ReactiveCallbackImpl implements Callback {
27+
private static final Log LOG = make( Log.class, lookup() );
28+
29+
private final List<ReactiveAfterLoadAction> afterLoadActions;
30+
31+
public ReactiveCallbackImpl() {
32+
this.afterLoadActions = new ArrayList<>( 1 );
33+
}
34+
35+
@Override
36+
public void registerAfterLoadAction(AfterLoadAction afterLoadAction) {
37+
throw LOG.nonReactiveMethodCall( "registerReactiveAfterLoadAction(ReactiveCallbackImpl)" );
38+
}
39+
40+
public void registerReactiveAfterLoadAction(ReactiveAfterLoadAction afterLoadAction) {
41+
afterLoadActions.add( afterLoadAction );
42+
}
43+
44+
@Override
45+
public void invokeAfterLoadActions(
46+
Object entity,
47+
EntityMappingType entityMappingType,
48+
SharedSessionContractImplementor session) {
49+
throw LOG.nonReactiveMethodCall( "invokeAfterLoadActions(Object, EntityMappingType, SharedSessionContractImplementor)" );
50+
}
51+
52+
/**
53+
* Reactive version of {@link org.hibernate.sql.exec.internal.CallbackImpl#invokeAfterLoadActions(Object, EntityMappingType, SharedSessionContractImplementor)}
54+
*/
55+
public CompletionStage<Void> invokeReactiveLoadActions(
56+
Object entity,
57+
EntityMappingType entityMappingType,
58+
SharedSessionContractImplementor session) {
59+
return loop(
60+
afterLoadActions, afterLoadAction ->
61+
afterLoadAction.reactiveAfterLoad( entity, entityMappingType, session )
62+
);
63+
}
64+
65+
@Override
66+
public boolean hasAfterLoadActions() {
67+
return !afterLoadActions.isEmpty();
68+
}
69+
70+
}

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/event/impl/DefaultReactivePostLoadEventListener.java

+1
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
*/
66
package org.hibernate.reactive.event.impl;
77

8+
89
import org.hibernate.AssertionFailure;
910
import org.hibernate.engine.spi.EntityEntry;
1011
import org.hibernate.event.spi.EventSource;

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/loader/ast/internal/ReactiveNaturalIdLoaderDelegate.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import org.hibernate.metamodel.mapping.NaturalIdMapping;
2828
import org.hibernate.query.internal.SimpleQueryOptions;
2929
import org.hibernate.query.spi.QueryOptions;
30+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
3031
import org.hibernate.reactive.loader.ast.spi.ReactiveNaturalIdLoader;
3132
import org.hibernate.reactive.sql.exec.internal.StandardReactiveSelectExecutor;
3233
import org.hibernate.reactive.sql.results.spi.ReactiveListResultsConsumer;
@@ -38,7 +39,6 @@
3839
import org.hibernate.sql.ast.tree.select.QuerySpec;
3940
import org.hibernate.sql.ast.tree.select.SelectStatement;
4041
import org.hibernate.sql.exec.internal.BaseExecutionContext;
41-
import org.hibernate.sql.exec.internal.CallbackImpl;
4242
import org.hibernate.sql.exec.internal.JdbcParameterBindingsImpl;
4343
import org.hibernate.sql.exec.spi.Callback;
4444
import org.hibernate.sql.exec.spi.JdbcOperationQuerySelect;
@@ -335,7 +335,7 @@ public NaturalIdLoaderWithOptionsExecutionContext(
335335
QueryOptions queryOptions) {
336336
super( session );
337337
this.queryOptions = queryOptions;
338-
callback = new CallbackImpl();
338+
callback = new ReactiveCallbackImpl();
339339
}
340340

341341
@Override

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/loader/ast/internal/ReactiveSingleIdLoadPlan.java

+14-10
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,19 @@
2020
import org.hibernate.query.internal.SimpleQueryOptions;
2121
import org.hibernate.query.spi.QueryOptions;
2222
import org.hibernate.query.spi.QueryParameterBindings;
23+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
2324
import org.hibernate.reactive.sql.exec.internal.StandardReactiveSelectExecutor;
2425
import org.hibernate.reactive.sql.results.spi.ReactiveListResultsConsumer;
2526
import org.hibernate.resource.jdbc.spi.LogicalConnectionImplementor;
2627
import org.hibernate.sql.ast.tree.select.SelectStatement;
27-
import org.hibernate.sql.exec.internal.CallbackImpl;
2828
import org.hibernate.sql.exec.internal.JdbcParameterBindingsImpl;
2929
import org.hibernate.sql.exec.spi.Callback;
3030
import org.hibernate.sql.exec.spi.ExecutionContext;
3131
import org.hibernate.sql.exec.spi.JdbcParameterBindings;
3232
import org.hibernate.sql.exec.spi.JdbcParametersList;
3333

34+
import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
35+
3436
public class ReactiveSingleIdLoadPlan<T> extends SingleIdLoadPlan<CompletionStage<T>> {
3537

3638
public ReactiveSingleIdLoadPlan(
@@ -61,7 +63,7 @@ public CompletionStage<T> load(Object restrictedValue, Object entityInstance, Bo
6163
}
6264
assert offset == getJdbcParameters().size();
6365
final QueryOptions queryOptions = new SimpleQueryOptions( getLockOptions(), readOnly );
64-
final Callback callback = new CallbackImpl();
66+
final ReactiveCallbackImpl callback = new ReactiveCallbackImpl();
6567
EntityMappingType loadable = (EntityMappingType) getLoadable();
6668
ExecutionContext executionContext = executionContext(
6769
restrictedValue,
@@ -74,17 +76,19 @@ public CompletionStage<T> load(Object restrictedValue, Object entityInstance, Bo
7476
// FIXME: Should we get this from jdbcServices.getSelectExecutor()?
7577
return StandardReactiveSelectExecutor.INSTANCE
7678
.list( getJdbcSelect(), jdbcParameterBindings, executionContext, getRowTransformer(), resultConsumer( singleResultExpected ) )
77-
.thenApply( this::extractEntity )
78-
.thenApply( entity -> {
79-
invokeAfterLoadActions( callback, session, entity );
80-
return (T) entity;
81-
} );
79+
.thenCompose( list -> {
80+
Object entity = extractEntity( list );
81+
return invokeAfterLoadActions( callback, session, entity )
82+
.thenApply( v -> (T) entity );
83+
}
84+
);
8285
}
8386

84-
private <G> void invokeAfterLoadActions(Callback callback, SharedSessionContractImplementor session, G entity) {
85-
if ( entity != null && getLoadable() != null) {
86-
callback.invokeAfterLoadActions( entity, (EntityMappingType) getLoadable(), session );
87+
private <G> CompletionStage<Void> invokeAfterLoadActions(ReactiveCallbackImpl callback, SharedSessionContractImplementor session, G entity) {
88+
if ( entity != null && getLoadable() != null ) {
89+
return callback.invokeReactiveLoadActions( entity, (EntityMappingType) getLoadable(), session );
8790
}
91+
return voidFuture();
8892
}
8993

9094
private Object extractEntity(List<?> list) {

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/loader/ast/internal/ReactiveSingleUniqueKeyEntityLoaderStandard.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,13 @@
2424
import org.hibernate.metamodel.mapping.SingularAttributeMapping;
2525
import org.hibernate.metamodel.mapping.internal.ToOneAttributeMapping;
2626
import org.hibernate.query.spi.QueryOptions;
27+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
2728
import org.hibernate.reactive.loader.ast.spi.ReactiveSingleUniqueKeyEntityLoader;
2829
import org.hibernate.reactive.sql.exec.internal.StandardReactiveSelectExecutor;
2930
import org.hibernate.reactive.sql.results.spi.ReactiveListResultsConsumer;
3031
import org.hibernate.sql.ast.SqlAstTranslatorFactory;
3132
import org.hibernate.sql.ast.tree.select.SelectStatement;
3233
import org.hibernate.sql.exec.internal.BaseExecutionContext;
33-
import org.hibernate.sql.exec.internal.CallbackImpl;
3434
import org.hibernate.sql.exec.internal.JdbcParameterBindingsImpl;
3535
import org.hibernate.sql.exec.spi.Callback;
3636
import org.hibernate.sql.exec.spi.JdbcOperationQuerySelect;
@@ -179,7 +179,7 @@ public SingleUKEntityLoaderExecutionContext(SharedSessionContractImplementor ses
179179
super( session );
180180
//Careful, readOnly is possibly null
181181
this.queryOptions = readOnly == null ? QueryOptions.NONE : readOnly ? QueryOptions.READ_ONLY : QueryOptions.READ_WRITE;
182-
callback = new CallbackImpl();
182+
callback = new ReactiveCallbackImpl();
183183
}
184184

185185
@Override

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/loader/ast/spi/ReactiveAfterLoadAction.java

+21
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
package org.hibernate.reactive.loader.ast.spi;
2+
3+
import java.util.concurrent.CompletionStage;
4+
5+
import org.hibernate.engine.spi.SharedSessionContractImplementor;
6+
import org.hibernate.metamodel.mapping.EntityMappingType;
7+
8+
/**
9+
* Reactive version of {@link org.hibernate.loader.ast.spi.AfterLoadAction}
10+
*/
11+
public interface ReactiveAfterLoadAction {
12+
/**
13+
* @see org.hibernate.loader.ast.spi.AfterLoadAction#afterLoad(Object, EntityMappingType, SharedSessionContractImplementor)
14+
*
15+
* The action trigger - the {@code entity} is being loaded
16+
*/
17+
CompletionStage<Void> reactiveAfterLoad(
18+
Object entity,
19+
EntityMappingType entityMappingType,
20+
SharedSessionContractImplementor session);
21+
}

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/query/spi/ReactiveAbstractSelectionQuery.java

+11
Original file line numberDiff line numberDiff line change
@@ -30,12 +30,14 @@
3030
import org.hibernate.query.sqm.internal.SqmInterpretationsKey.InterpretationsKeySource;
3131
import org.hibernate.query.sqm.tree.SqmStatement;
3232
import org.hibernate.query.sqm.tree.select.SqmSelectStatement;
33+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
3334
import org.hibernate.reactive.logging.impl.Log;
3435
import org.hibernate.reactive.logging.impl.LoggerFactory;
3536
import org.hibernate.reactive.query.sqm.internal.AggregatedSelectReactiveQueryPlan;
3637
import org.hibernate.reactive.query.sqm.internal.ConcreteSqmSelectReactiveQueryPlan;
3738
import org.hibernate.reactive.query.sqm.spi.ReactiveSelectQueryPlan;
3839
import org.hibernate.reactive.sql.results.spi.ReactiveSingleResultConsumer;
40+
import org.hibernate.sql.exec.spi.Callback;
3941
import org.hibernate.sql.results.internal.TupleMetadata;
4042

4143
import jakarta.persistence.NoResultException;
@@ -76,6 +78,8 @@ public class ReactiveAbstractSelectionQuery<R> {
7678
private final Function<List<R>, R> uniqueElement;
7779
private final InterpretationsKeySource interpretationsKeySource;
7880

81+
private Callback callback;
82+
7983
// I'm sure we can avoid some of this by making some methods public in ORM,
8084
// but this allows me to prototype faster. We can refactor the code later.
8185
public ReactiveAbstractSelectionQuery(
@@ -363,4 +367,11 @@ public void enableFetchProfile(String profileName) {
363367
}
364368
fetchProfiles.add( profileName );
365369
}
370+
371+
public Callback getCallback() {
372+
if ( callback == null ) {
373+
callback = new ReactiveCallbackImpl();
374+
}
375+
return callback;
376+
}
366377
}

‎hibernate-reactive-core/src/main/java/org/hibernate/reactive/query/sql/internal/ReactiveNativeQueryImpl.java

+6
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,7 @@
4646
import org.hibernate.reactive.query.sql.spi.ReactiveNativeQueryImplementor;
4747
import org.hibernate.reactive.query.sql.spi.ReactiveNonSelectQueryPlan;
4848
import org.hibernate.reactive.query.sqm.spi.ReactiveSelectQueryPlan;
49+
import org.hibernate.sql.exec.spi.Callback;
4950
import org.hibernate.type.BasicTypeReference;
5051

5152
import jakarta.persistence.AttributeConverter;
@@ -192,6 +193,11 @@ public R getSingleResultOrNull() {
192193
return selectionQueryDelegate.getSingleResultOrNull();
193194
}
194195

196+
@Override
197+
public Callback getCallback() {
198+
return selectionQueryDelegate.getCallback();
199+
}
200+
195201
@Override
196202
public CompletionStage<R> getReactiveSingleResultOrNull() {
197203
return selectionQueryDelegate.getReactiveSingleResultOrNull();

0 commit comments

Comments
 (0)