Spring MVC supports reactive return values

This commit adds support for reactive library types to be returned
directly from controller methods adapting them either to a
ResponseBodyEmitter (streaming) or DeferredResult (non-streaming).

The reactive libraries supported are the ones that can adapted to a
Reactive Streams Publisher through the ReactiveAdapterRegistry.

Issue: SPR-15365

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

@@ -0,0 +1,325 @@
+/*
+ * Copyright 2002-2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.web.servlet.mvc.method.annotation;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Optional;
+
+import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ReactiveAdapter;
+import org.springframework.core.ReactiveAdapterRegistry;
+import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
+import org.springframework.http.server.ServerHttpResponse;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MimeType;
+import org.springframework.web.HttpMediaTypeNotAcceptableException;
+import org.springframework.web.accept.ContentNegotiationManager;
+import org.springframework.web.context.request.NativeWebRequest;
+import org.springframework.web.context.request.RequestAttributes;
+import org.springframework.web.context.request.async.DeferredResult;
+import org.springframework.web.context.request.async.WebAsyncUtils;
+import org.springframework.web.method.support.ModelAndViewContainer;
+import org.springframework.web.servlet.HandlerMapping;
+
+
+/**
+ * Private helper class to assist with handling "reactive" return values types
+ * that can be adapted to a Reactive Streams {@link Publisher} through the
+ * {@link ReactiveAdapterRegistry}.
+ *
+ * <p>Such return values may be bridged to a {@link ResponseBodyEmitter} for
+ * streaming purposes at the presence of a streaming media type or based on the
+ * generic type.
+ *
+ * <p>For all other cases {@code Publisher} output is collected and bridged to
+ * {@link DeferredResult} for standard async request processing.
+ *
+ * @author Rossen Stoyanchev
+ * @since 5.0
+ */
+class ReactiveTypeHandler {
+
+	private static final MediaType JSON_TYPE = new MediaType("application", "*+json");
+
+
+	private final ReactiveAdapterRegistry reactiveRegistry;
+
+	private final ContentNegotiationManager contentNegotiationManager;
+
+
+	ReactiveTypeHandler(ReactiveAdapterRegistry registry, ContentNegotiationManager manager) {
+		Assert.notNull(registry, "ReactiveAdapterRegistry is required");
+		Assert.notNull(manager, "ContentNegotiationManager is required");
+		this.reactiveRegistry = registry;
+		this.contentNegotiationManager = manager;
+	}
+
+
+	/**
+	 * Whether the type can be adapted to a Reactive Streams {@link Publisher}.
+	 */
+	public boolean isReactiveType(Class<?> type) {
+		return this.reactiveRegistry.hasAdapters() && this.reactiveRegistry.getAdapter(type) != null;
+	}
+
+
+	/**
+	 * Process the given reactive return value and decide whether to adapt it
+	 * to a {@link ResponseBodyEmitter} or a {@link DeferredResult}.
+	 *
+	 * @return an emitter for streaming or {@code null} if handled internally
+	 * with a {@link DeferredResult}.
+	 */
+	public ResponseBodyEmitter handleValue(Object returnValue, MethodParameter returnType,
+			ModelAndViewContainer mav, NativeWebRequest request) throws Exception {
+
+		Assert.notNull(returnValue, "Expected return value");
+		ReactiveAdapter adapter = this.reactiveRegistry.getAdapter(returnValue.getClass());
+		Assert.state(adapter != null, "Unexpected return value: " + returnValue);
+
+		Class<?> elementType = returnType.nested().getNestedParameterType();
+		
+		Collection<MediaType> mediaTypes = getMediaTypes(request);
+		Optional<MediaType> mediaType = mediaTypes.stream().filter(MimeType::isConcrete).findFirst();
+		boolean jsonArrayOfStrings = isJsonArrayOfStrings(elementType, mediaType);
+
+		if (adapter.isMultiValue()) {
+			if (mediaTypes.stream().anyMatch(MediaType.TEXT_EVENT_STREAM::includes) ||
+					ServerSentEvent.class.isAssignableFrom(elementType)) {
+				SseEmitter emitter = new SseEmitter();
+				new SseEmitterSubscriber(emitter).connect(adapter, returnValue);
+				return emitter;
+			}
+			if (mediaTypes.stream().anyMatch(MediaType.APPLICATION_STREAM_JSON::includes)) {
+				ResponseBodyEmitter emitter = getEmitter(MediaType.APPLICATION_STREAM_JSON);
+				new JsonEmitterSubscriber(emitter).connect(adapter, returnValue);
+				return emitter;
+			}
+			if (CharSequence.class.isAssignableFrom(elementType) && !jsonArrayOfStrings) {
+				ResponseBodyEmitter emitter = getEmitter(mediaType.orElse(MediaType.TEXT_PLAIN));
+				new TextEmitterSubscriber(emitter).connect(adapter, returnValue);
+				return emitter;
+			}
+		}
+
+		// Not streaming...
+		DeferredResult<Object> result = new DeferredResult<>();
+		new DeferredResultSubscriber(result, jsonArrayOfStrings).connect(adapter, returnValue);
+		WebAsyncUtils.getAsyncManager(request).startDeferredResultProcessing(result, mav);
+
+		return null;
+	}
+
+	@SuppressWarnings("unchecked")
+	private Collection<MediaType> getMediaTypes(NativeWebRequest request)
+			throws HttpMediaTypeNotAcceptableException {
+
+		Collection<MediaType> mediaTypes = (Collection<MediaType>) request.getAttribute(
+				HandlerMapping.PRODUCIBLE_MEDIA_TYPES_ATTRIBUTE, RequestAttributes.SCOPE_REQUEST);
+
+		return CollectionUtils.isEmpty(mediaTypes) ?
+				this.contentNegotiationManager.resolveMediaTypes(request) : mediaTypes;
+	}
+
+	@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
+	private boolean isJsonArrayOfStrings(Class<?> elementType, Optional<MediaType> mediaType) {
+		return CharSequence.class.isAssignableFrom(elementType) && mediaType.filter(type ->
+				MediaType.APPLICATION_JSON.includes(type) || JSON_TYPE.includes(type)).isPresent();
+	}
+
+	private ResponseBodyEmitter getEmitter(MediaType mediaType) {
+		return new ResponseBodyEmitter() {
+
+			@Override
+			protected void extendResponse(ServerHttpResponse outputMessage) {
+				outputMessage.getHeaders().setContentType(mediaType);
+			}
+		};
+	}
+
+
+	private static abstract class AbstractEmitterSubscriber implements Subscriber<Object> {
+
+		private final ResponseBodyEmitter emitter;
+
+		private Subscription subscription;
+
+
+		protected AbstractEmitterSubscriber(ResponseBodyEmitter emitter) {
+			this.emitter = emitter;
+		}
+
+
+		public void connect(ReactiveAdapter adapter, Object returnValue) {
+			Publisher<Object> publisher = adapter.toPublisher(returnValue);
+			publisher.subscribe(this);
+		}
+
+
+		protected ResponseBodyEmitter getEmitter() {
+			return this.emitter;
+		}
+
+		@Override
+		public void onSubscribe(Subscription subscription) {
+			this.subscription = subscription;
+			this.emitter.onTimeout(subscription::cancel);
+			subscription.request(1);
+		}
+
+		@Override
+		public void onNext(Object element) {
+			try {
+				send(element);
+				this.subscription.request(1);
+			}
+			catch (IOException ex) {
+				this.subscription.cancel();
+			}
+		}
+
+		protected abstract void send(Object element) throws IOException;
+
+		@Override
+		public void onError(Throwable ex) {
+			this.emitter.completeWithError(ex);
+		}
+
+		@Override
+		public void onComplete() {
+			this.emitter.complete();
+		}
+	}
+
+
+	private static class SseEmitterSubscriber extends AbstractEmitterSubscriber {
+
+		SseEmitterSubscriber(SseEmitter sseEmitter) {
+			super(sseEmitter);
+		}
+
+		@Override
+		protected void send(Object element) throws IOException {
+			if (element instanceof ServerSentEvent) {
+				ServerSentEvent<?> event = (ServerSentEvent<?>) element;
+				((SseEmitter) getEmitter()).send(adapt(event));
+			}
+			else {
+				getEmitter().send(element, MediaType.APPLICATION_JSON);
+			}
+		}
+
+		private SseEmitter.SseEventBuilder adapt(ServerSentEvent<?> event) {
+			SseEmitter.SseEventBuilder builder = SseEmitter.event();
+			event.id().ifPresent(builder::id);
+			event.comment().ifPresent(builder::comment);
+			event.data().ifPresent(builder::data);
+			event.retry().ifPresent(duration -> builder.reconnectTime(duration.toMillis()));
+			return builder;
+		}
+	}
+
+
+	private static class JsonEmitterSubscriber extends AbstractEmitterSubscriber {
+
+		JsonEmitterSubscriber(ResponseBodyEmitter emitter) {
+			super(emitter);
+		}
+
+		@Override
+		protected void send(Object element) throws IOException {
+			getEmitter().send(element, MediaType.APPLICATION_JSON);
+			getEmitter().send("\n", MediaType.TEXT_PLAIN);
+		}
+	}
+
+
+	private static class TextEmitterSubscriber extends AbstractEmitterSubscriber {
+
+		TextEmitterSubscriber(ResponseBodyEmitter emitter) {
+			super(emitter);
+		}
+
+		@Override
+		protected void send(Object element) throws IOException {
+			getEmitter().send(element, MediaType.TEXT_PLAIN);
+		}
+	}
+
+
+	private static class DeferredResultSubscriber implements Subscriber<Object> {
+
+		private final DeferredResult<Object> result;
+
+		private final boolean jsonArrayOfStrings;
+
+		private final CollectedValuesList values = new CollectedValuesList();
+
+
+		DeferredResultSubscriber(DeferredResult<Object> result, boolean jsonArrayOfStrings) {
+			this.result = result;
+			this.jsonArrayOfStrings = jsonArrayOfStrings;
+		}
+
+
+		public void connect(ReactiveAdapter adapter, Object returnValue) {
+			Publisher<Object> publisher = adapter.toPublisher(returnValue);
+			publisher.subscribe(this);
+		}
+
+		@Override
+		public void onSubscribe(Subscription subscription) {
+			this.result.onTimeout(subscription::cancel);
+			subscription.request(Long.MAX_VALUE);
+		}
+
+		@Override
+		public void onNext(Object element) {
+			this.values.add(element);
+		}
+
+		@Override
+		public void onError(Throwable ex) {
+			this.result.setErrorResult(ex);
+		}
+
+		@Override
+		public void onComplete() {
+			if (this.values.size() > 1) {
+				this.result.setResult(this.values);
+			}
+			else if (this.values.size() == 1) {
+				this.result.setResult(this.values.get(0));
+			}
+			else {
+				this.result.setResult(this.jsonArrayOfStrings ? this.values : null);
+			}
+		}
+	}
+
+	@SuppressWarnings("serial")
+	static class CollectedValuesList extends ArrayList<Object> {
+	}
+
+}

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

@@ -36,6 +36,7 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory;
 import org.springframework.core.DefaultParameterNameDiscoverer;
 import org.springframework.core.MethodIntrospector;
 import org.springframework.core.ParameterNameDiscoverer;
+import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.annotation.AnnotationAwareOrderComparator;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.task.AsyncTaskExecutor;
@@ -143,6 +144,8 @@ public class RequestMappingHandlerAdapter extends AbstractHandlerMethodAdapter
 
 	private DeferredResultProcessingInterceptor[] deferredResultInterceptors = new DeferredResultProcessingInterceptor[0];
 
+	private ReactiveAdapterRegistry reactiveRegistry = new ReactiveAdapterRegistry();
+
 	private boolean ignoreDefaultModelOnRedirect = false;
 
 	private int cacheSecondsForSessionAttributeHandlers = 0;
@@ -408,6 +411,22 @@ public class RequestMappingHandlerAdapter extends AbstractHandlerMethodAdapter
 		this.deferredResultInterceptors = interceptors.toArray(new DeferredResultProcessingInterceptor[interceptors.size()]);
 	}
 
+	/**
+	 * Configure the registry for reactive library types to be supported as
+	 * return values from controller methods.
+	 */
+	public void setReactiveRegistry(ReactiveAdapterRegistry reactiveRegistry) {
+		Assert.notNull(reactiveRegistry, "ReactiveAdapterRegistry is required");
+		this.reactiveRegistry = this.reactiveRegistry;
+	}
+
+	/**
+	 * Return the configured reactive type registry of adapters.
+	 */
+	public ReactiveAdapterRegistry getReactiveAdapterRegistry() {
+		return this.reactiveRegistry;
+	}
+
 	/**
 	 * By default the content of the "default" model is used both during
 	 * rendering and redirect scenarios. Alternatively a controller method
@@ -663,7 +682,8 @@ public class RequestMappingHandlerAdapter extends AbstractHandlerMethodAdapter
 		handlers.add(new ModelAndViewMethodReturnValueHandler());
 		handlers.add(new ModelMethodProcessor());
 		handlers.add(new ViewMethodReturnValueHandler());
-		handlers.add(new ResponseBodyEmitterReturnValueHandler(getMessageConverters()));
+		handlers.add(new ResponseBodyEmitterReturnValueHandler(getMessageConverters(),
+				this.reactiveRegistry, this.contentNegotiationManager));
 		handlers.add(new StreamingResponseBodyReturnValueHandler());
 		handlers.add(new HttpEntityMethodProcessor(getMessageConverters(),
 				this.contentNegotiationManager, this.requestResponseBodyAdvice));

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

@@ -26,6 +26,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.core.MethodParameter;
+import org.springframework.core.ReactiveAdapterRegistry;
 import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
@@ -35,6 +36,7 @@ import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.http.server.ServerHttpResponse;
 import org.springframework.http.server.ServletServerHttpResponse;
 import org.springframework.util.Assert;
+import org.springframework.web.accept.ContentNegotiationManager;
 import org.springframework.web.context.request.NativeWebRequest;
 import org.springframework.web.context.request.async.DeferredResult;
 import org.springframework.web.context.request.async.WebAsyncUtils;
@@ -57,10 +59,15 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
 
 	private final List<HttpMessageConverter<?>> messageConverters;
 
+	private final ReactiveTypeHandler reactiveHandler;
+
+
+	public ResponseBodyEmitterReturnValueHandler(List<HttpMessageConverter<?>> messageConverters,
+			ReactiveAdapterRegistry reactiveRegistry, ContentNegotiationManager manager) {
 
-	public ResponseBodyEmitterReturnValueHandler(List<HttpMessageConverter<?>> messageConverters) {
 		Assert.notEmpty(messageConverters, "HttpMessageConverter List must not be empty");
 		this.messageConverters = messageConverters;
+		this.reactiveHandler = new ReactiveTypeHandler(reactiveRegistry, manager);
 	}
 
 
@@ -71,11 +78,8 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
 				ResolvableType.forMethodParameter(returnType).getGeneric(0).resolve() :
 				returnType.getParameterType();
 
-		return bodyType != null && supportsBodyType(bodyType);
-	}
-
-	private boolean supportsBodyType(Class<?> bodyType) {
-		return ResponseBodyEmitter.class.isAssignableFrom(bodyType);
+		return bodyType != null && (ResponseBodyEmitter.class.isAssignableFrom(bodyType) ||
+				this.reactiveHandler.isReactiveType(bodyType));
 	}
 
 	@Override
@@ -111,7 +115,11 @@ public class ResponseBodyEmitterReturnValueHandler implements HandlerMethodRetur
 			emitter = (ResponseBodyEmitter) returnValue;
 		}
 		else {
-			throw new IllegalStateException("Unexpected return value type: " + returnValue);
+			emitter = this.reactiveHandler.handleValue(returnValue, returnType, mavContainer, webRequest);
+		}
+
+		if (emitter == null) {
+			return;
 		}
 
 		emitter.extendResponse(outputMessage);

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

@@ -20,6 +20,7 @@ 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;
@@ -270,7 +271,13 @@ public class ServletInvocableHandlerMethod extends InvocableHandlerMethod {
 		public ConcurrentResultMethodParameter(Object returnValue) {
 			super(-1);
 			this.returnValue = returnValue;
-			this.returnType = ResolvableType.forType(super.getGenericParameterType()).getGeneric(0);
+
+			ResolvableType candidateReturnType =
+					ResolvableType.forType(super.getGenericParameterType()).getGeneric(0);
+
+			this.returnType = returnValue instanceof ReactiveTypeHandler.CollectedValuesList ?
+					ResolvableType.forClassWithGenerics(List.class, candidateReturnType) :
+					candidateReturnType;
 		}
 
 		public ConcurrentResultMethodParameter(ConcurrentResultMethodParameter original) {