refactor: Avoid concurrency issues during some writes and reads.

Author: michael-simons

src/main/java/org/springframework/data/neo4j/core/ReactiveNeo4jTemplate.java

@@ -395,7 +395,7 @@ <T, R> Flux<R> doSave(Iterable<R> instances, Class<T> domainType) {
 		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")
@@ -525,7 +525,7 @@ public <T, R> Flux<R> saveAllAs(Iterable<T> instances, Class<R> resultType) {
 		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));
@@ -586,7 +586,7 @@ private <T> Flux<T> saveAllImpl(Iterable<T> instances, @Nullable Collection<Prop
 							.all()
 							.collectMap(m -> (Value) m.getT1(), m -> (Long) 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));
@@ -713,7 +713,7 @@ private Mono<NodesAndRelationshipsByIdStatementProvider> createNodesAndRelations
 				Set<Long> processedRelationshipIds = ctx.get("processedRelationships");
 				Set<Long> 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())
@@ -781,7 +781,7 @@ private Flux<Tuple2<Collection<Long>, Collection<Long>>> iterateNextLevel(Collec
 							return queryFragments.includeField(prepend);
 						}
 				))
-			.flatMap(relDe -> {
+			.concatMap(relDe -> {
 				Node node = anyNode(Constants.NAME_OF_TYPED_ROOT_NODE.apply(target));
 
 				Statement statement = cypherGenerator
@@ -1193,7 +1193,7 @@ public Flux<T> getResults() {
 
 			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;
 			});

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

@@ -85,7 +85,7 @@ public Flux<T> findAllById(Iterable<ID> ids) {
 
 	@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 <S extends T> Flux<S> saveAll(Iterable<S> entities) {
 	@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 Mono<Void> deleteAll(Iterable<? extends T> entities) {
 	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();
 	}
 
 	/*