Ensure subscription order on multi document operations.

Author: christophstrobl

spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/ReactiveMongoTemplate.java

@@ -1413,7 +1413,7 @@ protected <T> Flux<T> doInsertAll(Collection<? extends T> listToSave, MongoWrite
 		});
 
 		return Flux.fromIterable(elementsByCollection.keySet())
-				.flatMap(collectionName -> doInsertBatch(collectionName, elementsByCollection.get(collectionName), writer));
+				.concatMap(collectionName -> doInsertBatch(collectionName, elementsByCollection.get(collectionName), writer));
 	}
 
 	protected <T> Flux<T> doInsertBatch(String collectionName, Collection<? extends T> batchToSave,

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

@@ -113,16 +113,16 @@ public <S extends T> Flux<S> saveAll(Iterable<S> entities) {
 		Streamable<S> source = Streamable.of(entities);
 
 		return source.stream().allMatch(entityInformation::isNew) ? //
-				mongoOperations.insert(source.stream().collect(Collectors.toList()), entityInformation.getCollectionName()) : //
-				Flux.fromIterable(entities).flatMapSequential(this::save);
+			    insert(entities) :
+				Flux.fromIterable(entities).concatMap(this::save);
 	}
 
 	@Override
 	public <S extends T> Flux<S> saveAll(Publisher<S> entityStream) {
 
 		Assert.notNull(entityStream, "The given Publisher of entities must not be null");
 
-		return Flux.from(entityStream).flatMapSequential(entity -> entityInformation.isNew(entity) ? //
+		return Flux.from(entityStream).concatMap(entity -> entityInformation.isNew(entity) ? //
 				mongoOperations.insert(entity, entityInformation.getCollectionName()) : //
 				mongoOperations.save(entity, entityInformation.getCollectionName()));
 	}
@@ -296,7 +296,7 @@ public Mono<Void> deleteAll(Publisher<? extends T> entityStream) {
 		Optional<ReadPreference> readPreference = getReadPreference();
 		return Flux.from(entityStream)//
 				.map(entityInformation::getRequiredId)//
-				.flatMap(id -> deleteById(id, readPreference))//
+				.concatMap(id -> deleteById(id, readPreference))//
 				.then();
 	}
 
@@ -337,17 +337,15 @@ public <S extends T> Flux<S> insert(Iterable<S> entities) {
 		Assert.notNull(entities, "The given Iterable of entities must not be null");
 
 		Collection<S> source = toCollection(entities);
-
-		return source.isEmpty() ? Flux.empty() : mongoOperations.insertAll(source);
+		return source.isEmpty() ? Flux.empty() : mongoOperations.insert(source, entityInformation.getCollectionName());
 	}
 
 	@Override
 	public <S extends T> Flux<S> insert(Publisher<S> entities) {
 
 		Assert.notNull(entities, "The given Publisher of entities must not be null");
 
-		return Flux.from(entities)
-				.flatMapSequential(entity -> mongoOperations.insert(entity, entityInformation.getCollectionName()));
+		return Flux.from(entities).concatMap(this::insert);
 	}
 
 	// -------------------------------------------------------------------------