This is an automated email from the ASF dual-hosted git repository.
sergeyb pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/master by this push:
new 3ebafed [CXF-7556][CXF-7535] Updating RxJava2 invoker to optionally
use the streaming subscriber, updating reactor pom.xml not to use optional deps
3ebafed is described below
commit 3ebafed47e8ddb65f19860fc9a5991562c85e949
Author: Sergey Beryozkin <[email protected]>
AuthorDate: Wed Nov 15 13:39:43 2017 +0000
[CXF-7556][CXF-7535] Updating RxJava2 invoker to optionally use the
streaming subscriber, updating reactor pom.xml not to use optional deps
---
.../reactivestreams/server/AbstractSubscriber.java | 7 +++++-
rt/rs/extensions/reactor/pom.xml | 1 -
rt/rs/extensions/rx/pom.xml | 7 ++++++
.../cxf/jaxrs/rx2/server/ReactiveIOInvoker.java | 29 ++++++++++++++++------
systests/jaxrs/pom.xml | 4 ---
.../jaxrs/reactive/JAXRSRxJava2FlowableTest.java | 5 ++++
.../jaxrs/reactive/RxJava2FlowableServer.java | 15 ++++++++---
.../jaxrs/reactive/RxJava2FlowableService.java | 9 ++++++-
8 files changed, 60 insertions(+), 17 deletions(-)
diff --git
a/rt/rs/extensions/reactivestreams/src/main/java/org/apache/cxf/jaxrs/reactivestreams/server/AbstractSubscriber.java
b/rt/rs/extensions/reactivestreams/src/main/java/org/apache/cxf/jaxrs/reactivestreams/server/AbstractSubscriber.java
index b7e4b59..b541d1b 100644
---
a/rt/rs/extensions/reactivestreams/src/main/java/org/apache/cxf/jaxrs/reactivestreams/server/AbstractSubscriber.java
+++
b/rt/rs/extensions/reactivestreams/src/main/java/org/apache/cxf/jaxrs/reactivestreams/server/AbstractSubscriber.java
@@ -19,6 +19,7 @@
package org.apache.cxf.jaxrs.reactivestreams.server;
import java.util.List;
+import java.util.concurrent.CancellationException;
import javax.ws.rs.container.AsyncResponse;
@@ -48,7 +49,11 @@ public abstract class AbstractSubscriber<T> implements
Subscriber<T> {
@Override
public void onError(Throwable t) {
- ar.resume(t);
+ if (t instanceof CancellationException) {
+ ar.cancel();
+ } else {
+ ar.resume(t);
+ }
}
@Override
diff --git a/rt/rs/extensions/reactor/pom.xml b/rt/rs/extensions/reactor/pom.xml
index c5fd5e3..40de637 100644
--- a/rt/rs/extensions/reactor/pom.xml
+++ b/rt/rs/extensions/reactor/pom.xml
@@ -45,7 +45,6 @@
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
- <scope>provided</scope>
</dependency>
</dependencies>
</project>
diff --git a/rt/rs/extensions/rx/pom.xml b/rt/rs/extensions/rx/pom.xml
index 4eba638..58defb8 100644
--- a/rt/rs/extensions/rx/pom.xml
+++ b/rt/rs/extensions/rx/pom.xml
@@ -54,6 +54,13 @@
<optional>true</optional>
</dependency>
<dependency>
+ <groupId>org.apache.cxf</groupId>
+ <artifactId>cxf-rt-rs-extension-reactivestreams</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
diff --git
a/rt/rs/extensions/rx/src/main/java/org/apache/cxf/jaxrs/rx2/server/ReactiveIOInvoker.java
b/rt/rs/extensions/rx/src/main/java/org/apache/cxf/jaxrs/rx2/server/ReactiveIOInvoker.java
index 8e8512c..44ac9c7 100644
---
a/rt/rs/extensions/rx/src/main/java/org/apache/cxf/jaxrs/rx2/server/ReactiveIOInvoker.java
+++
b/rt/rs/extensions/rx/src/main/java/org/apache/cxf/jaxrs/rx2/server/ReactiveIOInvoker.java
@@ -18,10 +18,11 @@
*/
package org.apache.cxf.jaxrs.rx2.server;
-import java.util.concurrent.CancellationException;
+import javax.ws.rs.core.MediaType;
import org.apache.cxf.jaxrs.JAXRSInvoker;
import org.apache.cxf.jaxrs.impl.AsyncResponseImpl;
+import
org.apache.cxf.jaxrs.reactivestreams.server.JsonStreamingAsyncSubscriber;
import org.apache.cxf.message.Message;
import io.reactivex.Flowable;
@@ -29,6 +30,7 @@ import io.reactivex.Observable;
//Work in Progress
public class ReactiveIOInvoker extends JAXRSInvoker {
+ private boolean useStreamingSubscriberIfPossible;
protected AsyncResponseImpl checkFutureResponse(Message inMessage, Object
result) {
if (result instanceof Flowable) {
return handleFlowable(inMessage, (Flowable<?>)result);
@@ -41,10 +43,18 @@ public class ReactiveIOInvoker extends JAXRSInvoker {
protected AsyncResponseImpl handleFlowable(Message inMessage, Flowable<?>
f) {
final AsyncResponseImpl asyncResponse = new
AsyncResponseImpl(inMessage);
- f.subscribe(v -> asyncResponse.resume(v), t ->
handleThrowable(asyncResponse, t));
+ if (isUseStreamingSubscriberIfPossible() && isJsonResponse(inMessage))
{
+ f.subscribe(new JsonStreamingAsyncSubscriber<>(asyncResponse));
+ } else {
+ f.subscribe(v -> asyncResponse.resume(v), t ->
handleThrowable(asyncResponse, t));
+ }
return asyncResponse;
}
+ protected boolean isJsonResponse(Message inMessage) {
+ return
MediaType.APPLICATION_JSON.equals(inMessage.getExchange().get(Message.CONTENT_TYPE));
+ }
+
protected AsyncResponseImpl handleObservable(Message inMessage,
Observable<?> obs) {
final AsyncResponseImpl asyncResponse = new
AsyncResponseImpl(inMessage);
obs.subscribe(v -> asyncResponse.resume(v), t ->
handleThrowable(asyncResponse, t));
@@ -52,11 +62,16 @@ public class ReactiveIOInvoker extends JAXRSInvoker {
}
private Object handleThrowable(AsyncResponseImpl asyncResponse, Throwable
t) {
- if (t instanceof CancellationException) {
- asyncResponse.cancel();
- } else {
- asyncResponse.resume(t);
- }
+ //TODO: if it is a Cancelation exception => asyncResponse.cancel();
+ asyncResponse.resume(t);
return null;
}
+
+ public boolean isUseStreamingSubscriberIfPossible() {
+ return useStreamingSubscriberIfPossible;
+ }
+
+ public void setUseStreamingSubscriberIfPossible(boolean
useStreamingSubscriberIfPossible) {
+ this.useStreamingSubscriberIfPossible =
useStreamingSubscriberIfPossible;
+ }
}
diff --git a/systests/jaxrs/pom.xml b/systests/jaxrs/pom.xml
index 9619e39..450aed6 100644
--- a/systests/jaxrs/pom.xml
+++ b/systests/jaxrs/pom.xml
@@ -66,10 +66,6 @@
<artifactId>rxjava</artifactId>
</dependency>
<dependency>
- <groupId>io.projectreactor</groupId>
- <artifactId>reactor-core</artifactId>
- </dependency>
- <dependency>
<groupId>org.apache.cxf</groupId>
<artifactId>cxf-rt-rs-extension-reactivestreams</artifactId>
<version>${project.version}</version>
diff --git
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/JAXRSRxJava2FlowableTest.java
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/JAXRSRxJava2FlowableTest.java
index 1b70db0..479f95b 100644
---
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/JAXRSRxJava2FlowableTest.java
+++
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/JAXRSRxJava2FlowableTest.java
@@ -90,6 +90,11 @@ public class JAXRSRxJava2FlowableTest extends
AbstractBusClientServerTestBase {
String address = "http://localhost:"; + PORT +
"/rx2/flowable/textJsonImplicitListAsyncStream";
doTestGetHelloWorldJsonList(address);
}
+ @Test
+ public void testGetHelloWorldJsonImplicitList() throws Exception {
+ String address = "http://localhost:"; + PORT +
"/rx22/flowable/textJsonImplicitList";
+ doTestGetHelloWorldJsonList(address);
+ }
private void doTestGetHelloWorldJsonList(String address) throws Exception {
WebClient wc = WebClient.create(address,
Collections.singletonList(new
JacksonJsonProvider()));
diff --git
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableServer.java
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableServer.java
index 8558bed..5063b5e 100644
---
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableServer.java
+++
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableServer.java
@@ -37,6 +37,7 @@ public class RxJava2FlowableServer extends
AbstractBusTestServerBase {
public static final String PORT =
allocatePort(RxJava2FlowableServer.class);
org.apache.cxf.endpoint.Server server;
+ org.apache.cxf.endpoint.Server server2;
public RxJava2FlowableServer() {
}
@@ -44,8 +45,16 @@ public class RxJava2FlowableServer extends
AbstractBusTestServerBase {
Bus bus = BusFactory.getDefaultBus();
// Make sure default JSONProvider is not loaded
bus.setProperty("skip.default.json.provider.registration", true);
+ server = createFactoryBean(bus, false, "/rx2").create();
+ server = createFactoryBean(bus, true, "/rx22").create();
+ }
+
+ private JAXRSServerFactoryBean createFactoryBean(Bus bus, boolean
useStreamingSubscriber,
+ String relAddress) {
JAXRSServerFactoryBean sf = new JAXRSServerFactoryBean();
- sf.setInvoker(new ReactiveIOInvoker());
+ ReactiveIOInvoker invoker = new ReactiveIOInvoker();
+ invoker.setUseStreamingSubscriberIfPossible(useStreamingSubscriber);
+ sf.setInvoker(invoker);
sf.setProvider(new JacksonJsonProvider());
StreamingResponseProvider<HelloWorldBean> streamProvider = new
StreamingResponseProvider<HelloWorldBean>();
streamProvider.setProduceMediaTypes(Collections.singletonList("application/json"));
@@ -54,8 +63,8 @@ public class RxJava2FlowableServer extends
AbstractBusTestServerBase {
sf.setResourceClasses(RxJava2FlowableService.class);
sf.setResourceProvider(RxJava2FlowableService.class,
new SingletonResourceProvider(new
RxJava2FlowableService(), true));
- sf.setAddress("http://localhost:"; + PORT + "/");
- server = sf.create();
+ sf.setAddress("http://localhost:"; + PORT + relAddress);
+ return sf;
}
public void tearDown() throws Exception {
diff --git
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableService.java
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableService.java
index a995edc..c8fea92 100644
---
a/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableService.java
+++
b/systests/jaxrs/src/test/java/org/apache/cxf/systest/jaxrs/reactive/RxJava2FlowableService.java
@@ -36,7 +36,7 @@ import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
-@Path("/rx2/flowable")
+@Path("/flowable")
public class RxJava2FlowableService {
@@ -76,6 +76,13 @@ public class RxJava2FlowableService {
}
@GET
+ @Produces("application/json")
+ @Path("textJsonImplicitList")
+ public Flowable<HelloWorldBean> getJsonImplicitList() {
+ return Flowable.just("Hello", "Ciao").map(HelloWorldBean::new);
+ }
+
+ @GET
@Produces("text/plain")
@Path("textAsync")
public void getTextAsync(@Suspended final AsyncResponse ar) {
--
To stop receiving notification emails like this one, please contact
['"[email protected]" <[email protected]>'].
