hacking

Author: christophstrobl

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/repository/support/SimpleReactiveMongoRepository.java

@@ -113,8 +113,7 @@ public <S extends T> Flux<S> saveAll(Iterable<S> entities) {
 		Streamable<S> source = Streamable.of(entities);
 
 		return source.stream().allMatch(entityInformation::isNew) ? //
-			    insert(entities) :
-				Flux.fromIterable(entities).concatMap(this::save);
+				insert(entities) : doItSomewhatSequentially(source, this::save);
 	}
 
 	@Override
@@ -127,6 +126,20 @@ public <S extends T> Flux<S> saveAll(Publisher<S> entityStream) {
 				mongoOperations.save(entity, entityInformation.getCollectionName()));
 	}
 
+	static <T> Flux<T> doItSomewhatSequentially/* how should we actually call this? */(Streamable<T> ts, Function<? super T, ? extends Publisher<? extends T>> mapper) {
+
+		List<T> list = ts.stream().toList();
+		if (list.size() == 1) {
+			return Flux.just(list.iterator().next()).flatMap(mapper);
+		} else if (list.size() == 2) {
+			return Flux.fromIterable(list).concatMap(mapper);
+		}
+
+		Flux<T> first = Flux.just(list.get(0)).flatMap(mapper);
+		Flux<T> theRest = Flux.fromIterable(list.subList(1, list.size())).flatMapSequential(mapper);
+		return first.concatWith(theRest);
+	}
+
 	@Override
 	public Mono<T> findById(ID id) {
 

spring-data-mongodb/src/test/java/org/springframework/data/mongodb/repository/SimpleReactiveMongoRepositoryTests.java

@@ -18,6 +18,10 @@
 import static org.assertj.core.api.Assertions.*;
 import static org.springframework.data.domain.ExampleMatcher.*;
 
+import org.junit.jupiter.api.RepeatedTest;
+import org.springframework.data.mongodb.ReactiveMongoTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
@@ -333,6 +337,18 @@ void savePublisherOfEntitiesShouldInsertEntity() {
 		assertThat(boyd.getId()).isNotNull();
 	}
 
+	@RepeatedTest(10)
+	void transactionalSaveAllForStuffThatIsConsideredAnUpdateOfExistingData() {
+
+		ReactiveMongoTransactionManager txmgr = new ReactiveMongoTransactionManager(template.getMongoDatabaseFactory());
+		TransactionalOperator.create(txmgr, TransactionDefinition.withDefaults()).execute(callback -> {
+			return repository.saveAll(Arrays.asList(oliver, dave, carter, boyd, stefan, leroi, alicia));
+		})
+		.as(StepVerifier::create) //
+			.expectNextCount(7) //
+			.verifyComplete();
+	}
+
 	@Test // GH-3609
 	void savePublisherOfImmutableEntitiesShouldInsertEntity() {