Polish things up a bit

Author: christophstrobl

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

@@ -21,6 +21,7 @@
 import reactor.core.publisher.Mono;
 
 import java.io.Serializable;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -47,7 +48,6 @@
 import org.springframework.data.mongodb.repository.query.MongoEntityInformation;
 import org.springframework.data.repository.query.FluentQuery;
 import org.springframework.data.util.StreamUtils;
-import org.springframework.data.util.Streamable;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -110,10 +110,9 @@ public <S extends T> Flux<S> saveAll(Iterable<S> entities) {
 
 		Assert.notNull(entities, "The given Iterable of entities must not be null");
 
-		Streamable<S> source = Streamable.of(entities);
-
+		List<S> source = toList(entities);
 		return source.stream().allMatch(entityInformation::isNew) ? //
-				insert(entities) : doItSomewhatSequentially(source, this::save);
+				insert(source) : concatMapSequentially(source, this::save);
 	}
 
 	@Override
@@ -126,20 +125,6 @@ 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.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) {
 
@@ -349,8 +334,11 @@ 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.insert(source, entityInformation.getCollectionName());
+		return insert(toCollection(entities));
+	}
+
+	private <S extends T> Flux<S> insert(Collection<S> entities) {
+		return entities.isEmpty() ? Flux.empty() : mongoOperations.insert(entities, entityInformation.getCollectionName());
 	}
 
 	@Override
@@ -453,6 +441,12 @@ void setRepositoryMethodMetadata(CrudMethodMetadata crudMethodMetadata) {
 		this.crudMethodMetadata = crudMethodMetadata;
 	}
 
+	private Flux<T> findAll(Query query) {
+
+		getReadPreference().ifPresent(query::withReadPreference);
+		return mongoOperations.find(query, entityInformation.getJavaType(), entityInformation.getCollectionName());
+	}
+
 	private Optional<ReadPreference> getReadPreference() {
 
 		if (crudMethodMetadata == null) {
@@ -474,15 +468,47 @@ private Query getIdQuery(Iterable<? extends ID> ids) {
 		return new Query(where(entityInformation.getIdAttribute()).in(toCollection(ids)));
 	}
 
-	private static <E> Collection<E> toCollection(Iterable<E> ids) {
-		return ids instanceof Collection<E> collection ? collection
-				: StreamUtils.createStreamFromIterator(ids.iterator()).collect(Collectors.toList());
+	/**
+	 * Transform the elements emitted by this Flux into Publishers, then flatten these inner publishers into a single
+	 * Flux. The operation does not allow interleave between performing the map operation for the first and second source
+	 * element guaranteeing the mapping operation completed before subscribing to its following inners, that will then be
+	 * subscribed to eagerly emitting elements in order of their source.
+	 * 
+	 * <pre class="code">
+	 * Flux.just(first-element).flatMap(...)
+	 *     .concatWith(Flux.fromIterable(remaining-elements).flatMapSequential(...))
+	 * </pre>
+	 *
+	 * @param source the collection of elements to transform.
+	 * @param mapper the transformation {@link Function}. Must not be {@literal null}.
+	 * @return never {@literal null}.
+	 * @param <T> source type
+	 */
+	static <T> Flux<T> concatMapSequentially(List<T> source,
+			Function<? super T, ? extends Publisher<? extends T>> mapper) {
+
+		if (source.isEmpty()) {
+			return Flux.empty();
+		}
+		if (source.size() == 1) {
+			return Flux.just(source.iterator().next()).flatMap(mapper);
+		}
+		if (source.size() == 2) {
+			return Flux.fromIterable(source).concatMap(mapper);
+		}
+
+		Flux<T> first = Flux.just(source.get(0)).flatMap(mapper);
+		Flux<T> theRest = Flux.fromIterable(source.subList(1, source.size())).flatMapSequential(mapper);
+		return first.concatWith(theRest);
 	}
 
-	private Flux<T> findAll(Query query) {
+	private static <E> List<E> toList(Iterable<E> source) {
+		return source instanceof List<E> list ? list : new ArrayList<>(toCollection(source));
+	}
 
-		getReadPreference().ifPresent(query::withReadPreference);
-		return mongoOperations.find(query, entityInformation.getJavaType(), entityInformation.getCollectionName());
+	private static <E> Collection<E> toCollection(Iterable<E> source) {
+		return source instanceof Collection<E> collection ? collection
+				: StreamUtils.createStreamFromIterator(source.iterator()).collect(Collectors.toList());
 	}
 
 	/**