Use more efficient Reactor operators

Use handle/flatMapIterable instead of flatMap/flatMapMany
when possible.

Closes gh-22727

Author: sdeleuze

spring-web/src/main/java/org/springframework/http/codec/json/AbstractJackson2Decoder.java

@@ -115,7 +115,7 @@ private Flux<Object> decodeInternal(Flux<TokenBuffer> tokens, ResolvableType ele
 				getObjectMapper().readerWithView(jsonView).forType(javaType) :
 				getObjectMapper().readerFor(javaType));
 
-		return tokens.flatMap(tokenBuffer -> {
+		return tokens.handle((tokenBuffer, sink) -> {
 			try {
 				Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
 				if (!Hints.isLoggingSuppressed(hints)) {
@@ -124,16 +124,18 @@ private Flux<Object> decodeInternal(Flux<TokenBuffer> tokens, ResolvableType ele
 						return Hints.getLogPrefix(hints) + "Decoded [" + formatted + "]";
 					});
 				}
-				return Mono.justOrEmpty(value);
+				if (value != null) {
+					sink.next(value);
+				}
 			}
 			catch (InvalidDefinitionException ex) {
-				return Mono.error(new CodecException("Type definition error: " + ex.getType(), ex));
+				sink.error(new CodecException("Type definition error: " + ex.getType(), ex));
 			}
 			catch (JsonProcessingException ex) {
-				return Mono.error(new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex));
+				sink.error(new DecodingException("JSON decoding error: " + ex.getOriginalMessage(), ex));
 			}
 			catch (IOException ex) {
-				return Mono.error(new DecodingException("I/O error while parsing input stream", ex));
+				sink.error(new DecodingException("I/O error while parsing input stream", ex));
 			}
 		});
 	}

spring-web/src/main/java/org/springframework/http/codec/xml/Jaxb2XmlDecoder.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiConsumer;
 import java.util.function.Function;
 import javax.xml.XMLConstants;
 import javax.xml.bind.JAXBElement;
@@ -35,6 +36,7 @@
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.SynchronousSink;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.core.codec.AbstractDecoder;
@@ -224,11 +226,11 @@ else if (outputClass.isAnnotationPresent(XmlType.class)) {
 	 * </ol>
 	 */
 	Flux<List<XMLEvent>> split(Flux<XMLEvent> xmlEventFlux, QName desiredName) {
-		return xmlEventFlux.flatMap(new SplitFunction(desiredName));
+		return xmlEventFlux.handle(new SplitHandler(desiredName));
 	}
 
 
-	private static class SplitFunction implements Function<XMLEvent, Publisher<? extends List<XMLEvent>>> {
+	private static class SplitHandler implements BiConsumer<XMLEvent, SynchronousSink<List<XMLEvent>>> {
 
 		private final QName desiredName;
 
@@ -239,12 +241,12 @@ private static class SplitFunction implements Function<XMLEvent, Publisher<? ext
 
 		private int barrier = Integer.MAX_VALUE;
 
-		public SplitFunction(QName desiredName) {
+		public SplitHandler(QName desiredName) {
 			this.desiredName = desiredName;
 		}
 
 		@Override
-		public Publisher<? extends List<XMLEvent>> apply(XMLEvent event) {
+		public void accept(XMLEvent event, SynchronousSink<List<XMLEvent>> sink) {
 			if (event.isStartElement()) {
 				if (this.barrier == Integer.MAX_VALUE) {
 					QName startElementName = event.asStartElement().getName();
@@ -264,10 +266,9 @@ public Publisher<? extends List<XMLEvent>> apply(XMLEvent event) {
 				if (this.elementDepth == this.barrier) {
 					this.barrier = Integer.MAX_VALUE;
 					Assert.state(this.events != null, "No XMLEvent List");
-					return Mono.just(this.events);
+					sink.next(this.events);
 				}
 			}
-			return Mono.empty();
 		}
 	}
 

spring-web/src/main/java/org/springframework/http/codec/xml/XmlEventDecoder.java

@@ -18,7 +18,6 @@
 
 import java.io.InputStream;
 import java.util.ArrayList;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
@@ -33,6 +32,7 @@
 import com.fasterxml.aalto.evt.EventAllocatorImpl;
 import com.fasterxml.aalto.stax.InputFactoryImpl;
 import org.reactivestreams.Publisher;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -106,18 +106,17 @@ public Flux<XMLEvent> decode(Publisher<DataBuffer> inputStream, ResolvableType e
 		}
 		else {
 			Mono<DataBuffer> singleBuffer = DataBufferUtils.join(flux);
-			return singleBuffer.
-					flatMapMany(dataBuffer -> {
-						try {
-							InputStream is = dataBuffer.asInputStream();
-							Iterator eventReader = inputFactory.createXMLEventReader(is);
-							return Flux.fromIterable((Iterable<XMLEvent>) () -> eventReader)
-									.doFinally(t -> DataBufferUtils.release(dataBuffer));
-						}
-						catch (XMLStreamException ex) {
-							return Mono.error(ex);
-						}
-					});
+			return singleBuffer.flatMapIterable(dataBuffer -> {
+				InputStream is = dataBuffer.asInputStream();
+				return () -> {
+					try {
+						return inputFactory.createXMLEventReader(is);
+					}
+					catch (XMLStreamException ex) {
+						throw Exceptions.propagate(ex);
+					}
+				};
+			});
 		}
 	}
 

spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java

@@ -441,13 +441,13 @@ public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> bodyType) {
 		@Override
 		public <T> Flux<T> bodyToFlux(Class<T> elementType) {
 			return this.responseMono.flatMapMany(response ->
-					handleBody(response, response.bodyToFlux(elementType), mono -> mono.flatMapMany(Flux::error)));
+					handleBody(response, response.bodyToFlux(elementType), mono -> mono.handle((t, sink) -> sink.error(t))));
 		}
 
 		@Override
 		public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> elementType) {
 			return this.responseMono.flatMapMany(response ->
-					handleBody(response, response.bodyToFlux(elementType), mono -> mono.flatMapMany(Flux::error)));
+					handleBody(response, response.bodyToFlux(elementType), mono -> mono.handle((t, sink) -> sink.error(t))));
 		}
 
 		private <T extends Publisher<?>> T handleBody(ClientResponse response,

spring-webflux/src/main/java/org/springframework/web/reactive/result/method/annotation/RequestPartMethodArgumentResolver.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package org.springframework.web.reactive.result.method.annotation;
 
+import java.util.Collections;
 import java.util.List;
 
 import reactor.core.publisher.Flux;
@@ -71,12 +72,15 @@ public Mono<Object> resolveArgument(
 		String name = getPartName(parameter, requestPart);
 
 		Flux<Part> parts = exchange.getMultipartData()
-				.flatMapMany(map -> {
+				.flatMapIterable(map -> {
 					List<Part> list = map.get(name);
 					if (CollectionUtils.isEmpty(list)) {
-						return (isRequired ? Flux.error(getMissingPartException(name, parameter)) : Flux.empty());
+						if (isRequired) {
+							throw getMissingPartException(name, parameter);
+						}
+						return Collections.emptyList();
 					}
-					return Flux.fromIterable(list);
+					return list;
 				});
 
 		if (Part.class.isAssignableFrom(parameter.getParameterType())) {