Skip to content

Added configuration for reactive scheduler #1032

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kdavisk6 merged 2 commits into from
Aug 7, 2019
Merged

Added configuration for reactive scheduler #1032

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2912b1a
Added configuration for reactive scheduler
SimY4 Aug 7, 2019
64a7c3c
make final things explicitly final.
SimY4 Aug 7, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 18 additions & 5 deletions reactive/src/main/java/feign/reactive/ReactorFeign.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,8 @@
package feign.reactive;

import feign.Feign;
import feign.reactive.ReactiveFeign.Builder;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Map;
Expand All @@ -29,24 +30,36 @@ public static Builder builder() {

public static class Builder extends ReactiveFeign.Builder {

private Scheduler scheduler = Schedulers.elastic();

@Override
public Feign build() {
super.invocationHandlerFactory(new ReactorInvocationHandlerFactory());
super.invocationHandlerFactory(new ReactorInvocationHandlerFactory(scheduler));
return super.build();
}

@Override
public Feign.Builder invocationHandlerFactory(
InvocationHandlerFactory invocationHandlerFactory) {
public Builder invocationHandlerFactory(InvocationHandlerFactory invocationHandlerFactory) {
throw new UnsupportedOperationException(
"Invocation Handler Factory overrides are not supported.");
}

public Builder scheduleOn(Scheduler scheduler) {
this.scheduler = scheduler;
return this;
}
}

private static class ReactorInvocationHandlerFactory implements InvocationHandlerFactory {
private final Scheduler scheduler;

private ReactorInvocationHandlerFactory(Scheduler scheduler) {
this.scheduler = scheduler;
}

@Override
public InvocationHandler create(Target target, Map<Method, MethodHandler> dispatch) {
return new ReactorInvocationHandler(target, dispatch);
return new ReactorInvocationHandler(target, dispatch, scheduler);
}
}
}
11 changes: 7 additions & 4 deletions reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,22 +20,25 @@
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.core.scheduler.Scheduler;

public class ReactorInvocationHandler extends ReactiveInvocationHandler {
private final Scheduler scheduler;

ReactorInvocationHandler(Target<?> target,
Map<Method, MethodHandler> dispatch) {
Map<Method, MethodHandler> dispatch,
Scheduler scheduler) {
super(target, dispatch);
this.scheduler = scheduler;
}

@Override
protected Publisher invoke(Method method, MethodHandler methodHandler, Object[] arguments) {
Publisher<?> invocation = this.invokeMethod(methodHandler, arguments);
if (Flux.class.isAssignableFrom(method.getReturnType())) {
return Flux.from(invocation).subscribeOn(Schedulers.elastic());
return Flux.from(invocation).subscribeOn(scheduler);
} else if (Mono.class.isAssignableFrom(method.getReturnType())) {
return Mono.from(invocation).subscribeOn(Schedulers.elastic());
return Mono.from(invocation).subscribeOn(scheduler);
}
throw new IllegalArgumentException(
"Return type " + method.getReturnType().getName() + " is not supported");
Expand Down
21 changes: 17 additions & 4 deletions reactive/src/main/java/feign/reactive/RxJavaFeign.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,8 @@
import feign.Feign;
import feign.InvocationHandlerFactory;
import feign.Target;
import io.reactivex.Scheduler;
import io.reactivex.schedulers.Schedulers;

public class RxJavaFeign extends ReactiveFeign {

Expand All @@ -28,25 +30,36 @@ public static Builder builder() {

public static class Builder extends ReactiveFeign.Builder {

private Scheduler scheduler = Schedulers.trampoline();

@Override
public Feign build() {
super.invocationHandlerFactory(new RxJavaInvocationHandlerFactory());
super.invocationHandlerFactory(new RxJavaInvocationHandlerFactory(scheduler));
return super.build();
}

@Override
public Feign.Builder invocationHandlerFactory(
InvocationHandlerFactory invocationHandlerFactory) {
public Builder invocationHandlerFactory(InvocationHandlerFactory invocationHandlerFactory) {
throw new UnsupportedOperationException(
"Invocation Handler Factory overrides are not supported.");
}

public Builder scheduleOn(Scheduler scheduler) {
this.scheduler = scheduler;
return this;
}
}

private static class RxJavaInvocationHandlerFactory implements InvocationHandlerFactory {
private final Scheduler scheduler;

private RxJavaInvocationHandlerFactory(Scheduler scheduler) {
this.scheduler = scheduler;
}

@Override
public InvocationHandler create(Target target, Map<Method, MethodHandler> dispatch) {
return new RxJavaInvocationHandler(target, dispatch);
return new RxJavaInvocationHandler(target, dispatch, scheduler);
}
}

Expand Down
9 changes: 6 additions & 3 deletions reactive/src/main/java/feign/reactive/RxJavaInvocationHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,21 +16,24 @@
import feign.InvocationHandlerFactory.MethodHandler;
import feign.Target;
import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.Scheduler;
import java.lang.reflect.Method;
import java.util.Map;
import org.reactivestreams.Publisher;

public class RxJavaInvocationHandler extends ReactiveInvocationHandler {
private final Scheduler scheduler;

RxJavaInvocationHandler(Target<?> target,
Map<Method, MethodHandler> dispatch) {
Map<Method, MethodHandler> dispatch,
Scheduler scheduler) {
super(target, dispatch);
this.scheduler = scheduler;
}

@Override
protected Publisher invoke(Method method, MethodHandler methodHandler, Object[] arguments) {
return Flowable.fromPublisher(this.invokeMethod(methodHandler, arguments))
.observeOn(Schedulers.trampoline());
.observeOn(scheduler);
}
}
20 changes: 11 additions & 9 deletions reactive/src/test/java/feign/reactive/ReactiveInvocationHandlerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,17 +23,16 @@
import feign.RequestLine;
import feign.Target;
import io.reactivex.Flowable;

import java.io.IOException;
import java.lang.reflect.Method;
import java.util.Collections;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.runners.MockitoJUnitRunner;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;

@RunWith(MockitoJUnitRunner.class)
Expand All @@ -57,7 +56,7 @@ public void setUp() throws NoSuchMethodException {
public void invokeOnSubscribeReactor() throws Throwable {
given(this.methodHandler.invoke(any())).willReturn("Result");
ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target,
Collections.singletonMap(method, this.methodHandler));
Collections.singletonMap(method, this.methodHandler), Schedulers.elastic());

Object result = handler.invoke(method, this.methodHandler, new Object[] {});
assertThat(result).isInstanceOf(Mono.class);
Expand All @@ -75,7 +74,7 @@ public void invokeOnSubscribeReactor() throws Throwable {
public void invokeOnSubscribeEmptyReactor() throws Throwable {
given(this.methodHandler.invoke(any())).willReturn(null);
ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target,
Collections.singletonMap(method, this.methodHandler));
Collections.singletonMap(method, this.methodHandler), Schedulers.elastic());

Object result = handler.invoke(method, this.methodHandler, new Object[] {});
assertThat(result).isInstanceOf(Mono.class);
Expand All @@ -92,7 +91,7 @@ public void invokeOnSubscribeEmptyReactor() throws Throwable {
public void invokeFailureReactor() throws Throwable {
given(this.methodHandler.invoke(any())).willThrow(new IOException("Could Not Decode"));
ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target,
Collections.singletonMap(this.method, this.methodHandler));
Collections.singletonMap(this.method, this.methodHandler), Schedulers.elastic());

Object result = handler.invoke(this.method, this.methodHandler, new Object[] {});
assertThat(result).isInstanceOf(Mono.class);
Expand All @@ -111,7 +110,8 @@ public void invokeOnSubscribeRxJava() throws Throwable {
given(this.methodHandler.invoke(any())).willReturn("Result");
RxJavaInvocationHandler handler =
new RxJavaInvocationHandler(this.target,
Collections.singletonMap(this.method, this.methodHandler));
Collections.singletonMap(this.method, this.methodHandler),
io.reactivex.schedulers.Schedulers.trampoline());

Object result = handler.invoke(this.method, this.methodHandler, new Object[] {});
assertThat(result).isInstanceOf(Flowable.class);
Expand All @@ -129,8 +129,9 @@ public void invokeOnSubscribeRxJava() throws Throwable {
public void invokeOnSubscribeEmptyRxJava() throws Throwable {
given(this.methodHandler.invoke(any())).willReturn(null);
RxJavaInvocationHandler handler =
new RxJavaInvocationHandler(this.target,
Collections.singletonMap(this.method, this.methodHandler));
new RxJavaInvocationHandler(this.target,
Collections.singletonMap(this.method, this.methodHandler),
io.reactivex.schedulers.Schedulers.trampoline());

Object result = handler.invoke(this.method, this.methodHandler, new Object[] {});
assertThat(result).isInstanceOf(Flowable.class);
Expand All @@ -148,7 +149,8 @@ public void invokeFailureRxJava() throws Throwable {
given(this.methodHandler.invoke(any())).willThrow(new IOException("Could Not Decode"));
RxJavaInvocationHandler handler =
new RxJavaInvocationHandler(this.target,
Collections.singletonMap(this.method, this.methodHandler));
Collections.singletonMap(this.method, this.methodHandler),
io.reactivex.schedulers.Schedulers.trampoline());

Object result = handler.invoke(this.method, this.methodHandler, new Object[] {});
assertThat(result).isInstanceOf(Flowable.class);
Expand Down