Spring/spring-data-neo4j
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor: Avoid concurrency issues during some writes and reads.

Browse Source
This commit is contained in:
Michael Simons
2023-10-24 10:26:00 +02:00
parent b3dc32c74c
commit e4f034858a
2 changed files with 9 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
src/main/java/org/springframework/data/neo4j/core/ReactiveNeo4jTemplate.java
View File

@@ -401,7 +401,7 @@ public final class ReactiveNeo4jTemplate implements
NestedRelationshipProcessingStateMachine stateMachine = new NestedRelationshipProcessingStateMachine(neo4jMappingContext);
EntityFromDtoInstantiatingConverter<T> converter = new EntityFromDtoInstantiatingConverter<>(domainType, neo4jMappingContext);
return Flux.fromIterable(instances)
.flatMap(instance -> {
.concatMap(instance -> {
T domainObject = converter.convert(instance);
@SuppressWarnings("unchecked")
@@ -534,7 +534,7 @@ public final class ReactiveNeo4jTemplate implements
Neo4jPersistentEntity<?> entityMetaData = neo4jMappingContext.getRequiredPersistentEntity(commonElementType);
Neo4jPersistentProperty idProperty = entityMetaData.getIdProperty();
return savedInstances.flatMap(savedInstance -> {
return savedInstances.concatMap(savedInstance -> {
PersistentPropertyAccessor<T> propertyAccessor = entityMetaData.getPropertyAccessor(savedInstance);
return findById(propertyAccessor.getProperty(idProperty), commonElementType);
}).map(instance -> localProjectionFactory.createProjection(resultType, instance));
@@ -595,7 +595,7 @@ public final class ReactiveNeo4jTemplate implements
.all()
.collectMap(m -> (Value) m.getT1(), m -> (String) m.getT2());
}).flatMapMany(idToInternalIdMapping -> Flux.fromIterable(entitiesToBeSaved)
.flatMap(t -> {
.concatMap(t -> {
PersistentPropertyAccessor<T> propertyAccessor = entityMetaData.getPropertyAccessor(t.getT3());
Neo4jPersistentProperty idProperty = entityMetaData.getRequiredIdProperty();
Object id = convertIdValues(idProperty, propertyAccessor.getProperty(idProperty));
@@ -722,7 +722,7 @@ public final class ReactiveNeo4jTemplate implements
Set<String> processedRelationshipIds = ctx.get("processedRelationships");
Set<String> processedNodeIds = ctx.get("processedNodes");
return Flux.fromIterable(entityMetaData.getRelationshipsInHierarchy(queryFragments::includeField))
.flatMap(relationshipDescription -> {
.concatMap(relationshipDescription -> {
Statement statement = cypherGenerator.prepareMatchOf(entityMetaData, relationshipDescription,
queryFragments.getMatchOn(), queryFragments.getCondition())
@@ -777,7 +777,7 @@ public final class ReactiveNeo4jTemplate implements
return queryFragments.includeField(prepend);
}
))
.flatMap(relDe -> {
.concatMap(relDe -> {
Node node = anyNode(Constants.NAME_OF_TYPED_ROOT_NODE.apply(target));
Statement statement = cypherGenerator
@@ -1188,7 +1188,7 @@ public final class ReactiveNeo4jTemplate implements
return fetchSpec.all().switchOnFirst((signal, f) -> {
if (signal.hasValue() && preparedQuery.resultsHaveBeenAggregated()) {
return f.flatMap(nested -> Flux.fromIterable((Collection<T>) nested).distinct()).distinct();
return f.concatMap(nested -> Flux.fromIterable((Collection<T>) nested).distinct()).distinct();
}
return f;
});

6
src/main/java/org/springframework/data/neo4j/repository/support/SimpleReactiveNeo4jRepository.java
View File

@@ -85,7 +85,7 @@ public class SimpleReactiveNeo4jRepository<T, ID> implements ReactiveSortingRepo
@Override
public Flux<T> findAllById(Publisher<ID> idStream) {
return Flux.from(idStream).buffer().flatMap(this::findAllById);
return Flux.from(idStream).buffer().concatMap(this::findAllById);
}
@Override
@@ -135,7 +135,7 @@ public class SimpleReactiveNeo4jRepository<T, ID> implements ReactiveSortingRepo
@Transactional
public <S extends T> Flux<S> saveAll(Publisher<S> entityStream) {
return Flux.from(entityStream).flatMap(this::save);
return Flux.from(entityStream).concatMap(this::save);
}
/*
@@ -218,7 +218,7 @@ public class SimpleReactiveNeo4jRepository<T, ID> implements ReactiveSortingRepo
public Mono<Void> deleteAll(Publisher<? extends T> entitiesPublisher) {
Assert.notNull(entitiesPublisher, "The given Publisher of entities must not be null");
return Flux.from(entitiesPublisher).flatMap(this::delete).then();
return Flux.from(entitiesPublisher).concatMap(this::delete).then();
}
/*