Skip to content

Commit 8879362

Browse files
dreab8dreab8
committed
[#1905] Reactive find with lock in Quarkus with reactive hibernate
1 parent 2da4c8e commit 8879362

15 files changed

+288
-26
lines changed

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

Lines changed: 79 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,18 +30,24 @@
3030
import org.hibernate.engine.spi.SessionImplementor;
3131
import org.hibernate.engine.spi.SharedSessionContractImplementor;
3232
import org.hibernate.engine.spi.Status;
33+
import org.hibernate.event.service.spi.EventListenerGroup;
34+
import org.hibernate.event.spi.PostLoadEvent;
35+
import org.hibernate.event.spi.PostLoadEventListener;
3336
import org.hibernate.persister.collection.CollectionPersister;
3437
import org.hibernate.persister.entity.EntityPersister;
38+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
3539
import org.hibernate.reactive.logging.impl.Log;
3640
import org.hibernate.reactive.persister.entity.impl.ReactiveEntityPersister;
3741
import org.hibernate.reactive.session.ReactiveSession;
42+
import org.hibernate.sql.exec.spi.Callback;
3843
import org.hibernate.sql.results.graph.entity.EntityInitializer;
3944
import org.hibernate.sql.results.jdbc.spi.JdbcValuesSourceProcessingState;
4045
import org.hibernate.sql.results.spi.LoadContexts;
4146

4247
import static java.lang.invoke.MethodHandles.lookup;
4348
import static org.hibernate.reactive.logging.impl.LoggerFactory.make;
4449
import static org.hibernate.reactive.util.impl.CompletionStages.completedFuture;
50+
import static org.hibernate.reactive.util.impl.CompletionStages.loop;
4551
import static org.hibernate.reactive.util.impl.CompletionStages.voidFuture;
4652

4753
/**
@@ -456,7 +462,7 @@ public EntityHolder removeEntityHolder(EntityKey key) {
456462

457463
@Override
458464
public void postLoad(JdbcValuesSourceProcessingState processingState, Consumer<EntityHolder> loadedConsumer) {
459-
delegate.postLoad( processingState, loadedConsumer );
465+
throw LOG.nonReactiveMethodCall( "reactivePostLoad(JdbcValuesSourceProcessingState, Consumer<EntityHolder>) )" );
460466
}
461467

462468
@Internal
@@ -710,4 +716,76 @@ public Iterator<Object> managedEntitiesIterator() {
710716
public NaturalIdResolutions getNaturalIdResolutions() {
711717
return delegate.getNaturalIdResolutions();
712718
}
719+
720+
/**
721+
* Reactive version of {@link StatefulPersistenceContext#postLoad(JdbcValuesSourceProcessingState, Consumer)}
722+
*
723+
*/
724+
public CompletionStage<Void> reactivePostLoad(JdbcValuesSourceProcessingState processingState, Consumer<EntityHolder> holderConsumer) {
725+
final ReactiveCallbackImpl callback = (ReactiveCallbackImpl) processingState.getExecutionContext().getCallback();
726+
727+
if ( processingState.getLoadingEntityHolders() != null ) {
728+
final EventListenerGroup<PostLoadEventListener> listenerGroup =
729+
getSession().getFactory().getEventListenerGroups().eventListenerGroup_POST_LOAD;
730+
final PostLoadEvent postLoadEvent = processingState.getPostLoadEvent();
731+
return loop(
732+
processingState.getLoadingEntityHolders(), entityHolder ->
733+
processLoadedEntityHolder(
734+
entityHolder,
735+
listenerGroup,
736+
postLoadEvent,
737+
callback,
738+
holderConsumer
739+
))
740+
.thenAccept( v -> processingState.getLoadingEntityHolders().clear() );
741+
}
742+
if ( processingState.getReloadedEntityHolders() != null ) {
743+
return loop(
744+
processingState.getLoadingEntityHolders(), entityHolder ->
745+
processLoadedEntityHolder(
746+
entityHolder,
747+
null,
748+
null,
749+
callback,
750+
holderConsumer
751+
))
752+
.thenAccept( v -> processingState.getLoadingEntityHolders().clear() );
753+
}
754+
return voidFuture();
755+
}
756+
757+
/**
758+
* Reactive version of {@link StatefulPersistenceContext#processLoadedEntityHolder(EntityHolder, EventListenerGroup, PostLoadEvent, Callback, Consumer)}
759+
*/
760+
private CompletionStage<Void> processLoadedEntityHolder(
761+
EntityHolder holder,
762+
EventListenerGroup<PostLoadEventListener> listenerGroup,
763+
PostLoadEvent postLoadEvent,
764+
ReactiveCallbackImpl callback,
765+
Consumer<EntityHolder> holderConsumer) {
766+
if ( holderConsumer != null ) {
767+
holderConsumer.accept( holder );
768+
}
769+
if ( holder.getEntity() == null ) {
770+
// It's possible that we tried to load an entity and found out it doesn't exist,
771+
// in which case we added an entry with a null proxy and entity.
772+
// Remove that empty entry on post load to avoid unwanted side effects
773+
getEntitiesByKey().remove( holder.getEntityKey() );
774+
}
775+
else {
776+
if ( postLoadEvent != null ) {
777+
postLoadEvent.reset();
778+
postLoadEvent.setEntity( holder.getEntity() )
779+
.setId( holder.getEntityKey().getIdentifier() )
780+
.setPersister( holder.getDescriptor() );
781+
listenerGroup.fireEventOnEachListener( postLoadEvent, PostLoadEventListener::onPostLoad );
782+
if ( callback != null ) {
783+
return callback
784+
.invokeReactiveLoadActions( holder.getEntity(), holder.getDescriptor(), getSession() )
785+
.thenAccept( v -> holder.resetEntityInitialier() );
786+
}
787+
}
788+
}
789+
return voidFuture();
790+
}
713791
}

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

Lines changed: 70 additions & 0 deletions
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/loader/ast/internal/ReactiveNaturalIdLoaderDelegate.java

Lines changed: 2 additions & 2 deletions
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

Lines changed: 14 additions & 10 deletions
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 <T> void invokeAfterLoadActions(Callback callback, SharedSessionContractImplementor session, T entity) {
85-
if ( entity != null && getLoadable() != null) {
86-
callback.invokeAfterLoadActions( entity, (EntityMappingType) getLoadable(), session );
87+
private <T> CompletionStage<Void> invokeAfterLoadActions(ReactiveCallbackImpl callback, SharedSessionContractImplementor session, T 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

Lines changed: 2 additions & 2 deletions
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

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
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.loader.ast.spi;
7+
8+
import java.util.concurrent.CompletionStage;
9+
10+
import org.hibernate.engine.spi.SharedSessionContractImplementor;
11+
import org.hibernate.metamodel.mapping.EntityMappingType;
12+
13+
/**
14+
* Reactive version of {@link org.hibernate.loader.ast.spi.AfterLoadAction}
15+
*/
16+
public interface ReactiveAfterLoadAction {
17+
/**
18+
* @see org.hibernate.loader.ast.spi.AfterLoadAction#afterLoad(Object, EntityMappingType, SharedSessionContractImplementor)
19+
*
20+
* The action trigger - the {@code entity} is being loaded
21+
*/
22+
CompletionStage<Void> reactiveAfterLoad(
23+
Object entity,
24+
EntityMappingType entityMappingType,
25+
SharedSessionContractImplementor session);
26+
}

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

Lines changed: 15 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,10 @@
66
package org.hibernate.reactive.query.spi;
77

88
import java.lang.invoke.MethodHandles;
9-
import java.util.*;
9+
import java.util.HashSet;
10+
import java.util.List;
11+
import java.util.Optional;
12+
import java.util.Set;
1013
import java.util.concurrent.CompletionException;
1114
import java.util.concurrent.CompletionStage;
1215
import java.util.function.Consumer;
@@ -30,12 +33,14 @@
3033
import org.hibernate.query.sqm.internal.SqmInterpretationsKey.InterpretationsKeySource;
3134
import org.hibernate.query.sqm.tree.SqmStatement;
3235
import org.hibernate.query.sqm.tree.select.SqmSelectStatement;
36+
import org.hibernate.reactive.engine.impl.ReactiveCallbackImpl;
3337
import org.hibernate.reactive.logging.impl.Log;
3438
import org.hibernate.reactive.logging.impl.LoggerFactory;
3539
import org.hibernate.reactive.query.sqm.internal.AggregatedSelectReactiveQueryPlan;
3640
import org.hibernate.reactive.query.sqm.internal.ConcreteSqmSelectReactiveQueryPlan;
3741
import org.hibernate.reactive.query.sqm.spi.ReactiveSelectQueryPlan;
3842
import org.hibernate.reactive.sql.results.spi.ReactiveSingleResultConsumer;
43+
import org.hibernate.sql.exec.spi.Callback;
3944
import org.hibernate.sql.results.internal.TupleMetadata;
4045

4146
import jakarta.persistence.NoResultException;
@@ -76,6 +81,8 @@ public class ReactiveAbstractSelectionQuery<R> {
7681
private final Function<List<R>, R> uniqueElement;
7782
private final InterpretationsKeySource interpretationsKeySource;
7883

84+
private Callback callback;
85+
7986
// I'm sure we can avoid some of this by making some methods public in ORM,
8087
// but this allows me to prototype faster. We can refactor the code later.
8188
public ReactiveAbstractSelectionQuery(
@@ -363,4 +370,11 @@ public void enableFetchProfile(String profileName) {
363370
}
364371
fetchProfiles.add( profileName );
365372
}
373+
374+
public Callback getCallback() {
375+
if ( callback == null ) {
376+
callback = new ReactiveCallbackImpl();
377+
}
378+
return callback;
379+
}
366380
}

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

Lines changed: 6 additions & 0 deletions
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)