diff --git a/build.gradle b/build.gradle index 2bda412fda..a4df6f1287 100644 --- a/build.gradle +++ b/build.gradle @@ -914,6 +914,7 @@ project("spring-webmvc") { exclude group: "org.springframework", module: "spring-web" } optional('org.webjars:webjars-locator:0.32-1') + optional("org.reactivestreams:reactive-streams") testCompile("org.xmlunit:xmlunit-matchers:${xmlunitVersion}") testCompile("dom4j:dom4j:1.6.1") { exclude group: "xml-apis", module: "xml-apis" @@ -937,6 +938,10 @@ project("spring-webmvc") { testCompile("joda-time:joda-time:${jodaVersion}") testCompile("org.slf4j:slf4j-jcl:${slf4jVersion}") testCompile("org.mozilla:rhino:1.7.7.1") + testCompile("io.projectreactor:reactor-core") + testCompile("io.reactivex:rxjava:${rxjavaVersion}") + testCompile("io.reactivex.rxjava2:rxjava:${rxjava2Version}") + testCompile("io.reactivex:rxjava-reactive-streams:${rxjavaAdapterVersion}") testRuntime("org.jruby:jruby:9.1.8.0") testRuntime("org.python:jython-standalone:2.5.3") testRuntime("org.jetbrains.kotlin:kotlin-compiler:${kotlinVersion}") diff --git a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java index 7954c8c8aa..dfe7625391 100644 --- a/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java +++ b/spring-core/src/main/java/org/springframework/core/ReactiveAdapterRegistry.java @@ -78,6 +78,16 @@ public class ReactiveAdapterRegistry { } + /** + * Whether the registry has any adapters which would be the case if any of + * Reactor, RxJava 2, or RxJava 1 (+ RxJava Reactive Streams bridge) are + * present on the classpath. + */ + public boolean hasAdapters() { + return !this.adapters.isEmpty(); + } + + /** * Register a reactive type along with functions to adapt to and from a * Reactive Streams {@link Publisher}. The functions can assume their @@ -113,6 +123,10 @@ public class ReactiveAdapterRegistry { Object sourceToUse = (source instanceof Optional ? ((Optional) source).orElse(null) : source); Class clazz = (sourceToUse != null ? sourceToUse.getClass() : reactiveType); + if (reactiveType == null) { + return null; + } + return this.adapters.stream() .filter(adapter -> adapter.getReactiveType() == clazz) .findFirst() diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandler.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandler.java new file mode 100644 index 0000000000..4dbfcaaa4c --- /dev/null +++ b/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}. + * + *

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. + * + *

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 mediaTypes = getMediaTypes(request); + Optional 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 result = new DeferredResult<>(); + new DeferredResultSubscriber(result, jsonArrayOfStrings).connect(adapter, returnValue); + WebAsyncUtils.getAsyncManager(request).startDeferredResultProcessing(result, mav); + + return null; + } + + @SuppressWarnings("unchecked") + private Collection getMediaTypes(NativeWebRequest request) + throws HttpMediaTypeNotAcceptableException { + + Collection mediaTypes = (Collection) 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) { + 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 { + + private final ResponseBodyEmitter emitter; + + private Subscription subscription; + + + protected AbstractEmitterSubscriber(ResponseBodyEmitter emitter) { + this.emitter = emitter; + } + + + public void connect(ReactiveAdapter adapter, Object returnValue) { + Publisher 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 { + + private final DeferredResult result; + + private final boolean jsonArrayOfStrings; + + private final CollectedValuesList values = new CollectedValuesList(); + + + DeferredResultSubscriber(DeferredResult result, boolean jsonArrayOfStrings) { + this.result = result; + this.jsonArrayOfStrings = jsonArrayOfStrings; + } + + + public void connect(ReactiveAdapter adapter, Object returnValue) { + Publisher 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 { + } + +} \ No newline at end of file diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/RequestMappingHandlerAdapter.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/RequestMappingHandlerAdapter.java index d15025470d..b97103a155 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/RequestMappingHandlerAdapter.java +++ b/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)); diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandler.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandler.java index ad461f1e2a..33cbeda02c 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandler.java +++ b/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> messageConverters; + private final ReactiveTypeHandler reactiveHandler; + + + public ResponseBodyEmitterReturnValueHandler(List> messageConverters, + ReactiveAdapterRegistry reactiveRegistry, ContentNegotiationManager manager) { - public ResponseBodyEmitterReturnValueHandler(List> 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); diff --git a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethod.java b/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethod.java index d1f7a1e009..4628849123 100644 --- a/spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethod.java +++ b/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) { diff --git a/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandlerTests.java b/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandlerTests.java new file mode 100644 index 0000000000..374dc7341b --- /dev/null +++ b/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandlerTests.java @@ -0,0 +1,412 @@ +/* + * 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.Arrays; +import java.util.Collections; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; + +import org.junit.Before; +import org.junit.Test; +import reactor.core.publisher.EmitterProcessor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.publisher.MonoProcessor; +import rx.Single; +import rx.SingleEmitter; + +import org.springframework.core.MethodParameter; +import org.springframework.core.ReactiveAdapterRegistry; +import org.springframework.core.ResolvableType; +import org.springframework.http.MediaType; +import org.springframework.http.codec.ServerSentEvent; +import org.springframework.http.server.ServletServerHttpResponse; +import org.springframework.mock.web.test.MockHttpServletRequest; +import org.springframework.mock.web.test.MockHttpServletResponse; +import org.springframework.web.accept.ContentNegotiationManager; +import org.springframework.web.accept.ContentNegotiationManagerFactoryBean; +import org.springframework.web.context.request.NativeWebRequest; +import org.springframework.web.context.request.ServletWebRequest; +import org.springframework.web.context.request.async.AsyncWebRequest; +import org.springframework.web.context.request.async.StandardServletAsyncWebRequest; +import org.springframework.web.context.request.async.WebAsyncUtils; +import org.springframework.web.method.support.ModelAndViewContainer; +import org.springframework.web.servlet.HandlerMapping; + +import static junit.framework.TestCase.assertNull; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.springframework.web.method.ResolvableMethod.on; + +/** + * Unit tests for {@link ReactiveTypeHandler}. + * @author Rossen Stoyanchev + */ +public class ReactiveTypeHandlerTests { + + private ReactiveTypeHandler handler; + + private MockHttpServletRequest servletRequest; + + private MockHttpServletResponse servletResponse; + + private NativeWebRequest webRequest; + + + @Before + public void setup() throws Exception { + ContentNegotiationManagerFactoryBean factoryBean = new ContentNegotiationManagerFactoryBean(); + factoryBean.afterPropertiesSet(); + ContentNegotiationManager manager = factoryBean.getObject(); + this.handler = new ReactiveTypeHandler(new ReactiveAdapterRegistry(), manager); + resetRequest(); + } + + private void resetRequest() { + this.servletRequest = new MockHttpServletRequest(); + this.servletResponse = new MockHttpServletResponse(); + this.webRequest = new ServletWebRequest(this.servletRequest, this.servletResponse); + + AsyncWebRequest asyncWebRequest = new StandardServletAsyncWebRequest(this.servletRequest, this.servletResponse); + WebAsyncUtils.getAsyncManager(this.webRequest).setAsyncWebRequest(asyncWebRequest); + this.servletRequest.setAsyncSupported(true); + } + + + @Test + public void supportsType() throws Exception { + assertTrue(this.handler.isReactiveType(Mono.class)); + assertTrue(this.handler.isReactiveType(Single.class)); + assertTrue(this.handler.isReactiveType(io.reactivex.Single.class)); + } + + @Test + public void doesNotSupportType() throws Exception { + assertFalse(this.handler.isReactiveType(String.class)); + } + + @Test + public void deferredResultSubscriberWithOneValue() throws Exception { + + // Mono + MonoProcessor mono = MonoProcessor.create(); + testDeferredResultSubscriber(mono, Mono.class, () -> mono.onNext("foo"), "foo"); + + // Mono empty + MonoProcessor monoEmpty = MonoProcessor.create(); + testDeferredResultSubscriber(monoEmpty, Mono.class, monoEmpty::onComplete, null); + + // RxJava 1 Single + AtomicReference> ref = new AtomicReference<>(); + Single single = Single.fromEmitter(ref::set); + testDeferredResultSubscriber(single, Single.class, () -> ref.get().onSuccess("foo"), "foo"); + + // RxJava 2 Single + AtomicReference> ref2 = new AtomicReference<>(); + io.reactivex.Single single2 = io.reactivex.Single.create(ref2::set); + testDeferredResultSubscriber(single2, io.reactivex.Single.class, () -> ref2.get().onSuccess("foo"), "foo"); + } + + @Test + public void deferredResultSubscriberWithNoValues() throws Exception { + + // Empty -> null + MonoProcessor monoEmpty = MonoProcessor.create(); + testDeferredResultSubscriber(monoEmpty, Mono.class, monoEmpty::onComplete, null); + + // Empty -> List[0] when JSON is preferred + this.servletRequest.addHeader("Accept", "application/json"); + MonoProcessor monoEmpty2 = MonoProcessor.create(); + testDeferredResultSubscriber(monoEmpty2, Mono.class, monoEmpty2::onComplete, new ArrayList<>()); + } + + @Test + public void deferredResultSubscriberWithMultipleValues() throws Exception { + + // JSON must be preferred for Flux -> List or else we stream + this.servletRequest.addHeader("Accept", "application/json"); + + EmitterProcessor emitter = EmitterProcessor.create(); + testDeferredResultSubscriber(emitter, Flux.class, () -> { + emitter.connect(); + emitter.onNext("foo"); + emitter.onNext("bar"); + emitter.onNext("baz"); + emitter.onComplete(); + }, Arrays.asList("foo", "bar", "baz")); + } + + @Test + public void deferredResultSubscriberWithError() throws Exception { + + IllegalStateException ex = new IllegalStateException(); + + // Mono + MonoProcessor mono = MonoProcessor.create(); + testDeferredResultSubscriber(mono, Mono.class, () -> mono.onError(ex), ex); + + // RxJava 1 Single + AtomicReference> ref = new AtomicReference<>(); + Single single = Single.fromEmitter(ref::set); + testDeferredResultSubscriber(single, Single.class, () -> ref.get().onError(ex), ex); + + // RxJava 2 Single + AtomicReference> ref2 = new AtomicReference<>(); + io.reactivex.Single single2 = io.reactivex.Single.create(ref2::set); + testDeferredResultSubscriber(single2, io.reactivex.Single.class, () -> ref2.get().onError(ex), ex); + } + + @Test + public void jsonArrayOfStrings() throws Exception { + + // Empty -> null + testJsonPreferred("text/plain", null); + testJsonPreferred("text/plain, application/json", null); + testJsonPreferred("text/markdown", null); + testJsonPreferred("foo/bar", null); + + // Empty -> List[0] when JSON is preferred + testJsonPreferred("application/json", Collections.emptyList()); + testJsonPreferred("application/foo+json", Collections.emptyList()); + testJsonPreferred("application/json, text/plain", Collections.emptyList()); + testJsonPreferred("*/*, application/json, text/plain", Collections.emptyList()); + } + + private void testJsonPreferred(String acceptHeaderValue, Object expected) throws Exception { + resetRequest(); + this.servletRequest.addHeader("Accept", acceptHeaderValue); + MonoProcessor mono = MonoProcessor.create(); + testDeferredResultSubscriber(mono, Mono.class, mono::onComplete, expected); + } + + @Test + public void mediaTypes() throws Exception { + + // Media type from request + this.servletRequest.addHeader("Accept", "text/event-stream"); + testSseResponse(true); + + // Media type from "produces" attribute + Set types = Collections.singleton(MediaType.TEXT_EVENT_STREAM); + this.servletRequest.setAttribute(HandlerMapping.PRODUCIBLE_MEDIA_TYPES_ATTRIBUTE, types); + testSseResponse(true); + + // No media type preferences + testSseResponse(false); + + // Requested media types are sorted + testJsonPreferred("text/plain;q=0.8, application/json;q=1.0", Collections.emptyList()); + testJsonPreferred("text/plain, application/json", null); + } + + private void testSseResponse(boolean expectSseEimtter) throws Exception { + ResponseBodyEmitter emitter = handleValue(Flux.empty(), Flux.class); + assertEquals(expectSseEimtter, emitter instanceof SseEmitter); + resetRequest(); + } + + @Test + public void writeServerSentEvents() throws Exception { + + this.servletRequest.addHeader("Accept", "text/event-stream"); + EmitterProcessor processor = EmitterProcessor.create(); + SseEmitter sseEmitter = (SseEmitter) handleValue(processor, Flux.class); + + EmitterHandler emitterHandler = new EmitterHandler(); + sseEmitter.initialize(emitterHandler); + + processor.connect(); + processor.onNext("foo"); + processor.onNext("bar"); + processor.onNext("baz"); + processor.onComplete(); + + assertEquals("data:foo\n\ndata:bar\n\ndata:baz\n\n", emitterHandler.getOutput()); + } + + @Test + public void writeSentEventsWithBuilder() throws Exception { + + ResolvableType type = ResolvableType.forClassWithGenerics(ServerSentEvent.class, String.class); + + EmitterProcessor> processor = EmitterProcessor.create(); + SseEmitter sseEmitter = (SseEmitter) handleValue(processor, Flux.class, type); + + EmitterHandler emitterHandler = new EmitterHandler(); + sseEmitter.initialize(emitterHandler); + + processor.connect(); + processor.onNext(ServerSentEvent.builder("foo").id("1").build()); + processor.onNext(ServerSentEvent.builder("bar").id("2").build()); + processor.onNext(ServerSentEvent.builder("baz").id("3").build()); + processor.onComplete(); + + assertEquals("id:1\ndata:foo\n\nid:2\ndata:bar\n\nid:3\ndata:baz\n\n", + emitterHandler.getOutput()); + } + + @Test + public void writeStreamJson() throws Exception { + + this.servletRequest.addHeader("Accept", "application/stream+json"); + + EmitterProcessor processor = EmitterProcessor.create(); + ResponseBodyEmitter emitter = handleValue(processor, Flux.class); + + EmitterHandler emitterHandler = new EmitterHandler(); + emitter.initialize(emitterHandler); + + ServletServerHttpResponse message = new ServletServerHttpResponse(this.servletResponse); + emitter.extendResponse(message); + + processor.connect(); + processor.onNext("[\"foo\",\"bar\"]"); + processor.onNext("[\"bar\",\"baz\"]"); + processor.onComplete(); + + assertEquals("application/stream+json", message.getHeaders().getContentType().toString()); + assertEquals("[\"foo\",\"bar\"]\n[\"bar\",\"baz\"]\n", emitterHandler.getOutput()); + } + + @Test + public void writeText() throws Exception { + + EmitterProcessor processor = EmitterProcessor.create(); + ResponseBodyEmitter emitter = handleValue(processor, Flux.class); + + EmitterHandler emitterHandler = new EmitterHandler(); + emitter.initialize(emitterHandler); + + processor.connect(); + processor.onNext("The quick"); + processor.onNext(" brown fox jumps over "); + processor.onNext("the lazy dog"); + processor.onComplete(); + + assertEquals("The quick brown fox jumps over the lazy dog", emitterHandler.getOutput()); + } + + @Test + public void writeTextContentType() throws Exception { + + // Any requested, concrete, "text" media type + this.servletRequest.addHeader("Accept", "*/*, text/*, text/markdown"); + testEmitterContentType("text/markdown"); + + // Or any requested concrete media type + this.servletRequest.addHeader("Accept", "*/*, text/*, foo/bar"); + testEmitterContentType("foo/bar"); + + // Or default to... + testEmitterContentType("text/plain"); + + // Or default to if not concrete.. + this.servletRequest.addHeader("Accept", "text/*"); + testEmitterContentType("text/plain"); + } + + private void testEmitterContentType(String expected) throws Exception { + ServletServerHttpResponse message = new ServletServerHttpResponse(this.servletResponse); + ResponseBodyEmitter emitter = handleValue(Flux.empty(), Flux.class); + emitter.extendResponse(message); + assertEquals(expected, message.getHeaders().getContentType().toString()); + resetRequest(); + } + + + private void testDeferredResultSubscriber(Object returnValue, Class asyncType, + Runnable produceTask, Object expected) throws Exception { + + ResponseBodyEmitter emitter = handleValue(returnValue, asyncType); + assertNull(emitter); + + assertTrue(this.servletRequest.isAsyncStarted()); + assertFalse(WebAsyncUtils.getAsyncManager(this.webRequest).hasConcurrentResult()); + + produceTask.run(); + + assertTrue(WebAsyncUtils.getAsyncManager(this.webRequest).hasConcurrentResult()); + assertEquals(expected, WebAsyncUtils.getAsyncManager(this.webRequest).getConcurrentResult()); + + resetRequest(); + } + + private ResponseBodyEmitter handleValue(Object returnValue, Class asyncType) throws Exception { + return handleValue(returnValue, asyncType, ResolvableType.forClass(String.class)); + } + + private ResponseBodyEmitter handleValue(Object returnValue, Class asyncType, + ResolvableType genericType) throws Exception { + + ModelAndViewContainer mavContainer = new ModelAndViewContainer(); + MethodParameter returnType = on(TestController.class).resolveReturnType(asyncType, genericType); + return this.handler.handleValue(returnValue, returnType, mavContainer, this.webRequest); + } + + + @SuppressWarnings("unused") + static class TestController { + + String handleString() { return null; } + + Mono handleMono() { return null; } + + Single handleSingle() { return null; } + + io.reactivex.Single handleSingleRxJava2() { return null; } + + Flux handleFlux() { return null; } + + Flux> handleFluxSseEventBuilder() { return null; } + } + + + private static class EmitterHandler implements ResponseBodyEmitter.Handler { + + private final StringBuilder stringBuilder = new StringBuilder(); + + + public String getOutput() { + return this.stringBuilder.toString(); + } + + @Override + public void send(Object data, MediaType mediaType) throws IOException { + this.stringBuilder.append(data); + } + + @Override + public void complete() { + } + + @Override + public void completeWithError(Throwable failure) { + } + + @Override + public void onTimeout(Runnable callback) { + } + + @Override + public void onCompletion(Runnable callback) { + } + } + +} diff --git a/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandlerTests.java b/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandlerTests.java index c572b33e73..e51dceba71 100644 --- a/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandlerTests.java +++ b/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterReturnValueHandlerTests.java @@ -25,6 +25,7 @@ import org.junit.Before; import org.junit.Test; import org.springframework.core.MethodParameter; +import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.http.ResponseEntity; import org.springframework.http.converter.HttpMessageConverter; import org.springframework.http.converter.StringHttpMessageConverter; @@ -32,6 +33,7 @@ import org.springframework.http.converter.json.MappingJackson2HttpMessageConvert import org.springframework.mock.web.test.MockAsyncContext; import org.springframework.mock.web.test.MockHttpServletRequest; import org.springframework.mock.web.test.MockHttpServletResponse; +import org.springframework.web.accept.ContentNegotiationManager; import org.springframework.web.context.request.NativeWebRequest; import org.springframework.web.context.request.ServletWebRequest; import org.springframework.web.context.request.async.AsyncWebRequest; @@ -63,7 +65,9 @@ public class ResponseBodyEmitterReturnValueHandlerTests { List> converters = Arrays.asList( new StringHttpMessageConverter(), new MappingJackson2HttpMessageConverter()); - this.handler = new ResponseBodyEmitterReturnValueHandler(converters); + ReactiveAdapterRegistry registry = new ReactiveAdapterRegistry(); + ContentNegotiationManager manager = new ContentNegotiationManager(); + this.handler = new ResponseBodyEmitterReturnValueHandler(converters, registry, manager); this.request = new MockHttpServletRequest(); this.response = new MockHttpServletResponse(); this.webRequest = new ServletWebRequest(this.request, this.response); diff --git a/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethodTests.java b/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethodTests.java index b2e886a9bc..6a33827f84 100644 --- a/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethodTests.java +++ b/spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ServletInvocableHandlerMethodTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * 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. @@ -19,22 +19,26 @@ package org.springframework.web.servlet.mvc.method.annotation; import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.reflect.Method; -import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import javax.servlet.http.HttpServletResponse; import org.junit.Test; +import reactor.core.publisher.Flux; import org.springframework.core.MethodParameter; +import org.springframework.core.ReactiveAdapterRegistry; import org.springframework.core.annotation.AliasFor; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.http.converter.HttpMessageConverter; import org.springframework.http.converter.HttpMessageNotWritableException; import org.springframework.http.converter.StringHttpMessageConverter; +import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter; import org.springframework.mock.web.test.MockHttpServletRequest; import org.springframework.mock.web.test.MockHttpServletResponse; +import org.springframework.web.accept.ContentNegotiationManager; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseStatus; @@ -48,8 +52,10 @@ import org.springframework.web.method.support.HandlerMethodReturnValueHandlerCom import org.springframework.web.method.support.ModelAndViewContainer; import org.springframework.web.servlet.view.RedirectView; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; /** * Test fixture with {@link ServletInvocableHandlerMethod}. @@ -60,9 +66,14 @@ import static org.mockito.Mockito.*; */ public class ServletInvocableHandlerMethodTests { - private final HandlerMethodArgumentResolverComposite argumentResolvers = new HandlerMethodArgumentResolverComposite(); + private final List> converters = + Collections.singletonList(new StringHttpMessageConverter()); - private final HandlerMethodReturnValueHandlerComposite returnValueHandlers = new HandlerMethodReturnValueHandlerComposite(); + private final HandlerMethodArgumentResolverComposite argumentResolvers = + new HandlerMethodArgumentResolverComposite(); + + private final HandlerMethodReturnValueHandlerComposite returnValueHandlers = + new HandlerMethodReturnValueHandlerComposite(); private final ModelAndViewContainer mavContainer = new ModelAndViewContainer(); @@ -199,10 +210,8 @@ public class ServletInvocableHandlerMethodTests { private void wrapConcurrentResult_ResponseBody(Object handler, Object result, Class expectedReturnType) throws Exception { - List> converters = new ArrayList<>(); - converters.add(new StringHttpMessageConverter()); this.returnValueHandlers.addHandler(new ModelAndViewMethodReturnValueHandler()); - this.returnValueHandlers.addHandler(new RequestResponseBodyMethodProcessor(converters)); + this.returnValueHandlers.addHandler(new RequestResponseBodyMethodProcessor(this.converters)); ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(handler, "handle"); handlerMethod = handlerMethod.wrapConcurrentResult(result); @@ -213,9 +222,7 @@ public class ServletInvocableHandlerMethodTests { @Test public void wrapConcurrentResult_ResponseEntity() throws Exception { - List> converters = new ArrayList<>(); - converters.add(new StringHttpMessageConverter()); - this.returnValueHandlers.addHandler(new HttpEntityMethodProcessor(converters)); + this.returnValueHandlers.addHandler(new HttpEntityMethodProcessor(this.converters)); ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new ResponseEntityHandler(), "handleDeferred"); handlerMethod = handlerMethod.wrapConcurrentResult(new ResponseEntity<>("bar", HttpStatus.OK)); handlerMethod.invokeAndHandle(this.webRequest, this.mavContainer); @@ -225,11 +232,7 @@ public class ServletInvocableHandlerMethodTests { @Test // SPR-12287 public void wrapConcurrentResult_ResponseEntityNullBody() throws Exception { - List> converters = new ArrayList<>(); - converters.add(new StringHttpMessageConverter()); - List advice = Collections.singletonList(mock(ResponseBodyAdvice.class)); - HttpEntityMethodProcessor processor = new HttpEntityMethodProcessor(converters, null, advice); - this.returnValueHandlers.addHandler(processor); + this.returnValueHandlers.addHandler(new HttpEntityMethodProcessor(this.converters)); ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new ResponseEntityHandler(), "handleDeferred"); handlerMethod = handlerMethod.wrapConcurrentResult(new ResponseEntity<>(HttpStatus.OK)); handlerMethod.invokeAndHandle(this.webRequest, this.mavContainer); @@ -240,11 +243,7 @@ public class ServletInvocableHandlerMethodTests { @Test public void wrapConcurrentResult_ResponseEntityNullReturnValue() throws Exception { - List> converters = new ArrayList<>(); - converters.add(new StringHttpMessageConverter()); - List advice = Collections.singletonList(mock(ResponseBodyAdvice.class)); - HttpEntityMethodProcessor processor = new HttpEntityMethodProcessor(converters, null, advice); - this.returnValueHandlers.addHandler(processor); + this.returnValueHandlers.addHandler(new HttpEntityMethodProcessor(this.converters)); ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new ResponseEntityHandler(), "handleDeferred"); handlerMethod = handlerMethod.wrapConcurrentResult(null); handlerMethod.invokeAndHandle(this.webRequest, this.mavContainer); @@ -255,10 +254,10 @@ public class ServletInvocableHandlerMethodTests { @Test public void wrapConcurrentResult_ResponseBodyEmitter() throws Exception { - List> converters = new ArrayList<>(); - converters.add(new StringHttpMessageConverter()); - this.returnValueHandlers.addHandler(new ResponseBodyEmitterReturnValueHandler(converters)); - ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new AsyncHandler(), "handleWithEmitter"); + ReactiveAdapterRegistry registry = new ReactiveAdapterRegistry(); + ContentNegotiationManager manager = new ContentNegotiationManager(); + this.returnValueHandlers.addHandler(new ResponseBodyEmitterReturnValueHandler(this.converters, registry, manager)); + ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new StreamingHandler(), "handleEmitter"); handlerMethod = handlerMethod.wrapConcurrentResult(null); handlerMethod.invokeAndHandle(this.webRequest, this.mavContainer); @@ -269,7 +268,7 @@ public class ServletInvocableHandlerMethodTests { @Test public void wrapConcurrentResult_StreamingResponseBody() throws Exception { this.returnValueHandlers.addHandler(new StreamingResponseBodyReturnValueHandler()); - ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new AsyncHandler(), "handleWithStreaming"); + ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new StreamingHandler(), "handleStreamBody"); handlerMethod = handlerMethod.wrapConcurrentResult(null); handlerMethod.invokeAndHandle(this.webRequest, this.mavContainer); @@ -277,12 +276,27 @@ public class ServletInvocableHandlerMethodTests { assertEquals("", this.response.getContentAsString()); } + @Test + public void wrapConcurrentResult_CollectedValuesList() throws Exception { + List> converters = Collections.singletonList(new MappingJackson2HttpMessageConverter()); + this.request.addHeader("Accept", "application/json"); + ReactiveTypeHandler.CollectedValuesList result = new ReactiveTypeHandler.CollectedValuesList(); + result.add(Arrays.asList("foo1", "bar1")); + result.add(Arrays.asList("foo2", "bar2")); + + ContentNegotiationManager manager = new ContentNegotiationManager(); + this.returnValueHandlers.addHandler(new RequestResponseBodyMethodProcessor(converters, manager)); + ServletInvocableHandlerMethod hm = getHandlerMethod(new MethodLevelResponseBodyHandler(), "handleFluxOfLists"); + hm = hm.wrapConcurrentResult(result); + hm.invokeAndHandle(this.webRequest, this.mavContainer); + + assertEquals(200, this.response.getStatus()); + assertEquals("[[\"foo1\",\"bar1\"],[\"foo2\",\"bar2\"]]", this.response.getContentAsString()); + } + @Test // SPR-12287 (16/Oct/14 comments) public void responseEntityRawTypeWithNullBody() throws Exception { - List> converters = Collections.singletonList(new StringHttpMessageConverter()); - List advice = Collections.singletonList(mock(ResponseBodyAdvice.class)); - HttpEntityMethodProcessor processor = new HttpEntityMethodProcessor(converters, null, advice); - this.returnValueHandlers.addHandler(processor); + this.returnValueHandlers.addHandler(new HttpEntityMethodProcessor(this.converters)); ServletInvocableHandlerMethod handlerMethod = getHandlerMethod(new ResponseEntityHandler(), "handleRawType"); handlerMethod.invokeAndHandle(this.webRequest, this.mavContainer); @@ -358,6 +372,12 @@ public class ServletInvocableHandlerMethodTests { public DeferredResult handle() { return new DeferredResult<>(); } + + // Unusual but legal return type + // Properly test generic type handling of Flux values collected to a List + + @ResponseBody + public Flux> handleFluxOfLists() { return null; } } @@ -413,15 +433,12 @@ public class ServletInvocableHandlerMethodTests { @SuppressWarnings("unused") - private static class AsyncHandler { + private static class StreamingHandler { - public ResponseBodyEmitter handleWithEmitter() { - return null; - } + public ResponseBodyEmitter handleEmitter() { return null; } + + public StreamingResponseBody handleStreamBody() { return null; } - public StreamingResponseBody handleWithStreaming() { - return null; - } } }