diff --git a/core/src/main/java/feign/Types.java b/core/src/main/java/feign/Types.java index 1e642444f6..f974036f49 100644 --- a/core/src/main/java/feign/Types.java +++ b/core/src/main/java/feign/Types.java @@ -30,7 +30,7 @@ * @author Bob Lee * @author Jesse Wilson */ -final class Types { +public final class Types { private static final Type[] EMPTY_TYPE_ARRAY = new Type[0]; @@ -38,7 +38,7 @@ private Types() { // No instances. } - static Class getRawType(Type type) { + public static Class getRawType(Type type) { if (type instanceof Class) { // Type is a normal class. return (Class) type; diff --git a/pom.xml b/pom.xml index e628b3ccf3..621e0a6b03 100644 --- a/pom.xml +++ b/pom.xml @@ -36,6 +36,7 @@ ribbon sax slf4j + reactive example-github example-wikipedia diff --git a/reactive/README.md b/reactive/README.md new file mode 100644 index 0000000000..ee910506a3 --- /dev/null +++ b/reactive/README.md @@ -0,0 +1,113 @@ +Reactive Streams Wrapper +--- + +This module wraps Feign's http requests in a [Reactive Streams](https://reactive-streams.org) +Publisher, enabling the use of Reactive Stream `Publisher` return types. Supported Reactive Streams implementations are: + +* [Reactor](https://project-reactor.org) (`Mono` and `Flux`) +* [ReactiveX (RxJava)](https://reactivex.io) (`Flowable` only) + +To use these wrappers, add the `feign-reactive-wrappers` module, and your desired `reactive-streams` +implementation to your classpath. Then configure Feign to use the reactive streams wrappers. + +```java +public interface GitHubReactor { + + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Flux contributors(@Param("owner") String owner, @Param("repo") String repo); + + class Contributor { + String login; + + public Contributor(String login) { + this.login = login; + } + } +} + +public class ExampleReactor { + public static void main(String args[]) { + GitHubReactor gitHub = ReactorFeign.builder() + .target(GitHubReactor.class, "https://api.github.com"); + + List contributors = gitHub.contributors("OpenFeign", "feign") + .map(Contributor::new) + .collect(Collectors.toList()) + .block(); + } +} + +public interface GitHubReactiveX { + + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Flowable contributors(@Param("owner") String owner, @Param("repo") String repo); + + class Contributor { + String login; + + public Contributor(String login) { + this.login = login; + } + } +} + +public class ExampleRxJava2 { + public static void main(String args[]) { + GitHubReactiveX gitHub = RxJavaFeign.builder() + .target(GitHub.class, "https://api.github.com"); + + List contributors = gitHub.contributors("OpenFeign", "feign") + .map(Contributor::new) + .collect(Collectors.toList()) + .block(); + } +} + +``` + +Considerations +--- + +These wrappers are not *reactive all the way down*, given that Feign generated requests are +synchronous. Requests still block, but execution is controlled by the `Publisher` and their +related `Scheduler`. While this may not be ideal in terms of a fully reactive application, providing these +wrappers provide an intermediate upgrade path for Feign. + +### Streaming + +Methods that return `java.util.streams` Types are not supported. Responses are read fully, +the wrapped in the appropriate reactive wrappers. + +### Iterable and Collections responses + +Due to the Synchronous nature of Feign requests, methods that return `Iterable` types must specify the collection +in the `Publisher`. For `Reactor` types, this limits the use of `Flux` as a response type. If you +want to use `Flux`, you will need to manually convert the `Mono` or `Iterable` response types into +`Flux` using the `fromIterable` method. + + +```java +public interface GitHub { + + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Mono> contributors(@Param("owner") String owner, @Param("repo") String repo); + + class Contributor { + String login; + + public Contributor(String login) { + this.login = login; + } + } +} + +public class ExampleApplication { + public static void main(String[] args) { + GitHub gitHub = ReactorFeign.builder() + .target(GitHub.class, "https://api.github.com"); + + Mono> contributors = gitHub.contributors("OpenFeign", "feign"); + Flux contributorFlux = Flux.fromIterable(contributors.block()); + } +} +``` diff --git a/reactive/pom.xml b/reactive/pom.xml new file mode 100644 index 0000000000..a80300bee0 --- /dev/null +++ b/reactive/pom.xml @@ -0,0 +1,95 @@ + + + + + io.github.openfeign + parent + 10.0.2-SNAPSHOT + + 4.0.0 + + Feign Reactive Wrappers + feign-reactive-wrappers + Reactive Wrapper for Feign Clients + + + ${project.basedir}/.. + 3.1.8.RELEASE + 1.0.2 + 2.2.2 + 1.9.5 + + + + + io.github.openfeign + feign-core + ${project.version} + + + org.reactivestreams + reactive-streams + ${reactive.streams.version} + + + io.projectreactor + reactor-core + ${reactor.version} + provided + true + + + io.reactivex.rxjava2 + rxjava + ${reactivex.version} + provided + true + + + org.mockito + mockito-all + ${mockito.version} + test + + + io.github.openfeign + feign-jackson + ${project.version} + test + + + io.github.openfeign + feign-okhttp + ${project.version} + test + + + io.github.openfeign + feign-jaxrs + ${project.version} + test + + + com.squareup.okhttp3 + mockwebserver + test + + + + diff --git a/reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java b/reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java new file mode 100644 index 0000000000..286c8893b2 --- /dev/null +++ b/reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java @@ -0,0 +1,81 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.Contract; +import feign.MethodMetadata; +import feign.Types; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Stream; +import org.reactivestreams.Publisher; + +public class ReactiveDelegatingContract implements Contract { + + private final Contract delegate; + + ReactiveDelegatingContract(Contract delegate) { + this.delegate = delegate; + } + + @Override + public List parseAndValidatateMetadata(Class targetType) { + List methodsMetadata = this.delegate.parseAndValidatateMetadata(targetType); + + for (final MethodMetadata metadata : methodsMetadata) { + final Type type = metadata.returnType(); + if (!isReactive(type)) { + throw new IllegalArgumentException(String.format( + "Method %s of contract %s doesn't returns a org.reactivestreams.Publisher", + metadata.configKey(), targetType.getSimpleName())); + } + + /* + * we will need to change the return type of the method to match the return type contained + * within the Publisher + */ + Type[] actualTypes = ((ParameterizedType) type).getActualTypeArguments(); + if (actualTypes.length > 1) { + throw new IllegalStateException("Expected only one contained type."); + } else { + Class actual = Types.getRawType(actualTypes[0]); + if (Stream.class.isAssignableFrom(actual)) { + throw new IllegalArgumentException( + "Streams are not supported when using Reactive Wrappers"); + } + metadata.returnType(actualTypes[0]); + } + } + + return methodsMetadata; + } + + /** + * Ensure that the type provided implements a Reactive Streams Publisher. + * + * @param type to inspect. + * @return true if the type implements the Reactive Streams Publisher specification. + */ + private boolean isReactive(Type type) { + if (!ParameterizedType.class.isAssignableFrom(type.getClass())) { + return false; + } + ParameterizedType parameterizedType = (ParameterizedType) type; + Type raw = parameterizedType.getRawType(); + return Arrays.asList(((Class) raw).getInterfaces()) + .contains(Publisher.class); + } +} diff --git a/reactive/src/main/java/feign/reactive/ReactiveFeign.java b/reactive/src/main/java/feign/reactive/ReactiveFeign.java new file mode 100644 index 0000000000..5f78ea1ed0 --- /dev/null +++ b/reactive/src/main/java/feign/reactive/ReactiveFeign.java @@ -0,0 +1,61 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.Contract; +import feign.Feign; +import feign.InvocationHandlerFactory; + +abstract class ReactiveFeign { + + + + public static class Builder extends Feign.Builder { + + private Contract contract = new Contract.Default(); + + /** + * Extend the current contract to support Reactive Stream return types. + * + * @param contract to extend. + * @return a Builder for chaining. + */ + @Override + public Builder contract(Contract contract) { + this.contract = contract; + return this; + } + + /** + * Build the Feign instance. + * + * @return a new Feign Instance. + */ + @Override + public Feign build() { + if (!(this.contract instanceof ReactiveDelegatingContract)) { + super.contract(new ReactiveDelegatingContract(this.contract)); + } else { + super.contract(this.contract); + } + return super.build(); + } + + @Override + public Feign.Builder doNotCloseAfterDecode() { + throw new UnsupportedOperationException("Streaming Decoding is not supported."); + } + } +} diff --git a/reactive/src/main/java/feign/reactive/ReactiveInvocationHandler.java b/reactive/src/main/java/feign/reactive/ReactiveInvocationHandler.java new file mode 100644 index 0000000000..ff851275b3 --- /dev/null +++ b/reactive/src/main/java/feign/reactive/ReactiveInvocationHandler.java @@ -0,0 +1,111 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.FeignException; +import feign.InvocationHandlerFactory.MethodHandler; +import feign.Target; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.lang.reflect.Type; +import java.util.Map; +import java.util.concurrent.Callable; +import org.reactivestreams.Publisher; + +public abstract class ReactiveInvocationHandler implements InvocationHandler { + + private final Target target; + private final Map dispatch; + + public ReactiveInvocationHandler(Target target, + Map dispatch) { + this.target = target; + this.dispatch = dispatch; + } + + @Override + public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if ("equals".equals(method.getName())) { + try { + Object otherHandler = + args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null; + return equals(otherHandler); + } catch (IllegalArgumentException e) { + return false; + } + } else if ("hashCode".equals(method.getName())) { + return hashCode(); + } else if ("toString".equals(method.getName())) { + return toString(); + } + return this.invoke(method, this.dispatch.get(method), args); + } + + @Override + public int hashCode() { + return this.target.hashCode(); + } + + @Override + public boolean equals(Object obj) { + if (obj == null) { + return false; + } + if (obj == this) { + return true; + } + if (ReactiveInvocationHandler.class.isAssignableFrom(obj.getClass())) { + return this.target.equals(obj); + } + return false; + } + + @Override + public String toString() { + return "Target [" + this.target.toString() + "]"; + } + + /** + * Invoke the Method Handler. + * + * @param method on the Target to invoke. + * @param methodHandler to invoke + * @param arguments for the method + * @return a reactive {@link Publisher} for the invocation. + */ + protected abstract Publisher invoke(Method method, + MethodHandler methodHandler, + Object[] arguments); + + /** + * Invoke the Method Handler as a Callable. + * + * @param methodHandler to invoke + * @param arguments for the method + * @return a Callable wrapper for the invocation. + */ + Callable invokeMethod(MethodHandler methodHandler, Object[] arguments) { + return () -> { + try { + return methodHandler.invoke(arguments); + } catch (Throwable th) { + if (th instanceof FeignException) { + throw (FeignException) th; + } + throw new RuntimeException(th); + } + }; + } +} diff --git a/reactive/src/main/java/feign/reactive/ReactorFeign.java b/reactive/src/main/java/feign/reactive/ReactorFeign.java new file mode 100644 index 0000000000..b3bc1ebc56 --- /dev/null +++ b/reactive/src/main/java/feign/reactive/ReactorFeign.java @@ -0,0 +1,53 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.Feign; +import feign.reactive.ReactiveFeign.Builder; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.util.Map; +import feign.InvocationHandlerFactory; +import feign.Target; + +public class ReactorFeign extends ReactiveFeign { + + public static Builder builder() { + return new Builder(); + } + + public static class Builder extends ReactiveFeign.Builder { + + @Override + public Feign build() { + super.invocationHandlerFactory(new ReactorInvocationHandlerFactory()); + return super.build(); + } + + @Override + public Feign.Builder invocationHandlerFactory( + InvocationHandlerFactory invocationHandlerFactory) { + throw new UnsupportedOperationException( + "Invocation Handler Factory overrides are not supported."); + } + } + + private static class ReactorInvocationHandlerFactory implements InvocationHandlerFactory { + @Override + public InvocationHandler create(Target target, Map dispatch) { + return new ReactorInvocationHandler(target, dispatch); + } + } +} diff --git a/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java b/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java new file mode 100644 index 0000000000..1ce1bae39a --- /dev/null +++ b/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java @@ -0,0 +1,44 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.InvocationHandlerFactory.MethodHandler; +import feign.Target; +import java.lang.reflect.Method; +import java.util.Map; +import java.util.concurrent.Callable; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +public class ReactorInvocationHandler extends ReactiveInvocationHandler { + + ReactorInvocationHandler(Target target, + Map dispatch) { + super(target, dispatch); + } + + @Override + protected Publisher invoke(Method method, MethodHandler methodHandler, Object[] arguments) { + Callable invocation = this.invokeMethod(methodHandler, arguments); + if (Flux.class.isAssignableFrom(method.getReturnType())) { + return Flux.from(Mono.fromCallable(invocation)).subscribeOn(Schedulers.elastic()); + } else if (Mono.class.isAssignableFrom(method.getReturnType())) { + return Mono.fromCallable(invocation).subscribeOn(Schedulers.elastic()); + } + throw new IllegalArgumentException( + "Return type " + method.getReturnType().getName() + " is not supported"); + } +} diff --git a/reactive/src/main/java/feign/reactive/RxJavaFeign.java b/reactive/src/main/java/feign/reactive/RxJavaFeign.java new file mode 100644 index 0000000000..7405d02bbc --- /dev/null +++ b/reactive/src/main/java/feign/reactive/RxJavaFeign.java @@ -0,0 +1,53 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.util.Map; +import feign.Feign; +import feign.InvocationHandlerFactory; +import feign.Target; + +public class RxJavaFeign extends ReactiveFeign { + + public static Builder builder() { + return new Builder(); + } + + public static class Builder extends ReactiveFeign.Builder { + + @Override + public Feign build() { + super.invocationHandlerFactory(new RxJavaInvocationHandlerFactory()); + return super.build(); + } + + @Override + public Feign.Builder invocationHandlerFactory( + InvocationHandlerFactory invocationHandlerFactory) { + throw new UnsupportedOperationException( + "Invocation Handler Factory overrides are not supported."); + } + + } + + private static class RxJavaInvocationHandlerFactory implements InvocationHandlerFactory { + @Override + public InvocationHandler create(Target target, Map dispatch) { + return new RxJavaInvocationHandler(target, dispatch); + } + } + +} diff --git a/reactive/src/main/java/feign/reactive/RxJavaInvocationHandler.java b/reactive/src/main/java/feign/reactive/RxJavaInvocationHandler.java new file mode 100644 index 0000000000..e4893136ac --- /dev/null +++ b/reactive/src/main/java/feign/reactive/RxJavaInvocationHandler.java @@ -0,0 +1,36 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.InvocationHandlerFactory.MethodHandler; +import feign.Target; +import io.reactivex.Flowable; +import io.reactivex.schedulers.Schedulers; +import java.lang.reflect.Method; +import java.util.Map; +import org.reactivestreams.Publisher; + +public class RxJavaInvocationHandler extends ReactiveInvocationHandler { + + RxJavaInvocationHandler(Target target, + Map dispatch) { + super(target, dispatch); + } + + @Override + protected Publisher invoke(Method method, MethodHandler methodHandler, Object[] arguments) { + return Flowable.fromCallable(this.invokeMethod(methodHandler, arguments)) + .observeOn(Schedulers.trampoline()); + } +} diff --git a/reactive/src/test/java/feign/reactive/ReactiveDelegatingContractTest.java b/reactive/src/test/java/feign/reactive/ReactiveDelegatingContractTest.java new file mode 100644 index 0000000000..231accbf5e --- /dev/null +++ b/reactive/src/test/java/feign/reactive/ReactiveDelegatingContractTest.java @@ -0,0 +1,85 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import feign.Contract; +import feign.Param; +import feign.RequestLine; +import feign.reactive.ReactiveDelegatingContract; +import io.reactivex.Flowable; +import java.util.stream.Stream; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +public class ReactiveDelegatingContractTest { + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + @Test + public void onlyReactiveReturnTypesSupported() { + this.thrown.expect(IllegalArgumentException.class); + Contract contract = new ReactiveDelegatingContract(new Contract.Default()); + contract.parseAndValidatateMetadata(TestSynchronousService.class); + } + + @Test + public void reactorTypes() { + Contract contract = new ReactiveDelegatingContract(new Contract.Default()); + contract.parseAndValidatateMetadata(TestReactorService.class); + } + + @Test + public void reactivexTypes() { + Contract contract = new ReactiveDelegatingContract(new Contract.Default()); + contract.parseAndValidatateMetadata(TestReactiveXService.class); + } + + @Test + public void streamsAreNotSupported() { + this.thrown.expect(IllegalArgumentException.class); + Contract contract = new ReactiveDelegatingContract(new Contract.Default()); + contract.parseAndValidatateMetadata(StreamsService.class); + } + + public interface TestSynchronousService { + @RequestLine("GET /version") + String version(); + } + + public interface TestReactiveXService { + @RequestLine("GET /version") + Flowable version(); + } + + + public interface TestReactorService { + @RequestLine("GET /version") + Mono version(); + + @RequestLine("GET /users/{username}") + Flux user(@Param("username") String username); + } + + public interface StreamsService { + + @RequestLine("GET /version") + Mono> version(); + } + +} diff --git a/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java b/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java new file mode 100644 index 0000000000..437dfdc34d --- /dev/null +++ b/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java @@ -0,0 +1,342 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import feign.Client; +import feign.InvocationHandlerFactory; +import feign.Logger; +import feign.Logger.Level; +import feign.Param; +import feign.QueryMap; +import feign.QueryMapEncoder; +import feign.Request; +import feign.Request.Options; +import feign.RequestInterceptor; +import feign.RequestLine; +import feign.RequestTemplate; +import feign.Response; +import feign.ResponseMapper; +import feign.RetryableException; +import feign.Retryer; +import feign.Target; +import feign.codec.Decoder; +import feign.codec.ErrorDecoder; +import feign.jackson.JacksonDecoder; +import feign.jackson.JacksonEncoder; +import feign.jaxrs.JAXRSContract; +import io.reactivex.Flowable; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Type; +import java.nio.charset.Charset; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import javax.ws.rs.GET; +import javax.ws.rs.Path; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.mockito.AdditionalAnswers; +import org.mockito.stubbing.Answer; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +public class ReactiveFeignIntegrationTest { + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + @Rule + public final MockWebServer webServer = new MockWebServer(); + + private String getServerUrl() { + return "http://localhost:" + this.webServer.getPort(); + } + + @Test + public void testDefaultMethodsNotProxied() { + TestReactorService service = ReactorFeign.builder() + .target(TestReactorService.class, this.getServerUrl()); + assertThat(service).isEqualTo(service); + assertThat(service.toString()).isNotNull(); + assertThat(service.hashCode()).isNotZero(); + } + + @Test + public void testReactorTargetFull() throws Exception { + this.webServer.enqueue(new MockResponse().setBody("1.0")); + this.webServer.enqueue(new MockResponse().setBody("{ \"username\": \"test\" }")); + + TestReactorService service = ReactorFeign.builder() + .encoder(new JacksonEncoder()) + .decoder(new JacksonDecoder()) + .logger(new ConsoleLogger()) + .decode404() + .options(new Options()) + .logLevel(Level.FULL) + .target(TestReactorService.class, this.getServerUrl()); + assertThat(service).isNotNull(); + + String version = service.version() + .block(); + assertThat(version).isNotNull(); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/version"); + + + /* test encoding and decoding */ + User user = service.user("test") + .blockFirst(); + assertThat(user).isNotNull().hasFieldOrPropertyWithValue("username", "test"); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users/test"); + + } + + @Test + public void testRxJavaTarget() throws Exception { + this.webServer.enqueue(new MockResponse().setBody("1.0")); + this.webServer.enqueue(new MockResponse().setBody("{ \"username\": \"test\" }")); + + TestReactiveXService service = RxJavaFeign.builder() + .encoder(new JacksonEncoder()) + .decoder(new JacksonDecoder()) + .logger(new ConsoleLogger()) + .logLevel(Level.FULL) + .target(TestReactiveXService.class, this.getServerUrl()); + assertThat(service).isNotNull(); + + String version = service.version() + .firstElement().blockingGet(); + assertThat(version).isNotNull(); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/version"); + + /* test encoding and decoding */ + User user = service.user("test") + .firstElement().blockingGet(); + assertThat(user).isNotNull().hasFieldOrPropertyWithValue("username", "test"); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users/test"); + } + + @Test + public void invocationFactoryIsNotSupported() { + this.thrown.expect(UnsupportedOperationException.class); + ReactorFeign.builder() + .invocationHandlerFactory( + (target, dispatch) -> null) + .target(TestReactiveXService.class, "http://localhost"); + } + + @Test + public void doNotCloseUnsupported() { + this.thrown.expect(UnsupportedOperationException.class); + ReactorFeign.builder() + .doNotCloseAfterDecode() + .target(TestReactiveXService.class, "http://localhost"); + } + + @Test + public void testRequestInterceptor() { + this.webServer.enqueue(new MockResponse().setBody("1.0")); + + RequestInterceptor mockInterceptor = mock(RequestInterceptor.class); + TestReactorService service = ReactorFeign.builder() + .requestInterceptor(mockInterceptor) + .target(TestReactorService.class, this.getServerUrl()); + service.version().block(); + verify(mockInterceptor, times(1)).apply(any(RequestTemplate.class)); + } + + @Test + public void testRequestInterceptors() { + this.webServer.enqueue(new MockResponse().setBody("1.0")); + + RequestInterceptor mockInterceptor = mock(RequestInterceptor.class); + TestReactorService service = ReactorFeign.builder() + .requestInterceptors(Arrays.asList(mockInterceptor, mockInterceptor)) + .target(TestReactorService.class, this.getServerUrl()); + service.version().block(); + verify(mockInterceptor, times(2)).apply(any(RequestTemplate.class)); + } + + @Test + public void testResponseMappers() throws Exception { + this.webServer.enqueue(new MockResponse().setBody("1.0")); + + ResponseMapper responseMapper = mock(ResponseMapper.class); + Decoder decoder = mock(Decoder.class); + given(responseMapper.map(any(Response.class), any(Type.class))) + .willAnswer(AdditionalAnswers.returnsFirstArg()); + given(decoder.decode(any(Response.class), any(Type.class))).willReturn("1.0"); + + TestReactorService service = ReactorFeign.builder() + .mapAndDecode(responseMapper, decoder) + .target(TestReactorService.class, this.getServerUrl()); + service.version().block(); + verify(responseMapper, times(1)) + .map(any(Response.class), any(Type.class)); + verify(decoder, times(1)).decode(any(Response.class), any(Type.class)); + } + + @Test + public void testQueryMapEncoders() { + this.webServer.enqueue(new MockResponse().setBody("No Results Found")); + + QueryMapEncoder encoder = mock(QueryMapEncoder.class); + given(encoder.encode(any(Object.class))).willReturn(Collections.emptyMap()); + TestReactiveXService service = RxJavaFeign.builder() + .queryMapEncoder(encoder) + .target(TestReactiveXService.class, this.getServerUrl()); + String results = service.search(new SearchQuery()) + .blockingSingle(); + assertThat(results).isNotEmpty(); + verify(encoder, times(1)).encode(any(Object.class)); + } + + @SuppressWarnings({"ResultOfMethodCallIgnored", "ThrowableNotThrown"}) + @Test + public void testErrorDecoder() { + this.thrown.expect(RuntimeException.class); + this.webServer.enqueue(new MockResponse().setBody("Bad Request").setResponseCode(400)); + + ErrorDecoder errorDecoder = mock(ErrorDecoder.class); + given(errorDecoder.decode(anyString(), any(Response.class))) + .willReturn(new IllegalStateException("bad request")); + + TestReactiveXService service = RxJavaFeign.builder() + .errorDecoder(errorDecoder) + .target(TestReactiveXService.class, this.getServerUrl()); + service.search(new SearchQuery()) + .blockingSingle(); + verify(errorDecoder, times(1)).decode(anyString(), any(Response.class)); + } + + @Test + public void testRetryer() { + this.webServer.enqueue(new MockResponse().setBody("Not Available").setResponseCode(-1)); + this.webServer.enqueue(new MockResponse().setBody("1.0")); + + Retryer retryer = new Retryer.Default(); + Retryer spy = spy(retryer); + when(spy.clone()).thenReturn(spy); + TestReactorService service = ReactorFeign.builder() + .retryer(spy) + .target(TestReactorService.class, this.getServerUrl()); + service.version().log().block(); + verify(spy, times(1)).continueOrPropagate(any(RetryableException.class)); + } + + @Test + public void testClient() throws Exception { + Client client = mock(Client.class); + given(client.execute(any(Request.class), any(Options.class))) + .willAnswer((Answer) invocation -> Response.builder() + .status(200) + .headers(Collections.emptyMap()) + .body("1.0", Charset.defaultCharset()) + .request((Request) invocation.getArguments()[0]) + .build()); + + TestReactorService service = ReactorFeign.builder() + .client(client) + .target(TestReactorService.class, this.getServerUrl()); + service.version().block(); + verify(client, times(1)).execute(any(Request.class), any(Options.class)); + } + + @Test + public void testDifferentContract() throws Exception { + this.webServer.enqueue(new MockResponse().setBody("1.0")); + + TestJaxRSReactorService service = ReactorFeign.builder() + .contract(new JAXRSContract()) + .target(TestJaxRSReactorService.class, this.getServerUrl()); + String version = service.version().block(); + assertThat(version).isNotNull(); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/version"); + } + + + interface TestReactorService { + @RequestLine("GET /version") + Mono version(); + + @RequestLine("GET /users/{username}") + Flux user(@Param("username") String username); + } + + + interface TestReactiveXService { + @RequestLine("GET /version") + Flowable version(); + + @RequestLine("GET /users/{username}") + Flowable user(@Param("username") String username); + + @RequestLine("GET /users/search") + Flowable search(@QueryMap SearchQuery query); + } + + interface TestJaxRSReactorService { + + @Path("/version") + @GET + Mono version(); + } + + + @SuppressWarnings("unused") + static class User { + private String username; + + public User() { + super(); + } + + public String getUsername() { + return username; + } + } + + + @SuppressWarnings("unused") + static class SearchQuery { + SearchQuery() { + super(); + } + + public String query() { + return "query"; + } + } + + + public static class ConsoleLogger extends Logger { + @Override + protected void log(String configKey, String format, Object... args) { + System.out.println(String.format(methodTag(configKey) + format, args)); + } + } +} diff --git a/reactive/src/test/java/feign/reactive/ReactiveInvocationHandlerTest.java b/reactive/src/test/java/feign/reactive/ReactiveInvocationHandlerTest.java new file mode 100644 index 0000000000..25b4e01b80 --- /dev/null +++ b/reactive/src/test/java/feign/reactive/ReactiveInvocationHandlerTest.java @@ -0,0 +1,132 @@ +/** + * Copyright 2012-2018 The Feign 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 feign.reactive; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; +import feign.FeignException; +import feign.InvocationHandlerFactory.MethodHandler; +import feign.RequestLine; +import feign.Target; +import feign.reactive.ReactorInvocationHandler; +import feign.reactive.RxJavaInvocationHandler; +import io.reactivex.Flowable; +import java.lang.reflect.Method; +import java.util.Collections; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.runners.MockitoJUnitRunner; +import reactor.core.publisher.Mono; + +@RunWith(MockitoJUnitRunner.class) +public class ReactiveInvocationHandlerTest { + + @Rule + public ExpectedException thrown = ExpectedException.none(); + + @Mock + private Target target; + + @Mock + private MethodHandler methodHandler; + + private Method method; + + @SuppressWarnings("unchecked") + @Test + public void invokeOnSubscribeReactor() throws Throwable { + Method method = TestReactorService.class.getMethod("version"); + ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target, + Collections.singletonMap(method, this.methodHandler)); + + Object result = handler.invoke(method, this.methodHandler, new Object[] {}); + assertThat(result).isInstanceOf(Mono.class); + verifyZeroInteractions(this.methodHandler); + + /* subscribe and execute the method */ + Mono mono = (Mono) result; + mono.log().block(); + verify(this.methodHandler, times(1)).invoke(any()); + } + + @SuppressWarnings("unchecked") + @Test + public void invokeFailureReactor() throws Throwable { + this.thrown.expect(RuntimeException.class); + given(this.methodHandler.invoke(any())).willThrow(new RuntimeException("Could Not Decode")); + given(this.method.getReturnType()).willReturn((Class) Class.forName(Mono.class.getName())); + ReactorInvocationHandler handler = new ReactorInvocationHandler(this.target, + Collections.singletonMap(this.method, this.methodHandler)); + + Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); + assertThat(result).isInstanceOf(Mono.class); + verifyZeroInteractions(this.methodHandler); + + /* subscribe and execute the method, should result in an error */ + Mono mono = (Mono) result; + mono.log().block(); + verify(this.methodHandler, times(1)).invoke(any()); + } + + @SuppressWarnings("ResultOfMethodCallIgnored") + @Test + public void invokeOnSubscribeRxJava() throws Throwable { + given(this.methodHandler.invoke(any())).willReturn("Result"); + RxJavaInvocationHandler handler = + new RxJavaInvocationHandler(this.target, + Collections.singletonMap(this.method, this.methodHandler)); + + Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); + assertThat(result).isInstanceOf(Flowable.class); + verifyZeroInteractions(this.methodHandler); + + /* subscribe and execute the method */ + Flowable flow = (Flowable) result; + flow.firstElement().blockingGet(); + verify(this.methodHandler, times(1)).invoke(any()); + } + + @SuppressWarnings("ResultOfMethodCallIgnored") + @Test + public void invokeFailureRxJava() throws Throwable { + this.thrown.expect(RuntimeException.class); + given(this.methodHandler.invoke(any())).willThrow(new RuntimeException("Could Not Decode")); + RxJavaInvocationHandler handler = + new RxJavaInvocationHandler(this.target, + Collections.singletonMap(this.method, this.methodHandler)); + + Object result = handler.invoke(this.method, this.methodHandler, new Object[] {}); + assertThat(result).isInstanceOf(Flowable.class); + verifyZeroInteractions(this.methodHandler); + + /* subscribe and execute the method */ + Flowable flow = (Flowable) result; + flow.firstElement().blockingGet(); + verify(this.methodHandler, times(1)).invoke(any()); + } + + + public interface TestReactorService { + @RequestLine("GET /version") + Mono version(); + } + +}