Add more subtle content negotiation in web layer

So that single Strings can be POSTed without JSON conversion.
There's still some work to do to support single POJOs in JSON, and
to reach parity with the WebFlux reactive type handlers, but it's
now closer to what we had before we moved the String conversion
out of the function layer.

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/ReactorAutoConfiguration.java

@@ -65,9 +65,9 @@ public class ReactorAutoConfiguration {
 	@ConditionalOnMissingClass("org.springframework.core.ReactiveAdapter")
 	protected static class FluxReturnValueConfiguration {
 		@Bean
-		public FluxReturnValueHandler fluxReturnValueHandler(
+		public FluxReturnValueHandler fluxReturnValueHandler(FunctionInspector inspector, 
 				HttpMessageConverters converters) {
-			return new FluxReturnValueHandler(converters.getConverters());
+			return new FluxReturnValueHandler(inspector, converters.getConverters());
 		}
 	}
 

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/request/FluxHandlerMethodArgumentResolver.java

@@ -16,7 +16,9 @@
 
 package org.springframework.cloud.function.web.flux.request;
 
+import java.nio.charset.Charset;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 
 import javax.servlet.http.HttpServletRequest;
@@ -26,6 +28,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.cloud.function.context.FunctionInspector;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.Ordered;
+import org.springframework.http.MediaType;
+import org.springframework.util.StreamUtils;
 import org.springframework.web.bind.support.WebDataBinderFactory;
 import org.springframework.web.context.request.NativeWebRequest;
 import org.springframework.web.method.support.HandlerMethodArgumentResolver;
@@ -67,13 +71,30 @@ public class FluxHandlerMethodArgumentResolver
 		if (type == null) {
 			type = Object.class;
 		}
-		List<Object> body = mapper.readValue(
-				webRequest.getNativeRequest(HttpServletRequest.class).getInputStream(),
-				mapper.getTypeFactory().constructCollectionLikeType(ArrayList.class,
-						type));
+		List<Object> body;
+		if (isPlainText(webRequest) && CharSequence.class.isAssignableFrom(type)) {
+			body = Arrays.asList(StreamUtils.copyToString(webRequest
+					.getNativeRequest(HttpServletRequest.class).getInputStream(),
+					Charset.forName("UTF-8")));
+		}
+		else {
+			body = mapper.readValue(
+					webRequest.getNativeRequest(HttpServletRequest.class)
+							.getInputStream(),
+					mapper.getTypeFactory().constructCollectionLikeType(ArrayList.class,
+							type));
+		}
 		return new FluxRequest<Object>(body);
 	}
 
+	private boolean isPlainText(NativeWebRequest webRequest) {
+		String value = webRequest.getHeader("Content-Type");
+		if (value!=null) {
+			return MediaType.valueOf(value).isCompatibleWith(MediaType.TEXT_PLAIN);
+		}
+		return false;
+	}
+
 	@Override
 	public boolean supportsParameter(MethodParameter parameter) {
 		return FluxRequest.class.isAssignableFrom(parameter.getParameterType());

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/flux/response/FluxReturnValueHandler.java

@@ -17,12 +17,15 @@
 package org.springframework.cloud.function.web.flux.response;
 
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.List;
 
 import javax.servlet.http.HttpServletResponse;
 
 import org.reactivestreams.Publisher;
 
+import org.springframework.cloud.function.context.FunctionInspector;
+import org.springframework.cloud.function.web.flux.request.FluxHandlerMethodArgumentResolver;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.MediaType;
@@ -49,8 +52,12 @@ public class FluxReturnValueHandler implements AsyncHandlerMethodReturnValueHand
 	private long timeout = 1000L;
 	private static final MediaType EVENT_STREAM = MediaType.valueOf("text/event-stream");
 
-	public FluxReturnValueHandler(List<HttpMessageConverter<?>> messageConverters) {
-		delegate = new ResponseBodyEmitterReturnValueHandler(messageConverters);
+	private FunctionInspector inspector;
+
+	public FluxReturnValueHandler(FunctionInspector inspector,
+			List<HttpMessageConverter<?>> messageConverters) {
+		this.inspector = inspector;
+		this.delegate = new ResponseBodyEmitterReturnValueHandler(messageConverters);
 	}
 
 	/**
@@ -108,23 +115,52 @@ public class FluxReturnValueHandler implements AsyncHandlerMethodReturnValueHand
 		}
 		Publisher<?> flux = (Publisher<?>) adaptFrom;
 
+		Object handler = webRequest.getAttribute(
+				FluxHandlerMethodArgumentResolver.HANDLER,
+				NativeWebRequest.SCOPE_REQUEST);
+		Class<?> type = inspector.getOutputType(inspector.getName(handler));
+
 		MediaType mediaType = null;
-		if (webRequest.getHeader("Accept") != null) {
-			for (MediaType type : MediaType
-					.parseMediaTypes(webRequest.getHeader("Accept"))) {
-				if (!MediaType.ALL.equals(type)
-						&& MediaType.APPLICATION_JSON.isCompatibleWith(type)) {
-					mediaType = MediaType.APPLICATION_JSON;
-					break;
-				} else if (mediaType==null) {
-					mediaType = type;
-				}
-			}
+		if (isPlainText(webRequest) && CharSequence.class.isAssignableFrom(type)) {
+			mediaType = MediaType.TEXT_PLAIN;
+		} else {
+			mediaType = findMediaType(webRequest);
 		}
 		delegate.handleReturnValue(getEmitter(timeout, flux, mediaType), returnType,
 				mavContainer, webRequest);
 	}
 
+	private MediaType findMediaType(NativeWebRequest webRequest) {
+		List<MediaType> accepts = Arrays.asList(MediaType.ALL);
+		MediaType mediaType = null;
+		if (webRequest.getHeader("Accept") != null) {
+			accepts = MediaType.parseMediaTypes(webRequest.getHeader("Accept"));
+			for (MediaType accept : accepts) {
+				if (!MediaType.ALL.equals(accept)
+						&& MediaType.APPLICATION_JSON.isCompatibleWith(accept)) {
+					mediaType = MediaType.APPLICATION_JSON;
+					// Prefer JSON if that is acceptable
+					break;
+				}
+				else if (mediaType == null) {
+					mediaType = accept;
+				}
+			}
+		}
+		if (mediaType == null) {
+			mediaType = MediaType.APPLICATION_JSON;
+		}
+		return mediaType;
+	}
+
+	private boolean isPlainText(NativeWebRequest webRequest) {
+		String value = webRequest.getHeader("Content-Type");
+		if (value != null) {
+			return MediaType.valueOf(value).isCompatibleWith(MediaType.TEXT_PLAIN);
+		}
+		return false;
+	}
+
 	private ResponseBodyEmitter getEmitter(Long timeout, Publisher<?> flux,
 			MediaType mediaType) {
 		Publisher<?> exported = flux instanceof Mono ? Mono.from(flux)