Update reactive saveAll flow.

pom.xml

@@ -5,7 +5,7 @@
 
 	<groupId>org.springframework.data</groupId>
 	<artifactId>spring-data-mongodb-parent</artifactId>
-	<version>4.5.0-SNAPSHOT</version>
+	<version>4.5.x-GH-4838-SNAPSHOT</version>
 	<packaging>pom</packaging>
 
 	<name>Spring Data MongoDB</name>

spring-data-mongodb-distribution/pom.xml

@@ -15,7 +15,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.5.0-SNAPSHOT</version>
+		<version>4.5.x-GH-4838-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

spring-data-mongodb/pom.xml

@@ -13,7 +13,7 @@
 	<parent>
 		<groupId>org.springframework.data</groupId>
 		<artifactId>spring-data-mongodb-parent</artifactId>
-		<version>4.5.0-SNAPSHOT</version>
+		<version>4.5.x-GH-4838-SNAPSHOT</version>
 		<relativePath>../pom.xml</relativePath>
 	</parent>
 

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

@@ -17,6 +17,7 @@
 
 import static org.springframework.data.mongodb.core.query.Criteria.*;
 
+import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -113,8 +114,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) : new AeonFlux<>(source).combatMap(this::save);
 	}
 
 	@Override
@@ -566,4 +566,46 @@ private ReactiveFindOperation.TerminatingFind<T> createQuery(UnaryOperator<Query
 		}
 
 	}
+
+	static class AeonFlux<T> extends Flux<T> {
+
+		private final Streamable<T> source;
+		private final Flux<T> delegate;
+
+		AeonFlux(Streamable<T> source) {
+			this(source, Flux.fromIterable(source));
+		}
+
+		private AeonFlux(Streamable<T> source, Flux<T> delegate) {
+			this.source = source;
+			this.delegate = delegate;
+		}
+
+		@Override
+		public void subscribe(CoreSubscriber<? super T> actual) {
+			delegate.subscribe(actual);
+		}
+
+		Flux<T> combatMap(Function<? super T, ? extends Publisher<? extends T>> mapper) {
+			return new AeonFlux<>(source, combatMapList(source.toList(), mapper));
+		}
+
+		private static <T> Flux<T> combatMapList(List<T> list,
+				Function<? super T, ? extends Publisher<? extends T>> mapper) {
+
+			if (list.isEmpty()) {
+				return Flux.empty();
+			}
+			if (list.size() == 1) {
+				return Flux.just(list.iterator().next()).flatMap(mapper);
+			}
+			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);
+		}
+	}
 }

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) //
+			.expectNext(oliver, dave, carter, boyd, stefan, leroi, alicia)
+			.verifyComplete();
+	}
+
 	@Test // GH-3609
 	void savePublisherOfImmutableEntitiesShouldInsertEntity() {