Polish

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/ErrorHandlingHttpHandler.java

@@ -44,23 +44,23 @@ public class ErrorHandlingHttpHandler extends HttpHandlerDecorator {
 
 	@Override
 	public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
-		Mono<Void> publisher;
+		Mono<Void> mono;
 		try {
-			publisher = getDelegate().handle(request, response);
+			mono = getDelegate().handle(request, response);
 		}
 		catch (Throwable ex) {
-			publisher = Mono.error(ex);
+			mono = Mono.error(ex);
 		}
 		for (HttpExceptionHandler handler : this.exceptionHandlers) {
-			publisher = applyExceptionHandler(publisher, handler, request, response);
+			mono = applyExceptionHandler(mono, handler, request, response);
 		}
-		return publisher;
+		return mono;
 	}
 
-	private static Mono<Void> applyExceptionHandler(Mono<Void> publisher,
-			HttpExceptionHandler handler, ServerHttpRequest request, ServerHttpResponse response) {
+	private static Mono<Void> applyExceptionHandler(Mono<Void> mono, HttpExceptionHandler handler,
+			ServerHttpRequest request, ServerHttpResponse response) {
 
-		return publisher.flux().onErrorResumeWith(ex -> handler.handle(request, response, ex)).after();
+		return mono.otherwise(ex -> handler.handle(request, response, ex)).after();
 	}
 
 }

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/HttpExceptionHandler.java

@@ -36,7 +36,7 @@ public interface HttpExceptionHandler {
 	 * @param request the current request
 	 * @param response the current response
 	 * @param ex the exception to handle
-	 * @return Publisher to indicate when exception handling is complete.
+	 * @return {@code Mono<Void>} to indicate when exception handling is complete.
 	 */
 	Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response, Throwable ex);
 

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/HttpFilter.java

@@ -39,7 +39,7 @@ public interface HttpFilter {
 	 * @param request current HTTP request.
 	 * @param response current HTTP response.
 	 * @param chain provides a way to delegate to the next HttpFilter.
-	 * @return Publisher to indicate when request processing is complete.
+	 * @return {@code Mono<Void>} to indicate when request processing is complete.
 	 */
 	Mono<Void> filter(ServerHttpRequest request, ServerHttpResponse response,
 			HttpFilterChain chain);

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/HttpFilterChain.java

@@ -29,7 +29,7 @@ public interface HttpFilterChain {
 	 *
 	 * @param request current HTTP request.
 	 * @param response current HTTP response.
-	 * @return Publisher to indicate when request handling is complete.
+	 * @return {@code Mono<Void>} to indicate when request handling is complete.
 	 */
 	Mono<Void> filter(ServerHttpRequest request, ServerHttpResponse response);
 

spring-web-reactive/src/main/java/org/springframework/http/server/reactive/HttpHandler.java

@@ -33,7 +33,7 @@ public interface HttpHandler {
 	 *
 	 * @param request current HTTP request.
 	 * @param response current HTTP response.
-	 * @return Publisher to indicate when request handling is complete.
+	 * @return {@code Mono<Void>} to indicate when request handling is complete.
 	 */
 	Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response);
 

spring-web-reactive/src/main/java/org/springframework/web/reactive/DispatcherHandler.java

@@ -121,14 +121,13 @@ public class DispatcherHandler implements HttpHandler, ApplicationContextAware {
 				.next()
 				.then(handler -> getHandlerAdapter(handler).handle(request, response, handler))
 				.then(result -> {
-					Mono<Void> publisher = (result.hasError() ? Mono.error(result.getError()) :
-							getResultHandler(result).handleResult(request, response, result));
+					Mono<Void> mono = (result.hasError() ? Mono.error(result.getError()) :
+							handleResult(request, response, result));
 					if (result.hasExceptionMapper()) {
-						return publisher
-								.otherwise(ex -> result.getExceptionMapper().apply(ex)
-								.then(errorResult -> getResultHandler(errorResult).handleResult(request, response, errorResult)));
+						return mono.otherwise(ex -> result.getExceptionMapper().apply(ex)
+								.then(exResult -> handleResult(request, response, exResult)));
 					}
-					return publisher;
+					return mono;
 				})
 				.otherwise(ex -> Mono.error(this.errorMapper.apply(ex)));
 	}
@@ -142,6 +141,10 @@ public class DispatcherHandler implements HttpHandler, ApplicationContextAware {
 		throw new IllegalStateException("No HandlerAdapter: " + handler);
 	}
 
+	protected Mono<Void> handleResult(ServerHttpRequest request, ServerHttpResponse response, HandlerResult result) {
+		return getResultHandler(result).handleResult(request, response, result);
+	}
+
 	protected HandlerResultHandler getResultHandler(HandlerResult handlerResult) {
 		for (HandlerResultHandler resultHandler : resultHandlers) {
 			if (resultHandler.supports(handlerResult)) {

spring-web-reactive/src/main/java/org/springframework/web/reactive/HandlerAdapter.java

@@ -53,7 +53,7 @@ public interface HandlerAdapter {
 	 * @param handler handler to use. This object must have previously been passed
 	 * to the {@code supports} method of this interface, which must have
 	 * returned {@code true}.
-	 * @return A {@link Publisher} object that produces a single {@link HandlerResult} element
+	 * @return A {@link Mono} that emits a single {@link HandlerResult} element
 	 */
 	Mono<HandlerResult> handle(ServerHttpRequest request, ServerHttpResponse response,
 			Object handler);

spring-web-reactive/src/main/java/org/springframework/web/reactive/HandlerMapping.java

@@ -32,7 +32,8 @@ public interface HandlerMapping {
 	/**
 	 * Return a handler for this request.
 	 * @param request current HTTP request
-	 * @return A {@link Mono} object that produces a single handler element
+	 * @return A {@link Mono} that emits one value or none in case the request
+	 * cannot be resolved to a handler.
 	 */
 	Mono<Object> getHandler(ServerHttpRequest request);
 

spring-web-reactive/src/main/java/org/springframework/web/reactive/HandlerResultHandler.java

@@ -30,8 +30,7 @@ import org.springframework.http.server.reactive.ServerHttpResponse;
 public interface HandlerResultHandler {
 
 	/**
-	 * Given a handler instance, return whether or not this {@code HandlerResultHandler}
-	 * can support it.
+	 * Whether this handler supports the given {@link HandlerResult}.
 	 *
 	 * @param result result object to check
 	 * @return whether or not this object can use the given result
@@ -39,14 +38,10 @@ public interface HandlerResultHandler {
 	boolean supports(HandlerResult result);
 
 	/**
-	 * Process the given result in an asynchronous non blocking way, by eventually modifying
-	 * response headers, or writing some data stream into the response.
-	 * Implementations should not throw exceptions but signal them via the returned
-	 * {@code Mono<Void>}.
+	 * Process the given result modifying response headers and/or writing data
+	 * to the response.
 	 *
-	 * @return A {@code Mono<Void>} used to signal the demand, and receive a notification
-	 * when the handling is complete (success or error) including the flush of the data on the
-	 * network.
+	 * @return {@code Mono<Void>} to indicate when request handling is complete.
 	 */
 	Mono<Void> handleResult(ServerHttpRequest request, ServerHttpResponse response,
 			HandlerResult result);

spring-web-reactive/src/main/java/org/springframework/web/reactive/handler/HttpHandlerAdapter.java

@@ -50,9 +50,7 @@ public class HttpHandlerAdapter implements HandlerAdapter {
 	}
 
 	@Override
-	public Mono<HandlerResult> handle(ServerHttpRequest request,
-			ServerHttpResponse response, Object handler) {
-
+	public Mono<HandlerResult> handle(ServerHttpRequest request, ServerHttpResponse response, Object handler) {
 		HttpHandler httpHandler = (HttpHandler)handler;
 		Mono<Void> completion = httpHandler.handle(request, response);
 		return Mono.just(new HandlerResult(httpHandler, completion, PUBLISHER_VOID));

spring-web-reactive/src/main/java/org/springframework/web/reactive/handler/SimpleHandlerResultHandler.java

@@ -69,7 +69,8 @@ public class SimpleHandlerResultHandler implements Ordered, HandlerResultHandler
 
 	private boolean isConvertibleToPublisher(ResolvableType type) {
 		return Publisher.class.isAssignableFrom(type.getRawClass()) ||
-				((this.conversionService != null) && this.conversionService.canConvert(type.getRawClass(), Publisher.class));
+				((this.conversionService != null) &&
+						this.conversionService.canConvert(type.getRawClass(), Publisher.class));
 	}
 
 	@SuppressWarnings("unchecked")

spring-web-reactive/src/main/java/org/springframework/web/reactive/method/HandlerMethodArgumentResolver.java

@@ -31,13 +31,9 @@ public interface HandlerMethodArgumentResolver {
 	boolean supportsParameter(MethodParameter parameter);
 
 	/**
-	 * The returned Publisher is expected to produce a single value -- i.e. the
-	 * value to use to invoke the handler method. Any additional values will be
-	 * ignored.
-	 *
-	 * <p>The publisher may also produce zero values if the argument does not
-	 * resolve to any value which will result in passing {@code null} as the
-	 * argument value.
+	 * The returned {@link Mono} may produce one or zero values if the argument
+	 * does not resolve to any value, which will result in {@code null} passed
+	 * as the argument value.
 	 */
 	Mono<Object> resolveArgument(MethodParameter parameter, ServerHttpRequest request);
 

spring-web-reactive/src/main/java/org/springframework/web/reactive/method/InvocableHandlerMethod.java

@@ -25,9 +25,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
-import reactor.Flux;
 import reactor.Mono;
-import reactor.fn.tuple.Tuple;
 
 import org.springframework.core.DefaultParameterNameDiscoverer;
 import org.springframework.core.GenericTypeResolver;
@@ -46,7 +44,7 @@ import org.springframework.web.reactive.HandlerResult;
  */
 public class InvocableHandlerMethod extends HandlerMethod {
 
-	public static final Mono<Object[]> NO_ARGS = Mono.just(new Object[0]);
+	private static final Mono<Object[]> NO_ARGS = Mono.just(new Object[0]);
 
 	private final static Object NO_VALUE = new Object();
 
@@ -85,19 +83,7 @@ public class InvocableHandlerMethod extends HandlerMethod {
 	 * never throws an exception.
 	 */
 	public Mono<HandlerResult> invokeForRequest(ServerHttpRequest request, Object... providedArgs) {
-
-		Mono<Object[]> argsPublisher = NO_ARGS;
-		try {
-			if (!ObjectUtils.isEmpty(getMethodParameters())) {
-				List<Mono<Object>> publishers = resolveArguments(request, providedArgs);
-				argsPublisher = Flux.zip(publishers, this::initArgs).next();
-			}
-		}
-		catch (Throwable ex) {
-			return Mono.error(ex);
-		}
-
-		return Flux.from(argsPublisher).concatMap(args -> {
+		return resolveArguments(request, providedArgs).then(args -> {
 			try {
 				Object value = doInvoke(args);
 				ResolvableType type =  ResolvableType.forMethodParameter(getReturnType());
@@ -111,35 +97,46 @@ public class InvocableHandlerMethod extends HandlerMethod {
 				String s = getInvocationErrorMessage(args);
 				return Mono.error(new IllegalStateException(s));
 			}
-		}).next();
+		});
 	}
 
-	private List<Mono<Object>> resolveArguments(ServerHttpRequest request, Object... providedArgs) {
-		return Stream.of(getMethodParameters())
-				.map(parameter -> {
-					parameter.initParameterNameDiscovery(this.parameterNameDiscoverer);
-					GenericTypeResolver.resolveParameterType(parameter, getBean().getClass());
-					if (!ObjectUtils.isEmpty(providedArgs)) {
-						for (Object providedArg : providedArgs) {
-							if (parameter.getParameterType().isInstance(providedArg)) {
-								return Mono.just(providedArg);
+	private Mono<Object[]> resolveArguments(ServerHttpRequest request, Object... providedArgs) {
+		if (ObjectUtils.isEmpty(getMethodParameters())) {
+			return NO_ARGS;
+		}
+		try {
+			List<Mono<Object>> monos = Stream.of(getMethodParameters())
+					.map(param -> {
+						param.initParameterNameDiscovery(this.parameterNameDiscoverer);
+						GenericTypeResolver.resolveParameterType(param, getBean().getClass());
+						if (!ObjectUtils.isEmpty(providedArgs)) {
+							for (Object providedArg : providedArgs) {
+								if (param.getParameterType().isInstance(providedArg)) {
+									return Mono.just(providedArg);
+								}
 							}
 						}
-					}
-					HandlerMethodArgumentResolver resolver = this.resolvers.stream()
-							.filter(r -> r.supportsParameter(parameter))
-							.findFirst()
-							.orElseThrow(() -> getArgError("No resolver for ", parameter, null));
-					try {
-						return resolver.resolveArgument(parameter, request)
-								.defaultIfEmpty(NO_VALUE)
-								.otherwise(ex -> Mono.error(getArgError("Error resolving ", parameter, ex)));
-					}
-					catch (Exception ex) {
-						throw getArgError("Error resolving ", parameter, ex);
-					}
-				})
-				.collect(Collectors.toList());
+						HandlerMethodArgumentResolver resolver = this.resolvers.stream()
+								.filter(r -> r.supportsParameter(param))
+								.findFirst()
+								.orElseThrow(() -> getArgError("No resolver for ", param, null));
+						try {
+							return resolver.resolveArgument(param, request)
+									.defaultIfEmpty(NO_VALUE)
+									.otherwise(ex -> Mono.error(getArgError("Error resolving ", param, ex)));
+						}
+						catch (Exception ex) {
+							throw getArgError("Error resolving ", param, ex);
+						}
+					})
+					.collect(Collectors.toList());
+
+			return Mono.when(monos).map(args ->
+					Stream.of(args.toArray()).map(o -> o != NO_VALUE ? o : null).toArray());
+		}
+		catch (Throwable ex) {
+			return Mono.error(ex);
+		}
 	}
 
 	private IllegalStateException getArgError(String message, MethodParameter param, Throwable cause) {
@@ -173,8 +170,4 @@ public class InvocableHandlerMethod extends HandlerMethod {
 				"on method [" + getBridgedMethod().toGenericString() + "]";
 	}
 
-	private Object[] initArgs(Tuple tuple) {
-		return Stream.of(tuple.toArray()).map(o -> o != NO_VALUE ? o : null).toArray();
-	}
-
 }

spring-web-reactive/src/main/java/org/springframework/web/reactive/method/annotation/RequestBodyArgumentResolver.java

@@ -65,19 +65,19 @@ public class RequestBodyArgumentResolver implements HandlerMethodArgumentResolve
 		}
 		ResolvableType type = ResolvableType.forMethodParameter(parameter);
 		Flux<ByteBuffer> body = request.getBody();
-		Flux<?> elementStream = body;
+		Flux<?> elementFlux = body;
 		ResolvableType elementType = type.hasGenerics() ? type.getGeneric(0) : type;
 
 		Decoder<?> decoder = resolveDecoder(elementType, mediaType);
 		if (decoder != null) {
-			elementStream = decoder.decode(body, elementType, mediaType);
+			elementFlux = decoder.decode(body, elementType, mediaType);
 		}
 
 		if (this.conversionService.canConvert(Publisher.class, type.getRawClass())) {
-			return Mono.just(this.conversionService.convert(elementStream, type.getRawClass()));
+			return Mono.just(this.conversionService.convert(elementFlux, type.getRawClass()));
 		}
 
-		return (Mono<Object>)Mono.from(elementStream);
+		return elementFlux.next().map(o -> o);
 	}
 
 	private Decoder<?> resolveDecoder(ResolvableType type, MediaType mediaType, Object... hints) {

spring-web-reactive/src/main/java/org/springframework/web/reactive/method/annotation/RequestMappingHandlerAdapter.java

@@ -25,7 +25,6 @@ import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import reactor.Flux;
 import reactor.Mono;
 
 import org.springframework.beans.factory.InitializingBean;
@@ -106,23 +105,18 @@ public class RequestMappingHandlerAdapter implements HandlerAdapter, Initializin
 	}
 
 	@Override
-	public Mono<HandlerResult> handle(ServerHttpRequest request,
-			ServerHttpResponse response, Object handler) {
+	public Mono<HandlerResult> handle(ServerHttpRequest request, ServerHttpResponse response,
+			Object handler) {
 
 		HandlerMethod handlerMethod = (HandlerMethod) handler;
 
 		InvocableHandlerMethod invocable = new InvocableHandlerMethod(handlerMethod);
 		invocable.setHandlerMethodArgumentResolvers(this.argumentResolvers);
 
-		Flux<HandlerResult> publisher = invocable.invokeForRequest(request).flux();
-		publisher = publisher.onErrorResumeWith(ex -> Mono.just(new HandlerResult(handler, ex)));
-
-
-		publisher = publisher.map(
-				result -> result.setExceptionMapper(
+		return invocable.invokeForRequest(request)
+				.otherwise(ex -> Mono.just(new HandlerResult(handler, ex)))
+				.map(result -> result.setExceptionMapper(
 						ex -> mapException(ex, handlerMethod, request, response)));
-
-		return publisher.next();
 	}
 
 	private Mono<HandlerResult> mapException(Throwable ex, HandlerMethod handlerMethod,