From bca86191b366b1fcb1cb737a5814097ee0ad5d0d Mon Sep 17 00:00:00 2001 From: Jakub Narloch Date: Tue, 26 Jan 2016 23:57:28 +0100 Subject: [PATCH] Spring Mvc RxJava integration. Support for Single via a ReturnValueHandler. Support for Observable via SseEmitter. --- README.adoc | 2 +- .../main/asciidoc/spring-cloud-netflix.adoc | 33 ++++ .../netflix/rx/DeferredResultSubscriber.java | 68 +++++++ .../rx/ObservableReturnValueHandler.java | 75 -------- .../netflix/rx/ObservableSseEmitter.java | 45 +++++ .../rx/ResponseBodyEmitterSubscriber.java | 90 +++++++++ .../netflix/rx/RxJavaAutoConfiguration.java | 15 +- .../cloud/netflix/rx/RxResponse.java | 58 ++++++ .../netflix/rx/SingleDeferredResult.java | 50 +++++ .../netflix/rx/SingleReturnValueHandler.java | 117 ++++++++++++ .../rx/ObservableReturnValueHandlerTest.java | 174 ++++++++++++++++++ .../rx/ObservableReturnValueHandlerTests.java | 71 ------- .../netflix/rx/ObservableSseEmitterTest.java | 165 +++++++++++++++++ .../rx/SingleReturnValueHandlerTest.java | 120 ++++++++++++ 14 files changed, 929 insertions(+), 154 deletions(-) create mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/DeferredResultSubscriber.java delete mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandler.java create mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableSseEmitter.java create mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ResponseBodyEmitterSubscriber.java create mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxResponse.java create mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleDeferredResult.java create mode 100644 spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandler.java create mode 100644 spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTest.java delete mode 100644 spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTests.java create mode 100644 spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableSseEmitterTest.java create mode 100644 spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandlerTest.java diff --git a/README.adoc b/README.adoc index df695e27..a34519e3 100644 --- a/README.adoc +++ b/README.adoc @@ -139,7 +139,7 @@ or you will encounter compile problems. It also needs a specific version of mave enabled. Intellij 14.1+ requires some configuration to ensure these are setup properly. 1. Click Preferences, Plugins. *Ensure Lombok is installed* - 2. Click New, Project from Existing Sources, choose your spring-cloud-sleuth directory + 2. Click New, Project from Existing Sources, choose your spring-cloud project directory 3. Choose Maven, and select Environment Settings. *Ensure you are using Maven 3.3.3* 4. In the next screen, *Select the profile `spring`* click Next until Finish. 5. Click Preferences, "Build, Execution, Deployment", Compiler, Annotation Processors. *Click Enable Annotation Processing* diff --git a/docs/src/main/asciidoc/spring-cloud-netflix.adoc b/docs/src/main/asciidoc/spring-cloud-netflix.adoc index 651beb1c..0b04e4fa 100644 --- a/docs/src/main/asciidoc/spring-cloud-netflix.adoc +++ b/docs/src/main/asciidoc/spring-cloud-netflix.adoc @@ -1483,6 +1483,36 @@ info: url: https://github.com/spring-cloud-samples ---- +[[netflix-rxjava-springmvc]] +== RxJava with Spring MVC +Spring Cloud Netflix includes the https://github.com/ReactiveX/RxJava[RxJava]. + +> RxJava is a Java VM implementation of http://reactivex.io/[Reactive Extensions]: a library for composing asynchronous and event-based programs by using observable sequences. + +Spring Cloud Netflix provides support for returning `rx.Single` objects from Spring MVC Controllers. It also supports using `rx.Observable` objects for https://en.wikipedia.org/wiki/Server-sent_events[Server-sent events (SSE)]. This can be very convenient if your internal APIs are already built using RxJava (see <> for examples). + +Here are some examples of using `rx.Single`: + +[source,java] +---- +include::../../../../spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandlerTest.java[tags=rx_single,indent=0] +---- + +If you have an `Observable`, rather than a single, you can use `.toSingle()` or `.toList().toSingle()`. Here are some examples: + +[source,java] +---- +include::../../../../spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTest.java[tags=rx_observable,indent=0] +---- + +If you have a streaming endpoint and client, SSE could be an option. To convert `rx.Observable` to a Spring `SseEmitter` use `RxResponse.sse()`. Here are some examples: + +[source,java] +---- +include::../../../../spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableSseEmitterTest.java[tags=rx_observable_sse,indent=0] +---- + +[[netflix-metrics]] == Metrics: Spectator, Servo, and Atlas When used together, Spectator/Servo and Atlas provide a near real-time operational insight platform. @@ -1545,6 +1575,7 @@ If Spring AOP is enabled and `org.aspectj:aspectjweaver` is present on your runt 3. URI, sanitized for Atlas 4. Client name +[[netflix-metrics-spectator]] === Metrics Collection: Spectator To enable Spectator metrics, include a dependency on `spring-boot-starter-spectator`: @@ -1626,6 +1657,7 @@ DistributionSummary ds = registry.distributionSummary("dsName", "tagKey1", "tagV ds.record(request.sizeInBytes()); ---- +[[netflix-metrics-servo]] === Metrics Collection: Servo WARNING: If your code is compiled on Java 8, please use Spectator instead of Servo as Spectator is destined to replace Servo entirely in the long term. @@ -1649,6 +1681,7 @@ Timer timer = new BasicTimer(config); monitorRegistry.register(timer); ---- +[[netflix-metrics-atlas]] === Metrics Backend: Atlas Atlas was developed by Netflix to manage dimensional time series data for near real-time operational insight. Atlas features in-memory data storage, allowing it to gather and report very large numbers of metrics, very quickly. diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/DeferredResultSubscriber.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/DeferredResultSubscriber.java new file mode 100644 index 00000000..e84e6742 --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/DeferredResultSubscriber.java @@ -0,0 +1,68 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import org.springframework.web.context.request.async.DeferredResult; + +import rx.Observable; +import rx.Subscriber; +import rx.Subscription; + +/** + * A subscriber that sets the single value produced by the {@link Observable} on the {@link DeferredResult}. + * + * @author Jakub Narloch + * @see DeferredResult + */ +class DeferredResultSubscriber extends Subscriber implements Runnable { + + private final DeferredResult deferredResult; + + private final Subscription subscription; + + private boolean completed; + + public DeferredResultSubscriber(Observable observable, DeferredResult deferredResult) { + + this.deferredResult = deferredResult; + this.deferredResult.onTimeout(this); + this.deferredResult.onCompletion(this); + this.subscription = observable.subscribe(this); + } + + @Override + public void onNext(T value) { + if (!completed) { + deferredResult.setResult(value); + } + } + + @Override + public void onError(Throwable e) { + deferredResult.setErrorResult(e); + } + + @Override + public void onCompleted() { + completed = true; + } + + @Override + public void run() { + this.subscription.unsubscribe(); + } +} diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandler.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandler.java deleted file mode 100644 index 2a95022a..00000000 --- a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandler.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Copyright 2013-2015 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.cloud.netflix.rx; - -import org.springframework.core.MethodParameter; -import org.springframework.web.context.request.NativeWebRequest; -import org.springframework.web.context.request.async.DeferredResult; -import org.springframework.web.context.request.async.WebAsyncUtils; -import org.springframework.web.method.support.AsyncHandlerMethodReturnValueHandler; -import org.springframework.web.method.support.ModelAndViewContainer; - -import rx.Observable; -import rx.functions.Action1; - -/** - * MVC handler for return values of type {@link rx.Observable}. - * - * @author Spencer Gibb - */ -public class ObservableReturnValueHandler - implements AsyncHandlerMethodReturnValueHandler { - - @Override - public boolean isAsyncReturnValue(Object returnValue, MethodParameter returnType) { - return returnValue != null && returnValue instanceof Observable; - } - - @Override - public boolean supportsReturnType(MethodParameter returnType) { - return Observable.class.isAssignableFrom(returnType.getParameterType()); - } - - @Override - public void handleReturnValue(Object returnValue, MethodParameter returnType, - ModelAndViewContainer mavContainer, NativeWebRequest webRequest) - throws Exception { - if (returnValue == null) { - mavContainer.setRequestHandled(true); - return; - } - - Observable observable = Observable.class.cast(returnValue); - - final DeferredResult deferredResult = new DeferredResult<>(); - - observable.subscribe(new Action1() { - @Override - public void call(Object o) { - deferredResult.setResult(o); - } - }, new Action1() { - @Override - public void call(Throwable throwable) { - deferredResult.setErrorResult(throwable); - } - }); - - WebAsyncUtils.getAsyncManager(webRequest) - .startDeferredResultProcessing(deferredResult, mavContainer); - } -} diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableSseEmitter.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableSseEmitter.java new file mode 100644 index 00000000..14a5ae58 --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ObservableSseEmitter.java @@ -0,0 +1,45 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import org.springframework.http.MediaType; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import rx.Observable; + +/** + * A specialized {@link SseEmitter} that handles {@link Observable} return types. + * + * @author Jakub Narloch + * @see SseEmitter + */ +class ObservableSseEmitter extends SseEmitter { + + private final ResponseBodyEmitterSubscriber subscriber; + + public ObservableSseEmitter(Observable observable) { + this(null, observable); + } + + public ObservableSseEmitter(MediaType mediaType, Observable observable) { + this(null, mediaType, observable); + } + + public ObservableSseEmitter(Long timeout, MediaType mediaType, Observable observable) { + super(timeout); + this.subscriber = new ResponseBodyEmitterSubscriber<>(mediaType, observable, this); + } +} diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ResponseBodyEmitterSubscriber.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ResponseBodyEmitterSubscriber.java new file mode 100644 index 00000000..d00eb8c9 --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/ResponseBodyEmitterSubscriber.java @@ -0,0 +1,90 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import java.io.IOException; + +import org.springframework.http.MediaType; +import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter; + +import rx.Observable; +import rx.Subscriber; +import rx.Subscription; + + +/** + * Subscriber that emits any value produced by the {@link Observable} into the delegated {@link ResponseBodyEmitter}. + * + * @author Jakub Narloch + */ +class ResponseBodyEmitterSubscriber extends Subscriber implements Runnable { + + private final MediaType mediaType; + + private final Subscription subscription; + + private final ResponseBodyEmitter responseBodyEmitter; + + private boolean completed; + + /** + * Creates new instance of {@link ResponseBodyEmitterSubscriber} with response media type, observable and response + * emitter. + * + * @param mediaType the marshaled object media type + * @param observable the observable + * @param responseBodyEmitter the response emitter + */ + public ResponseBodyEmitterSubscriber(MediaType mediaType, Observable observable, ResponseBodyEmitter responseBodyEmitter) { + + this.mediaType = mediaType; + this.responseBodyEmitter = responseBodyEmitter; + this.responseBodyEmitter.onTimeout(this); + this.responseBodyEmitter.onCompletion(this); + this.subscription = observable.subscribe(this); + } + + @Override + public void onNext(T value) { + + try { + if(!completed) { + responseBodyEmitter.send(value, mediaType); + } + } catch (IOException e) { + throw new RuntimeException(e.getMessage(), e); + } + } + + @Override + public void onError(Throwable e) { + responseBodyEmitter.completeWithError(e); + } + + @Override + public void onCompleted() { + if(!completed) { + completed = true; + responseBodyEmitter.complete(); + } + } + + @Override + public void run() { + subscription.unsubscribe(); + } +} diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxJavaAutoConfiguration.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxJavaAutoConfiguration.java index 574e90ba..d6c704b5 100644 --- a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxJavaAutoConfiguration.java +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxJavaAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2013-2015 the original author or authors. + * Copyright 2013-2016 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. @@ -28,6 +28,7 @@ import org.springframework.web.method.support.HandlerMethodReturnValueHandler; import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter; import rx.Observable; +import rx.Single; /** * @author Spencer Gibb @@ -39,25 +40,25 @@ public class RxJavaAutoConfiguration { @Configuration @ConditionalOnClass(AsyncHandlerMethodReturnValueHandler.class) - protected static class ObservableReturnValueHandlerConfig { + protected static class RxJavaReturnValueHandlerConfig { @Bean - public ObservableReturnValueHandler observableReturnValueHandler() { - return new ObservableReturnValueHandler(); + public SingleReturnValueHandler singleReturnValueHandler() { + return new SingleReturnValueHandler(); } @Bean - public WebMvcConfigurerAdapter observableMVCConfiguration() { + public WebMvcConfigurerAdapter observableMVCConfiguration(final SingleReturnValueHandler singleReturnValueHandler) { return new WebMvcConfigurerAdapter() { @Override public void addReturnValueHandlers(List returnValueHandlers) { - returnValueHandlers.add(observableReturnValueHandler()); + returnValueHandlers.add(singleReturnValueHandler); } }; } @Bean public HasFeatures rxFeature() { - return HasFeatures.namedFeature("MVC Observable", Observable.class); + return HasFeatures.namedFeatures("MVC Observable", Observable.class, "MVC Single", Single.class); } } } diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxResponse.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxResponse.java new file mode 100644 index 00000000..48a966a3 --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/RxResponse.java @@ -0,0 +1,58 @@ +package org.springframework.cloud.netflix.rx; + +import org.springframework.http.MediaType; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import rx.Observable; + +/** + * A convenient class allowing to wrap either the {@link Observable} into a response supported by the + * Spring MVC. + * + * @author Jakub Narloch + */ +public final class RxResponse { + + private RxResponse() { + + } + + /** + * Wraps the {@link Observable} into a {@link SseEmitter}. Every value produced by the observable will be emitted + * as server side event. + * + * @param observable the observable instance + * @param the result type + * @return the sse emitter + */ + public static SseEmitter sse(Observable observable) { + return new ObservableSseEmitter<>(observable); + } + + /** + * Wraps the {@link Observable} into a {@link SseEmitter}. Every value produced by the observable will be emitted + * as server side event. + * + * @param mediaType the media type of produced entry + * @param observable the observable instance + * @param the result type + * @return the sse emitter + */ + public static SseEmitter sse(MediaType mediaType, Observable observable) { + return new ObservableSseEmitter<>(mediaType, observable); + } + + /** + * Wraps the {@link Observable} into a {@link SseEmitter}. Every value produced by the observable will be emitted + * as server side event. + * + * @param timeout the response timeout + * @param mediaType the media type of produced entry + * @param observable the observable instance + * @param the result type + * @return the sse emitter + */ + public static SseEmitter sse(long timeout, MediaType mediaType, Observable observable) { + return new ObservableSseEmitter<>(timeout, mediaType, observable); + } +} diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleDeferredResult.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleDeferredResult.java new file mode 100644 index 00000000..af313392 --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleDeferredResult.java @@ -0,0 +1,50 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import org.springframework.util.Assert; +import org.springframework.web.context.request.async.DeferredResult; + +import rx.Single; + +/** + * A specialized {@link DeferredResult} that handles {@link Single} return type. + * + * @author Jakub Narloch + * @see DeferredResult + */ +class SingleDeferredResult extends DeferredResult { + + private static final Object EMPTY_RESULT = new Object(); + + private final DeferredResultSubscriber subscriber; + + public SingleDeferredResult(Single single) { + this(null, EMPTY_RESULT, single); + } + + public SingleDeferredResult(long timeout, Single single) { + this(timeout, EMPTY_RESULT, single); + } + + public SingleDeferredResult(Long timeout, Object timeoutResult, Single single) { + super(timeout, timeoutResult); + Assert.notNull(single, "single can not be null"); + + subscriber = new DeferredResultSubscriber<>(single.toObservable(), this); + } +} diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandler.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandler.java new file mode 100644 index 00000000..190a06d7 --- /dev/null +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandler.java @@ -0,0 +1,117 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import org.springframework.core.MethodParameter; +import org.springframework.core.ResolvableType; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.web.context.request.NativeWebRequest; +import org.springframework.web.context.request.async.DeferredResult; +import org.springframework.web.context.request.async.WebAsyncUtils; +import org.springframework.web.method.support.AsyncHandlerMethodReturnValueHandler; +import org.springframework.web.method.support.ModelAndViewContainer; + +import rx.Single; +import rx.functions.Func1; + +/** + * A specialized {@link AsyncHandlerMethodReturnValueHandler} that handles {@link Single} return types. + * + * @author Spencer Gibb + * @author Jakub Narloch + */ +public class SingleReturnValueHandler implements AsyncHandlerMethodReturnValueHandler { + + @Override + public boolean isAsyncReturnValue(Object returnValue, MethodParameter returnType) { + return returnValue != null && supportsReturnType(returnType); + } + + @Override + public boolean supportsReturnType(MethodParameter returnType) { + return Single.class.isAssignableFrom(returnType.getParameterType()) || isResponseEntity(returnType); + } + + private boolean isResponseEntity(MethodParameter returnType) { + if(ResponseEntity.class.isAssignableFrom(returnType.getParameterType())) { + Class bodyType = ResolvableType.forMethodParameter(returnType).getGeneric(0).resolve(); + return bodyType != null && Single.class.isAssignableFrom(bodyType); + } + return false; + } + + @SuppressWarnings("unchecked") + @Override + public void handleReturnValue(Object returnValue, MethodParameter returnType, ModelAndViewContainer mavContainer, NativeWebRequest webRequest) throws Exception { + + if (returnValue == null) { + mavContainer.setRequestHandled(true); + return; + } + + ResponseEntity> responseEntity = getResponseEntity(returnValue); + if(responseEntity != null) { + returnValue = responseEntity.getBody(); + if (returnValue == null) { + mavContainer.setRequestHandled(true); + return; + } + } + + final Single single = Single.class.cast(returnValue); + WebAsyncUtils.getAsyncManager(webRequest) + .startDeferredResultProcessing(convertToDeferredResult(responseEntity, single), mavContainer); + } + + @SuppressWarnings("unchecked") + private ResponseEntity> getResponseEntity(Object returnValue) { + if (ResponseEntity.class.isAssignableFrom(returnValue.getClass())) { + return (ResponseEntity>) returnValue; + + } + return null; + } + + protected DeferredResult convertToDeferredResult(final ResponseEntity> responseEntity, Single single) { + + //TODO: fix when java8 :-) + Single singleResponse = single.map(new Func1() { + @Override + public ResponseEntity call(Object object) { + return new ResponseEntity<>(object, getHttpHeaders(responseEntity), getHttpStatus(responseEntity)); + } + }); + + return new SingleDeferredResult<>(singleResponse); + } + + private HttpStatus getHttpStatus(ResponseEntity responseEntity) { + if(responseEntity == null) { + return HttpStatus.OK; + } + return responseEntity.getStatusCode(); + } + + private HttpHeaders getHttpHeaders(ResponseEntity responseEntity) { + if(responseEntity == null) { + return new HttpHeaders(); + } + return responseEntity.getHeaders(); + } +} diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTest.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTest.java new file mode 100644 index 00000000..cff2d202 --- /dev/null +++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTest.java @@ -0,0 +1,174 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; + +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.IntegrationTest; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.boot.test.TestRestTemplate; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; + +import rx.Observable; +import rx.Single; +import rx.functions.Func1; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.springframework.http.MediaType.APPLICATION_JSON_UTF8; + +/** + * Tests the demonstrate using {@link Observable} with {@link SingleReturnValueHandler} class. + * + * @author Spencer Gibb + * @author Jakub Narloch + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = ObservableReturnValueHandlerTest.Application.class) +@WebAppConfiguration +@IntegrationTest({"server.port=0"}) +@DirtiesContext +public class ObservableReturnValueHandlerTest { + + @Value("${local.server.port}") + private int port = 0; + + private TestRestTemplate restTemplate = new TestRestTemplate(); + + @Configuration + @EnableAutoConfiguration + @RestController + protected static class Application { + + // tag::rx_observable[] + @RequestMapping(method = RequestMethod.GET, value = "/single") + public Single single() { + return Observable.just("single value").toSingle(); + } + + @RequestMapping(method = RequestMethod.GET, value = "/multiple") + public Single> multiple() { + return Observable.just("multiple", "values").toList().toSingle(); + } + + @RequestMapping(method = RequestMethod.GET, value = "/responseWithObservable") + public ResponseEntity> responseWithObservable() { + + Observable observable = Observable.just("single value"); + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(APPLICATION_JSON_UTF8); + return new ResponseEntity<>(observable.toSingle(), headers, HttpStatus.CREATED); + } + + @RequestMapping(method = RequestMethod.GET, value = "/timeout") + public Observable timeout() { + return Observable.timer(1, TimeUnit.MINUTES).map(new Func1() { + @Override + public String call(Long aLong) { + return "single value"; + } + }); + } + // end::rx_observable[] + + @RequestMapping(method = RequestMethod.GET, value = "/throw") + public Single error() { + return Observable.error(new RuntimeException("Unexpected")).toSingle(); + } + } + + @Test + public void shouldRetrieveSingleValue() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/single"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals("single value", response.getBody()); + } + + @Test + public void shouldRetrieveMultipleValues() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/multiple"), List.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals(Arrays.asList("multiple", "values"), response.getBody()); + } + + @Test + public void shouldRetrieveSingleValueWithStatusCodeAndCustomHeader() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/responseWithObservable"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.CREATED, response.getStatusCode()); + assertEquals(MediaType.APPLICATION_JSON_UTF8, response.getHeaders().getContentType()); + assertEquals("single value", response.getBody()); + } + + @Test + public void shouldRetrieveErrorResponse() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/throw"), Object.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.INTERNAL_SERVER_ERROR, response.getStatusCode()); + } + + @Test + @Ignore("adds 30s to build") + public void shouldTimeoutOnConnection() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/timeout"), Object.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.INTERNAL_SERVER_ERROR, response.getStatusCode()); + } + + private String path(String context) { + return String.format("http://localhost:%d%s", port, context); + } +} \ No newline at end of file diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTests.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTests.java deleted file mode 100644 index 210b6cf3..00000000 --- a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableReturnValueHandlerTests.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Copyright 2013-2015 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.cloud.netflix.rx; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; - -import org.junit.Test; -import org.junit.runner.RunWith; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.test.IntegrationTest; -import org.springframework.boot.test.SpringApplicationConfiguration; -import org.springframework.boot.test.TestRestTemplate; -import org.springframework.context.annotation.Configuration; -import org.springframework.http.HttpStatus; -import org.springframework.http.ResponseEntity; -import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.test.context.web.WebAppConfiguration; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RequestMethod; -import org.springframework.web.bind.annotation.RestController; - -import rx.Observable; - -/** - * @author Spencer Gibb - */ -@RunWith(SpringJUnit4ClassRunner.class) -@SpringApplicationConfiguration(classes = ObservableReturnValueHandlerTests.Application.class) -@WebAppConfiguration -@IntegrationTest({ "server.port=0" }) -@DirtiesContext -public class ObservableReturnValueHandlerTests { - - @Value("${local.server.port}") - private int port = 0; - - @Configuration - @EnableAutoConfiguration - @RestController - protected static class Application { - @RequestMapping(method = RequestMethod.GET, value = "/") - public Observable hi() { - return Observable.just("hello world"); - } - } - - @Test - public void observableReturns() { - ResponseEntity response = new TestRestTemplate().getForEntity("http://localhost:" + port, String.class); - assertNotNull("response was null", response); - assertEquals("response code was wrong", HttpStatus.OK, response.getStatusCode()); - assertEquals("response was wrong", "hello world", response.getBody()); - } -} diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableSseEmitterTest.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableSseEmitterTest.java new file mode 100644 index 00000000..74b6c5b5 --- /dev/null +++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/ObservableSseEmitterTest.java @@ -0,0 +1,165 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import java.util.Date; +import java.util.GregorianCalendar; +import java.util.TimeZone; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.IntegrationTest; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.boot.test.TestRestTemplate; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +import rx.Observable; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.springframework.http.MediaType.APPLICATION_JSON_UTF8; + +/** + * Tests the {@link ObservableSseEmitter} class. + * + * @author Spencer Gibb + * @author Jakub Narloch + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = ObservableSseEmitterTest.Application.class) +@WebAppConfiguration +@IntegrationTest({"server.port=0"}) +@DirtiesContext +public class ObservableSseEmitterTest { + + @Value("${local.server.port}") + private int port = 0; + + private TestRestTemplate restTemplate = new TestRestTemplate(); + + @Configuration + @EnableAutoConfiguration + @RestController + protected static class Application { + + // tag::rx_observable_sse[] + @RequestMapping(method = RequestMethod.GET, value = "/sse") + public SseEmitter single() { + return RxResponse.sse(Observable.just("single value")); + } + + @RequestMapping(method = RequestMethod.GET, value = "/messages") + public SseEmitter messages() { + return RxResponse.sse(Observable.just("message 1", "message 2", "message 3")); + } + + @RequestMapping(method = RequestMethod.GET, value = "/events") + public SseEmitter event() { + return RxResponse.sse(APPLICATION_JSON_UTF8, Observable.just( + new EventDto("Spring io", getDate(2016, 5, 19)), + new EventDto("SpringOnePlatform", getDate(2016, 8, 1)) + )); + } + // end::rx_observable_sse[] + } + + @Test + public void shouldRetrieveSse() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/sse"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals("data:single value\n\n", response.getBody()); + } + + @Test + public void shouldRetrieveSseWithMultipleMessages() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/messages"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals("data:message 1\n\ndata:message 2\n\ndata:message 3\n\n", response.getBody()); + } + + @Test + public void shouldRetrieveJsonOverSseWithMultipleMessages() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/events"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals("data:{\"name\":\"Spring io\",\"date\":1466337600000}\n\ndata:{\"name\":\"SpringOnePlatform\",\"date\":1472731200000}\n\n", response.getBody()); + } + + private String path(String context) { + return String.format("http://localhost:%d%s", port, context); + } + + private static Date getDate(int year, int month, int day) { + GregorianCalendar calendar = new GregorianCalendar(year, month, day, 12, 0, 0); + calendar.setTimeZone(TimeZone.getTimeZone("UTC")); + return calendar.getTime(); + } + + /** + * A simple DTO used for testing purpose. + * + * @author Jakub Narloch + */ + static class EventDto { + + private final String name; + + private final Date date; + + @JsonCreator + public EventDto(@JsonProperty("name") String name, @JsonProperty("date") Date date) { + this.name = name; + this.date = date; + } + + public String getName() { + return name; + } + + public Date getDate() { + return date; + } + } +} \ No newline at end of file diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandlerTest.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandlerTest.java new file mode 100644 index 00000000..976145f6 --- /dev/null +++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/rx/SingleReturnValueHandlerTest.java @@ -0,0 +1,120 @@ +/* + * Copyright 2013-2016 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.cloud.netflix.rx; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.IntegrationTest; +import org.springframework.boot.test.SpringApplicationConfiguration; +import org.springframework.boot.test.TestRestTemplate; +import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.web.WebAppConfiguration; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; +import org.springframework.web.bind.annotation.RestController; + +import rx.Single; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * Tests the {@link SingleReturnValueHandler} class. + * + * @author Spencer Gibb + * @author Jakub Narloch + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringApplicationConfiguration(classes = SingleReturnValueHandlerTest.Application.class) +@WebAppConfiguration +@IntegrationTest({"server.port=0"}) +@DirtiesContext +public class SingleReturnValueHandlerTest { + + @Value("${local.server.port}") + private int port = 0; + + private TestRestTemplate restTemplate = new TestRestTemplate(); + + @Configuration + @EnableAutoConfiguration + @RestController + protected static class Application { + + // tag::rx_single[] + @RequestMapping(method = RequestMethod.GET, value = "/single") + public Single single() { + return Single.just("single value"); + } + + @RequestMapping(method = RequestMethod.GET, value = "/singleWithResponse") + public ResponseEntity> singleWithResponse() { + return new ResponseEntity<>(Single.just("single value"), HttpStatus.NOT_FOUND); + } + + @RequestMapping(method = RequestMethod.GET, value = "/throw") + public Single error() { + return Single.error(new RuntimeException("Unexpected")); + } + // end::rx_single[] + } + + @Test + public void shouldRetrieveSingleValue() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/single"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.OK, response.getStatusCode()); + assertEquals("single value", response.getBody()); + } + + @Test + public void shouldRetrieveSingleValueWithStatusCode() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/singleWithResponse"), String.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.NOT_FOUND, response.getStatusCode()); + assertEquals("single value", response.getBody()); + } + + @Test + public void shouldRetrieveErrorResponse() { + + // when + ResponseEntity response = restTemplate.getForEntity(path("/throw"), Object.class); + + // then + assertNotNull(response); + assertEquals(HttpStatus.INTERNAL_SERVER_ERROR, response.getStatusCode()); + } + + private String path(String context) { + return String.format("http://localhost:%d%s", port, context); + } +} \ No newline at end of file