[CXF-7535] Adding client & server support for Project Reactor #331

johnament · 2017-10-29T18:34:44Z

This is still a WIP, so not ready to be merged.

I'm looking to add support for Project Reactor as an rx invoker for CXF. It effectively mirrors what the other two are doing, but using reactor's APIs.

I'm still working on tests (e.g. I don't know if this works). the biggest I'm having right now is getting the systests to import in intellij. I've done it previously, but for some reason after switching to 3.2 it's no good. I've even gone in, blown away all .iml and the .idea folder, no luck. Anyone else seen something like this?

deki · 2017-10-29T18:49:25Z

Usually I have this issue if I miss some Maven profile selections. Adjusting the checkboxes normally fixes this.

johnament · 2017-10-30T02:35:44Z

Hi @deki thanks for the tip. What profiles would you recommend? note that I only have the jaxrs systests module open in my IDE, but can go back to the broader one.

One other thing, i'm noticing that the settings jar from intellij doesn't put imports in the right order.

deki · 2017-10-30T13:42:37Z

I usually check the everything profile. But in your case it looks more like the dependencies are not properly resolve. Hit the refresh button in the Maven projects tab (first one on the left). If this doesn't work, run File -> Invalidate Caches/ Restart.

reta · 2017-11-04T18:06:57Z

.../extensions/reactor/src/main/java/org/apache/cxf/jaxrs/reactor/client/FluxRxInvokerImpl.java

+    }
+
+    @Override
+    public Flux get() {


Flux<Response> please :)

Should be fixed.

reta · 2017-11-04T18:23:26Z

.../extensions/reactor/src/main/java/org/apache/cxf/jaxrs/reactor/client/FluxRxInvokerImpl.java

+    }
+
+    private <R> Flux<R> flux(Future<R> future) {
+        Flux<R> flux = Flux.from(Mono.fromFuture(toCompletableFuture(future)));


This part concerns. I would suggest to not use scheduler but pass executor directly to toCompletableFuture (so the supplyAsync call would be done outside of common ForkJoin pool). Otherwise we have too many pools and threads involved for no real reason.

Regarding the desired behavior, we would certainly want the subscription to be invoked on the separate thread (this is what subscribeOn does now). However, because we create Flux from CompletableFuture, intuitively the completion callback would be invoked from another thread, either the one from common ForkJoin pool or the provided one (unless then value is available immediately).

Thoughts?

Ok, the executor part is fixed. If there's an executor passed in we use that.

However, there's an interesting conundrum. I'm not sure how rx() is meant to work this way. For a Mono object its pretty clear, I had to tweak the test to use a version of the json processor that didn't treat the result as an array. Which makes sense, its meant to be a single object.

However, I need to figure out a way to check if the opening character when reading the result is a single or multiple result. Effectively I need the MessageBodyReader equivalent of StreamingResponseProvider. I'm kind of wondering how this works for the RxJava versions (but then again, there's no tests for this method, so I wonder if you guys ran into that issue already).

On the client side one works against a specific URI and thus the expectations are known in advance, i.e, it is the collection or not which is returned

That's what I'm thinking as well, to make it less complicated. So what that means is if you work with it, by default you're getting Mono objects back (since it's typically one element). Any method that supports a GenericType I'm going to add a Flux version for that just takes the singular attribute, so that you end up with something like <R> Flux<R> getFlux(Class<R> responseType) which should give a pretty clean programming model.

Can you clarify a bit, may be with the code examples in the JIRA issue ? I'm not sure what is a distinction between a single and multiple objects on the client side, example, on the server one can be streaming a sequence of Book objects, but on the client it can be a single Books object.

cschneider · 2017-11-06T06:10:47Z

Do we need to specifically support reactor or is it enough to just depend on reactive streams?

johnament · 2017-11-06T11:23:51Z

@cschneider there's a new rx() feature in the client adapter that is meant to support converting the response into the appropriate objects.

cschneider · 2017-11-06T11:30:41Z

Got it now .. looks good this way.

reta · 2017-11-07T02:24:22Z

rt/rs/extensions/reactor/pom.xml

+    <parent>
+        <groupId>org.apache.cxf</groupId>
+        <artifactId>cxf-parent</artifactId>
+        <version>3.2.1-SNAPSHOT</version>


3.2.2-SNAPSHOT :)

Yep, just saw it.

reta · 2017-11-07T02:30:46Z

.../extensions/reactor/src/main/java/org/apache/cxf/jaxrs/reactor/client/FluxRxInvokerImpl.java

+
+    private <R> Flux<R> flux(Future<R> future) {
+        Flux<R> flux = Flux.from(Mono.fromFuture(toCompletableFuture(future, executorService)));
+        if (scheduler != null) {


@johnament Thanks for passing the executorService to toCompletableFuture. The scheduler and this snippet:

if (scheduler != null) { flux = flux.subscribeOn(scheduler); }

are not needed. The reason is that the completion callback for Mono.fromFuture will be called in one of the executorService threads or ForkJoin pool (you could easily trace it). Moreover, we could run into issues with this code because it uses two threads from the same pool for toCompletableFuture and subscription, exposing ourselves for likelihood of livelocks.

Since the underlying call is happening via async webclient, we can't take full advantage of the reactive nature (but not sure the JSON format works well, i want to play around with SSE next after this is merged). Scheduler should not be used.

reta · 2017-11-07T02:33:42Z

rt/rs/extensions/reactor/src/main/java/org/apache/cxf/jaxrs/reactor/client/MonoRxInvoker.java

+import reactor.core.publisher.Mono;
+
+@SuppressWarnings("rawtypes")
+public interface MonoRxInvoker extends RxInvoker<Mono> {


If you use extends RxInvoker<Mono<?>>, you could get rid of @SuppressWarnings("rawtypes")

Yes, but then the overrides don't line up properly.

Nevermind, whatever was happening isn't any longer.

reta · 2017-11-07T02:35:45Z

rt/rs/extensions/reactor/src/main/java/org/apache/cxf/jaxrs/reactor/client/FluxRxInvoker.java

+import reactor.core.publisher.Flux;
+
+@SuppressWarnings("rawtypes")
+public interface FluxRxInvoker extends RxInvoker<Flux> {


Same as for MonoRxInvoker, if you use extends RxInvoker<Flux<?>> you could remove @SuppressWarnings("rawtypes")

reta · 2017-11-07T02:39:18Z

rt/rs/extensions/reactor/src/main/java/org/apache/cxf/jaxrs/reactor/client/ReactorUtils.java

+            try {
+                return future.get();
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(e);


Using CompletionException would be more appropriate I think. Otherwise we would end up with longer ex-> RuntimeException(ex) -> CompletionException chain of causes.

reta · 2017-11-07T13:37:38Z

@johnament It looks pretty good in general, thanks for addressing the comments. Are there any outstanding issues you are struggling with / need help?

johnament · 2017-11-07T23:55:23Z

@reta there's open questions I have:

Leave the tests here or move into systests? Personally I don't like not having tests with the main code, and it's a trivial impact (14s build + test time).
Creating examples? Maybe later.
Should I add CDI integration? E.g. an implementation of JAXRSServerFactoryCustomizationExtension that adds the ReactorInvoker to the runtime.

reta · 2017-11-08T02:56:20Z

@johnament my few cents on the questions you have:

I would suggest to move it to systests, not necessarily due to execution time but because of the fact we spawn servers (external components). This is not the official rule I think but it serves as the litmus test where tests belong.
Yes, this would be very good to have, at least 1 example in the samples distribution, but I can help with that (as well as adding the documentation to the CXF Wiki)
CDI would be good to have but it could be done later I think (unless you really need it from day 1).

@sberyozkin What do you think?

cschneider · 2017-11-08T06:02:21Z

I would keep tests with the module unless they are slow or very involved. We spawn CXF endpoints in a lot of other modules tests.
The full cxf build takes quite a while. So it is very useful to be able to build a single module and get some good feedback if it is correct.
One other thing about the tests. There are a lot of sleeps in the tests. This is not good. I propose to replace this with Awaitility based retries.
For example:
assertEquals("Hello, world!", holder.value);
could be:
Awaitility.await().until(()->holder.value, equalTo("Hello world!"));

sberyozkin · 2017-11-08T10:02:06Z

Hi All, my understanding is that it is a CXF convention to have unit tests with the modules and the end to end tests involving starting endpoints and clients and sending the data over some transports - in systests, so IMHO it makes sense to keep that way. John, Andriy, re CDI, yes, nice idea, but may be indeed at the next stage (and would that make a ReactorInvoker a default one when CDI is loaded ? something we can discuss later)

johnament · 2017-11-08T13:07:33Z

Ok, I just pushed up the move to systests. Assuming that all passes in jenkins I'll merge and circle back on the remaining parts.

…#331)

Entesb 16463

johnament requested review from andymc12, dkulp, reta and sberyozkin October 29, 2017 18:34

reta reviewed Nov 4, 2017

View reviewed changes

johnament added 3 commits November 5, 2017 19:15

[CXF-7535] Adding client & server support for Project Reactor

f607e39

[CXF-7535] Making sure Fluxes work client and server. Starting on monos.

63fafd9

[CXF-7535] Finishing up flux and mono tests.

b1e99a5

johnament force-pushed the CXF-7535 branch from 09427ba to b1e99a5 Compare November 6, 2017 00:16

reta reviewed Nov 7, 2017

View reviewed changes

johnament added 3 commits November 6, 2017 21:41

[CXF-7535] Adding client & server support for Project Reactor

0365b73

[CXF-7535] Adding client & server support for Project Reactor

f34a162

[CXF-7535] Adding client & server support for Project Reactor

c0a97f5

[CXF-7535] Adding client & server support for Project Reactor

81000db

johnament merged commit f634fb8 into apache:master Nov 9, 2017

rmannibucau pushed a commit to rmannibucau/cxf that referenced this pull request Sep 23, 2018

[CXF-7535] Adding client & server support for Project Reactor (apache…

831a01f

…#331)

rmannibucau pushed a commit to rmannibucau/cxf that referenced this pull request Mar 18, 2019

[CXF-7535] Adding client & server support for Project Reactor (apache…

fbc172c

…#331)

rmannibucau pushed a commit to rmannibucau/cxf that referenced this pull request May 31, 2019

[CXF-7535] Adding client & server support for Project Reactor (apache…

0562f2a

…#331)

rnetuka pushed a commit to rnetuka/cxf that referenced this pull request Oct 3, 2022

Merge pull request apache#331 from ffang/ENTESB-16463

68c68cb

Entesb 16463

[CXF-7535] Adding client & server support for Project Reactor #331

[CXF-7535] Adding client & server support for Project Reactor #331

Uh oh!

Conversation

johnament commented Oct 29, 2017

Uh oh!

deki commented Oct 29, 2017

Uh oh!

johnament commented Oct 30, 2017

Uh oh!

deki commented Oct 30, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reta Nov 4, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cschneider commented Nov 6, 2017

Uh oh!

johnament commented Nov 6, 2017

Uh oh!

cschneider commented Nov 6, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reta Nov 7, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

reta commented Nov 7, 2017

Uh oh!

johnament commented Nov 7, 2017

Uh oh!

reta commented Nov 8, 2017

Uh oh!

cschneider commented Nov 8, 2017

Uh oh!

sberyozkin commented Nov 8, 2017

Uh oh!

johnament commented Nov 8, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

reta Nov 4, 2017 •

edited

Loading

reta Nov 7, 2017 •

edited

Loading