Properly support ResponseEntity<Flux<T>> in Spring MVC

Issue: SPR-15478

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandler.java

@@ -18,6 +18,7 @@ package org.springframework.web.servlet.mvc.method.annotation;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
@@ -31,6 +32,7 @@ import org.reactivestreams.Subscription;
 import org.springframework.core.MethodParameter;
 import org.springframework.core.ReactiveAdapter;
 import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.core.ResolvableType;
 import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.core.task.TaskExecutor;
 import org.springframework.http.MediaType;
@@ -116,14 +118,15 @@ class ReactiveTypeHandler {
 		ReactiveAdapter adapter = this.reactiveRegistry.getAdapter(returnValue.getClass());
 		Assert.state(adapter != null, "Unexpected return value: " + returnValue);
 
-		Class<?> elementType = returnType.nested().getNestedParameterType();
-		
+		ResolvableType elementType = ResolvableType.forMethodParameter(returnType).getGeneric(0);
+		Class<?> elementClass = elementType.resolve(Object.class);
+
 		Collection<MediaType> mediaTypes = getMediaTypes(request);
 		Optional<MediaType> mediaType = mediaTypes.stream().filter(MimeType::isConcrete).findFirst();
 
 		if (adapter.isMultiValue()) {
 			if (mediaTypes.stream().anyMatch(MediaType.TEXT_EVENT_STREAM::includes) ||
-					ServerSentEvent.class.isAssignableFrom(elementType)) {
+					ServerSentEvent.class.isAssignableFrom(elementClass)) {
 				SseEmitter emitter = new SseEmitter();
 				new SseEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
 				return emitter;
@@ -133,7 +136,7 @@ class ReactiveTypeHandler {
 				new JsonEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
 				return emitter;
 			}
-			if (CharSequence.class.isAssignableFrom(elementType) && !isJsonStringArray(elementType, mediaType)) {
+			if (CharSequence.class.isAssignableFrom(elementClass) && !isJsonStringArray(elementClass, mediaType)) {
 				ResponseBodyEmitter emitter = getEmitter(mediaType.orElse(MediaType.TEXT_PLAIN));
 				new TextEmitterSubscriber(emitter, this.taskExecutor).connect(adapter, returnValue);
 				return emitter;
@@ -142,7 +145,7 @@ class ReactiveTypeHandler {
 
 		// Not streaming...
 		DeferredResult<Object> result = new DeferredResult<>();
-		new DeferredResultSubscriber(result, adapter).connect(adapter, returnValue);
+		new DeferredResultSubscriber(result, adapter, elementType).connect(adapter, returnValue);
 		WebAsyncUtils.getAsyncManager(request).startDeferredResultProcessing(result, mav);
 
 		return null;
@@ -389,12 +392,15 @@ class ReactiveTypeHandler {
 
 		private final boolean multiValueSource;
 
-		private final CollectedValuesList values = new CollectedValuesList();
+		private final CollectedValuesList values;
 
 
-		DeferredResultSubscriber(DeferredResult<Object> result, ReactiveAdapter adapter) {
+		DeferredResultSubscriber(DeferredResult<Object> result, ReactiveAdapter adapter,
+				ResolvableType elementType) {
+
 			this.result = result;
 			this.multiValueSource = adapter.isMultiValue();
+			this.values = new CollectedValuesList(elementType);
 		}
 
 
@@ -435,6 +441,16 @@ class ReactiveTypeHandler {
 
 	@SuppressWarnings("serial")
 	static class CollectedValuesList extends ArrayList<Object> {
+
+		private final ResolvableType elementType;
+
+		CollectedValuesList(ResolvableType elementType) {
+			this.elementType = elementType;
+		}
+
+		public ResolvableType getReturnType() {
+			return ResolvableType.forClassWithGenerics(List.class, this.elementType);
+		}
 	}
 
 }

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethod.java

@@ -20,7 +20,6 @@ import java.io.IOException;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Method;
 import java.lang.reflect.Type;
-import java.util.List;
 import java.util.concurrent.Callable;
 
 import org.springframework.core.MethodParameter;
@@ -28,6 +27,7 @@ import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpStatus;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.StringUtils;
+import org.springframework.web.bind.annotation.ResponseBody;
 import org.springframework.web.bind.annotation.ResponseStatus;
 import org.springframework.web.context.request.ServletWebRequest;
 import org.springframework.web.method.HandlerMethod;
@@ -249,10 +249,9 @@ public class ServletInvocableHandlerMethod extends InvocableHandlerMethod {
 		public ConcurrentResultMethodParameter(Object returnValue) {
 			super(-1);
 			this.returnValue = returnValue;
-			ResolvableType candidateReturnType =
-					ResolvableType.forType(super.getGenericParameterType()).getGeneric(0);
 			this.returnType = (returnValue instanceof ReactiveTypeHandler.CollectedValuesList ?
-					ResolvableType.forClassWithGenerics(List.class, candidateReturnType) : candidateReturnType);
+					((ReactiveTypeHandler.CollectedValuesList) returnValue).getReturnType() :
+					ResolvableType.forType(super.getGenericParameterType()).getGeneric(0));
 		}
 
 		public ConcurrentResultMethodParameter(ConcurrentResultMethodParameter original) {
@@ -277,6 +276,17 @@ public class ServletInvocableHandlerMethod extends InvocableHandlerMethod {
 			return this.returnType.getType();
 		}
 
+		@Override
+		public <T extends Annotation> boolean hasMethodAnnotation(Class<T> annotationType) {
+
+			// Ensure @ResponseBody-style handling for values collected from a reactive type
+			// even if actual return type is ResponseEntity<Flux<T>>
+
+			return ResponseBody.class.equals(annotationType) &&
+					this.returnValue instanceof ReactiveTypeHandler.CollectedValuesList ||
+					super.hasMethodAnnotation(annotationType);
+		}
+
 		@Override
 		public ConcurrentResultMethodParameter clone() {
 			return new ConcurrentResultMethodParameter(this);

spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethodTests.java

@@ -28,6 +28,7 @@ import org.junit.Test;
 import reactor.core.publisher.Flux;
 
 import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
 import org.springframework.core.annotation.AliasFor;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
@@ -278,8 +279,8 @@ public class ServletInvocableHandlerMethodTests {
 	@Test
 	public void wrapConcurrentResult_CollectedValuesList() throws Exception {
 		List<HttpMessageConverter<?>> converters = Collections.singletonList(new MappingJackson2HttpMessageConverter());
-		this.request.addHeader("Accept", "application/json");
-		ReactiveTypeHandler.CollectedValuesList result = new ReactiveTypeHandler.CollectedValuesList();
+		ResolvableType elementType = ResolvableType.forClass(List.class);
+		ReactiveTypeHandler.CollectedValuesList result = new ReactiveTypeHandler.CollectedValuesList(elementType);
 		result.add(Arrays.asList("foo1", "bar1"));
 		result.add(Arrays.asList("foo2", "bar2"));
 
@@ -293,6 +294,24 @@ public class ServletInvocableHandlerMethodTests {
 		assertEquals("[[\"foo1\",\"bar1\"],[\"foo2\",\"bar2\"]]", this.response.getContentAsString());
 	}
 
+	@Test // SPR-15478
+	public void wrapConcurrentResult_CollectedValuesListWithResponseEntity() throws Exception {
+		List<HttpMessageConverter<?>> converters = Collections.singletonList(new MappingJackson2HttpMessageConverter());
+		ResolvableType elementType = ResolvableType.forClass(Bar.class);
+		ReactiveTypeHandler.CollectedValuesList result = new ReactiveTypeHandler.CollectedValuesList(elementType);
+		result.add(new Bar("foo"));
+		result.add(new Bar("bar"));
+
+		ContentNegotiationManager manager = new ContentNegotiationManager();
+		this.returnValueHandlers.addHandler(new RequestResponseBodyMethodProcessor(converters, manager));
+		ServletInvocableHandlerMethod hm = getHandlerMethod(new ResponseEntityHandler(), "handleFlux");
+		hm = hm.wrapConcurrentResult(result);
+		hm.invokeAndHandle(this.webRequest, this.mavContainer);
+
+		assertEquals(200, this.response.getStatus());
+		assertEquals("[{\"value\":\"foo\"},{\"value\":\"bar\"}]", this.response.getContentAsString());
+	}
+
 	@Test  // SPR-12287 (16/Oct/14 comments)
 	public void responseEntityRawTypeWithNullBody() throws Exception {
 		this.returnValueHandlers.addHandler(new HttpEntityMethodProcessor(this.converters));
@@ -360,17 +379,14 @@ public class ServletInvocableHandlerMethodTests {
 	@ResponseStatus(HttpStatus.BAD_REQUEST)
 	private static class ResponseStatusHandler {
 
-		public void handle() {
-		}
+		public void handle() { }
 	}
 
 
 	private static class MethodLevelResponseBodyHandler {
 
 		@ResponseBody
-		public DeferredResult<String> handle() {
-			return new DeferredResult<>();
-		}
+		public DeferredResult<String> handle() { return null; }
 
 		// Unusual but legal return type
 		// Properly test generic type handling of Flux values collected to a List
@@ -384,18 +400,14 @@ public class ServletInvocableHandlerMethodTests {
 	@ResponseBody
 	private static class TypeLevelResponseBodyHandler {
 
-		public DeferredResult<String> handle() {
-			return new DeferredResult<>();
-		}
+		public DeferredResult<String> handle() { return null; }
 	}
 
 
 	private static class DeferredResultSubclassHandler {
 
 		@ResponseBody
-		public CustomDeferredResult handle() {
-			return new CustomDeferredResult();
-		}
+		public CustomDeferredResult handle() { return null; }
 	}
 
 
@@ -406,13 +418,11 @@ public class ServletInvocableHandlerMethodTests {
 	@SuppressWarnings("unused")
 	private static class ResponseEntityHandler {
 
-		public DeferredResult<ResponseEntity<String>> handleDeferred() {
-			return new DeferredResult<>();
-		}
+		public DeferredResult<ResponseEntity<String>> handleDeferred() { return null; }
 
-		public ResponseEntity<Void> handleRawType() {
-			return ResponseEntity.ok().build();
-		}
+		public ResponseEntity<Void> handleRawType() { return null; }
+
+		public ResponseEntity<Flux<Bar>> handleFlux() { return null; }
 	}
 
 
@@ -440,4 +450,17 @@ public class ServletInvocableHandlerMethodTests {
 
 	}
 
+	private static class Bar {
+
+		private final String value;
+
+		public Bar(String value) {
+			this.value = value;
+		}
+
+		public String getValue() {
+			return this.value;
+		}
+	}
+
 }