Fix handling for ResponseEntity<Flux<T>> in Spring MVC

Issue: SPR-15456

Author: rstoyanchev

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

@@ -126,6 +126,7 @@ public void handleReturnValue(Object returnValue, MethodParameter returnType,
 			response.setStatus(responseEntity.getStatusCodeValue());
 			outputMessage.getHeaders().putAll(responseEntity.getHeaders());
 			returnValue = responseEntity.getBody();
+			returnType = returnType.nested();
 			if (returnValue == null) {
 				mavContainer.setRequestHandled(true);
 				outputMessage.flush();

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

@@ -22,8 +22,11 @@
 
 import org.junit.Before;
 import org.junit.Test;
+import reactor.core.publisher.EmitterProcessor;
+import reactor.core.publisher.Flux;
 
 import org.springframework.core.MethodParameter;
+import org.springframework.core.ResolvableType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.http.converter.StringHttpMessageConverter;
@@ -95,6 +98,13 @@ public void supportsReturnTypes() throws Exception {
 
 		assertTrue(this.handler.supportsReturnType(
 				on(TestController.class).resolveReturnType(ResponseEntity.class, ResponseBodyEmitter.class)));
+
+		assertTrue(this.handler.supportsReturnType(
+				on(TestController.class).resolveReturnType(Flux.class, String.class)));
+
+		assertTrue(this.handler.supportsReturnType(
+				on(TestController.class).resolveReturnType(forClassWithGenerics(ResponseEntity.class,
+								forClassWithGenerics(Flux.class, String.class)))));
 	}
 
 	@Test
@@ -103,8 +113,8 @@ public void doesNotSupportReturnTypes() throws Exception {
 		assertFalse(this.handler.supportsReturnType(
 				on(TestController.class).resolveReturnType(ResponseEntity.class, String.class)));
 
-		assertFalse(this.handler.supportsReturnType(on(TestController.class)
-				.resolveReturnType(forClassWithGenerics(ResponseEntity.class,
+		assertFalse(this.handler.supportsReturnType(
+				on(TestController.class).resolveReturnType(forClassWithGenerics(ResponseEntity.class,
 						forClassWithGenerics(AtomicReference.class, String.class)))));
 
 		assertFalse(this.handler.supportsReturnType(
@@ -195,6 +205,27 @@ public void sseEmitter() throws Exception {
 						"\n", this.response.getContentAsString());
 	}
 
+	@Test
+	public void responseBodyFlux() throws Exception {
+
+		this.request.addHeader("Accept", "text/event-stream");
+
+		MethodParameter type = on(TestController.class).resolveReturnType(Flux.class, String.class);
+		EmitterProcessor<String> processor = EmitterProcessor.create();
+		this.handler.handleReturnValue(processor, type, this.mavContainer, this.webRequest);
+
+		assertTrue(this.request.isAsyncStarted());
+		assertEquals(200, this.response.getStatus());
+		assertEquals("text/event-stream;charset=UTF-8", this.response.getContentType());
+
+		processor.onNext("foo");
+		processor.onNext("bar");
+		processor.onNext("baz");
+		processor.onComplete();
+
+		assertEquals("data:foo\n\ndata:bar\n\ndata:baz\n\n", this.response.getContentAsString());
+	}
+
 	@Test
 	public void responseEntitySse() throws Exception {
 		MethodParameter type = on(TestController.class).resolveReturnType(ResponseEntity.class, SseEmitter.class);
@@ -218,6 +249,27 @@ public void responseEntitySseNoContent() throws Exception {
 		assertEquals(Collections.singletonList("bar"), this.response.getHeaders("foo"));
 	}
 
+	@Test
+	public void responseEntityFlux() throws Exception {
+
+		EmitterProcessor<String> processor = EmitterProcessor.create();
+		ResponseEntity<Flux<String>> entity = ResponseEntity.ok().body(processor);
+		ResolvableType bodyType = forClassWithGenerics(Flux.class, String.class);
+		MethodParameter type = on(TestController.class).resolveReturnType(ResponseEntity.class, bodyType);
+		this.handler.handleReturnValue(entity, type, this.mavContainer, this.webRequest);
+
+		assertTrue(this.request.isAsyncStarted());
+		assertEquals(200, this.response.getStatus());
+		assertEquals("text/plain", this.response.getContentType());
+
+		processor.onNext("foo");
+		processor.onNext("bar");
+		processor.onNext("baz");
+		processor.onComplete();
+
+		assertEquals("foobarbaz", this.response.getContentAsString());
+	}
+
 
 	@SuppressWarnings("unused")
 	private static class TestController {
@@ -236,6 +288,10 @@ private static class TestController {
 
 		private ResponseEntity h7() { return null; }
 
+		private Flux<String> h8() { return null; }
+
+		private ResponseEntity<Flux<String>> h9() { return null; }
+
 	}