diff --git a/core/src/main/java/feign/Types.java b/core/src/main/java/feign/Types.java index e9d30bc9c..bdad6f814 100644 --- a/core/src/main/java/feign/Types.java +++ b/core/src/main/java/feign/Types.java @@ -23,6 +23,8 @@ import java.util.Arrays; import java.util.NoSuchElementException; +import static feign.Util.checkState; + /** * Static methods for working with types. * @@ -325,6 +327,40 @@ public static Type resolveReturnType(Type baseType, Type overridingType) { return baseType; } + /** + * Resolves the last type parameter of the parameterized {@code supertype}, based on the {@code + * genericContext}, into its upper bounds. + *

+ * Implementation copied from {@code + * retrofit.RestMethodInfo}. + * + * @param genericContext Ex. {@link java.lang.reflect.Field#getGenericType()} + * @param supertype Ex. {@code Decoder.class} + * @return in the example above, the type parameter of {@code Decoder}. + * @throws IllegalStateException if {@code supertype} cannot be resolved into a parameterized type + * using {@code context}. + */ + public static Type resolveLastTypeParameter(Type genericContext, Class supertype) + throws IllegalStateException { + Type resolvedSuperType = + Types.getSupertype(genericContext, Types.getRawType(genericContext), supertype); + checkState(resolvedSuperType instanceof ParameterizedType, + "could not resolve %s into a parameterized type %s", + genericContext, supertype); + Type[] types = ParameterizedType.class.cast(resolvedSuperType).getActualTypeArguments(); + for (int i = 0; i < types.length; i++) { + Type type = types[i]; + if (type instanceof WildcardType) { + types[i] = ((WildcardType) type).getUpperBounds()[0]; + } + } + return types[types.length - 1]; + } + + public static ParameterizedType parameterize(Class rawClass, Type... typeArguments) { + return new ParameterizedTypeImpl(rawClass.getEnclosingClass(), rawClass, typeArguments); + } + static final class ParameterizedTypeImpl implements ParameterizedType { private final Type ownerType; diff --git a/core/src/main/java/feign/Util.java b/core/src/main/java/feign/Util.java index cb542d73f..a82b075a1 100644 --- a/core/src/main/java/feign/Util.java +++ b/core/src/main/java/feign/Util.java @@ -215,33 +215,12 @@ public static void ensureClosed(Closeable closeable) { } /** - * Resolves the last type parameter of the parameterized {@code supertype}, based on the {@code - * genericContext}, into its upper bounds. - *

- * Implementation copied from {@code - * retrofit.RestMethodInfo}. - * - * @param genericContext Ex. {@link java.lang.reflect.Field#getGenericType()} - * @param supertype Ex. {@code Decoder.class} - * @return in the example above, the type parameter of {@code Decoder}. - * @throws IllegalStateException if {@code supertype} cannot be resolved into a parameterized type - * using {@code context}. + * Moved to {@code feign.Types.resolveLastTypeParameter} */ + @Deprecated public static Type resolveLastTypeParameter(Type genericContext, Class supertype) throws IllegalStateException { - Type resolvedSuperType = - Types.getSupertype(genericContext, Types.getRawType(genericContext), supertype); - checkState(resolvedSuperType instanceof ParameterizedType, - "could not resolve %s into a parameterized type %s", - genericContext, supertype); - Type[] types = ParameterizedType.class.cast(resolvedSuperType).getActualTypeArguments(); - for (int i = 0; i < types.length; i++) { - Type type = types[i]; - if (type instanceof WildcardType) { - types[i] = ((WildcardType) type).getUpperBounds()[0]; - } - } - return types[types.length - 1]; + return Types.resolveLastTypeParameter(genericContext, supertype); } /** diff --git a/reactive/README.md b/reactive/README.md index 0937060c6..772a64583 100644 --- a/reactive/README.md +++ b/reactive/README.md @@ -14,24 +14,34 @@ implementation to your classpath. Then configure Feign to use the reactive stre public interface GitHubReactor { @RequestLine("GET /repos/{owner}/{repo}/contributors") - Flux contributors(@Param("owner") String owner, @Param("repo") String repo); + Flux contributorsFlux(@Param("owner") String owner, @Param("repo") String repo); + + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Mono> contributorsMono(@Param("owner") String owner, @Param("repo") String repo); class Contributor { - String login; - - public Contributor(String login) { - this.login = login; - } + String login; + + public String getLogin() { + return login; + } + + public void setLogin(String login) { + this.login = login; + } } } public class ExampleReactor { public static void main(String args[]) { - GitHubReactor gitHub = ReactorFeign.builder() + GitHubReactor gitHub = ReactorFeign.builder() + .decoder(new ReactorDecoder(new JacksonDecoder())) .target(GitHubReactor.class, "https://api.github.com"); - List contributors = gitHub.contributors("OpenFeign", "feign") - .collect(Collectors.toList()) + List contributorsFromFlux = gitHub.contributorsFlux("OpenFeign", "feign") + .collectList() + .block(); + List contributorsFromMono = gitHub.contributorsMono("OpenFeign", "feign") .block(); } } @@ -52,7 +62,8 @@ public interface GitHubReactiveX { public class ExampleRxJava2 { public static void main(String args[]) { - GitHubReactiveX gitHub = RxJavaFeign.builder() + GitHubReactiveX gitHub = RxJavaFeign.builder() + .decoder(new RxJavaDecoder(new JacksonDecoder())) .target(GitHub.class, "https://api.github.com"); List contributors = gitHub.contributors("OpenFeign", "feign") @@ -79,33 +90,5 @@ 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. - +in the `Publisher`. For `Reactor` types, this limits the use of `Flux` as a response type. -```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/src/main/java/feign/reactive/ReactiveDelegatingContract.java b/reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java index 62b2d48b0..ca76327ac 100644 --- a/reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java +++ b/reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java @@ -55,7 +55,7 @@ public List parseAndValidateMetadata(Class targetType) { throw new IllegalArgumentException( "Streams are not supported when using Reactive Wrappers"); } - metadata.returnType(actualTypes[0]); + metadata.returnType(type); } } diff --git a/reactive/src/main/java/feign/reactive/ReactorDecoder.java b/reactive/src/main/java/feign/reactive/ReactorDecoder.java new file mode 100644 index 000000000..d165cd0e8 --- /dev/null +++ b/reactive/src/main/java/feign/reactive/ReactorDecoder.java @@ -0,0 +1,50 @@ +/* + * Copyright 2012-2023 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.Response; +import feign.Types; +import feign.codec.Decoder; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.io.IOException; +import java.lang.reflect.Type; +import java.util.List; + +public class ReactorDecoder implements Decoder { + + private final Decoder delegate; + + public ReactorDecoder(Decoder decoder) { + this.delegate = decoder; + } + + @Override + public Object decode(Response response, Type type) throws IOException, FeignException { + Class rawType = Types.getRawType(type); + if (rawType.isAssignableFrom(Mono.class)) { + Type lastType = Types.resolveLastTypeParameter(type, Mono.class); + return delegate.decode(response, lastType); + } + if (rawType.isAssignableFrom(Flux.class)) { + Type lastType = Types.resolveLastTypeParameter(type, Flux.class); + Type listType = Types.parameterize(List.class, lastType); + return delegate.decode(response, listType); + } + + return delegate.decode(response, type); + } +} diff --git a/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java b/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java index cdd2645e5..7a61d68b8 100644 --- a/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java +++ b/reactive/src/main/java/feign/reactive/ReactorInvocationHandler.java @@ -36,7 +36,7 @@ public class ReactorInvocationHandler extends ReactiveInvocationHandler { 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(scheduler); + return Flux.from(invocation).flatMapIterable(e -> (Iterable) e).subscribeOn(scheduler); } else if (Mono.class.isAssignableFrom(method.getReturnType())) { return Mono.from(invocation).subscribeOn(scheduler); } diff --git a/reactive/src/main/java/feign/reactive/RxJavaDecoder.java b/reactive/src/main/java/feign/reactive/RxJavaDecoder.java new file mode 100644 index 000000000..057f3ae08 --- /dev/null +++ b/reactive/src/main/java/feign/reactive/RxJavaDecoder.java @@ -0,0 +1,43 @@ +/* + * Copyright 2012-2023 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.Response; +import feign.Types; +import feign.codec.Decoder; +import io.reactivex.Flowable; + +import java.io.IOException; +import java.lang.reflect.Type; + +public class RxJavaDecoder implements Decoder { + + private final Decoder delegate; + + public RxJavaDecoder(Decoder decoder) { + this.delegate = decoder; + } + + @Override + public Object decode(Response response, Type type) throws IOException, FeignException { + Class rawType = Types.getRawType(type); + if (rawType.isAssignableFrom(Flowable.class)) { + Type lastType = Types.resolveLastTypeParameter(type, Flowable.class); + return delegate.decode(response, lastType); + } + + return delegate.decode(response, type); + } +} diff --git a/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java b/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java index 1b3d4ea77..32727871b 100644 --- a/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java +++ b/reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java @@ -47,6 +47,7 @@ import java.nio.charset.Charset; import java.util.Arrays; import java.util.Collections; +import java.util.List; import javax.ws.rs.GET; import javax.ws.rs.Path; import okhttp3.mockwebserver.MockResponse; @@ -85,10 +86,12 @@ public void testDefaultMethodsNotProxied() { public void testReactorTargetFull() throws Exception { this.webServer.enqueue(new MockResponse().setBody("1.0")); this.webServer.enqueue(new MockResponse().setBody("{ \"username\": \"test\" }")); + this.webServer.enqueue(new MockResponse().setBody("[{ \"username\": \"test\" }]")); + this.webServer.enqueue(new MockResponse().setBody("[{ \"username\": \"test\" }]")); TestReactorService service = ReactorFeign.builder() .encoder(new JacksonEncoder()) - .decoder(new JacksonDecoder()) + .decoder(new ReactorDecoder(new JacksonDecoder())) .logger(new ConsoleLogger()) .dismiss404() .options(new Options()) @@ -102,7 +105,6 @@ public void testReactorTargetFull() throws Exception { .verify(); assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/version"); - /* test encoding and decoding */ StepVerifier.create(service.user("test")) .assertNext(user -> assertThat(user).hasFieldOrPropertyWithValue("username", "test")) @@ -110,16 +112,28 @@ public void testReactorTargetFull() throws Exception { .verify(); assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users/test"); + StepVerifier.create(service.usersFlux()) + .assertNext(user -> assertThat(user).hasFieldOrPropertyWithValue("username", "test")) + .expectComplete() + .verify(); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users"); + + StepVerifier.create(service.usersMono()) + .assertNext(users -> assertThat(users.get(0)).hasFieldOrPropertyWithValue("username", "test")) + .expectComplete() + .verify(); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users"); } @Test public void testRxJavaTarget() throws Exception { this.webServer.enqueue(new MockResponse().setBody("1.0")); this.webServer.enqueue(new MockResponse().setBody("{ \"username\": \"test\" }")); + this.webServer.enqueue(new MockResponse().setBody("[{ \"username\": \"test\" }]")); TestReactiveXService service = RxJavaFeign.builder() .encoder(new JacksonEncoder()) - .decoder(new JacksonDecoder()) + .decoder(new RxJavaDecoder(new JacksonDecoder())) .logger(new ConsoleLogger()) .logLevel(Level.FULL) .target(TestReactiveXService.class, this.getServerUrl()); @@ -137,6 +151,12 @@ public void testRxJavaTarget() throws Exception { .expectComplete() .verify(); assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users/test"); + + StepVerifier.create(service.users()) + .assertNext(users -> assertThat(users.get(0)).hasFieldOrPropertyWithValue("username", "test")) + .expectComplete() + .verify(); + assertThat(webServer.takeRequest().getPath()).isEqualToIgnoringCase("/users"); } @Test @@ -163,6 +183,7 @@ public void testRequestInterceptor() { RequestInterceptor mockInterceptor = mock(RequestInterceptor.class); TestReactorService service = ReactorFeign.builder() .requestInterceptor(mockInterceptor) + .decoder(new ReactorDecoder(new Decoder.Default())) .target(TestReactorService.class, this.getServerUrl()); StepVerifier.create(service.version()) .expectNext("1.0") @@ -178,6 +199,7 @@ public void testRequestInterceptors() { RequestInterceptor mockInterceptor = mock(RequestInterceptor.class); TestReactorService service = ReactorFeign.builder() .requestInterceptors(Arrays.asList(mockInterceptor, mockInterceptor)) + .decoder(new ReactorDecoder(new Decoder.Default())) .target(TestReactorService.class, this.getServerUrl()); StepVerifier.create(service.version()) .expectNext("1.0") @@ -216,6 +238,7 @@ public void testQueryMapEncoders() { given(encoder.encode(any(Object.class))).willReturn(Collections.emptyMap()); TestReactiveXService service = RxJavaFeign.builder() .queryMapEncoder(encoder) + .decoder(new RxJavaDecoder(new Decoder.Default())) .target(TestReactiveXService.class, this.getServerUrl()); StepVerifier.create(service.search(new SearchQuery())) .expectNext("No Results Found") @@ -254,6 +277,7 @@ public void testRetryer() { when(spy.clone()).thenReturn(spy); TestReactorService service = ReactorFeign.builder() .retryer(spy) + .decoder(new ReactorDecoder(new Decoder.Default())) .target(TestReactorService.class, this.getServerUrl()); StepVerifier.create(service.version()) .expectNext("1.0") @@ -275,6 +299,7 @@ public void testClient() throws Exception { TestReactorService service = ReactorFeign.builder() .client(client) + .decoder(new ReactorDecoder(new Decoder.Default())) .target(TestReactorService.class, this.getServerUrl()); StepVerifier.create(service.version()) .expectNext("1.0") @@ -289,6 +314,7 @@ public void testDifferentContract() throws Exception { TestJaxRSReactorService service = ReactorFeign.builder() .contract(new JAXRSContract()) + .decoder(new ReactorDecoder(new Decoder.Default())) .target(TestJaxRSReactorService.class, this.getServerUrl()); StepVerifier.create(service.version()) .expectNext("1.0") @@ -303,7 +329,13 @@ interface TestReactorService { Mono version(); @RequestLine("GET /users/{username}") - Flux user(@Param("username") String username); + Mono user(@Param("username") String username); + + @RequestLine("GET /users") + Flux usersFlux(); + + @RequestLine("GET /users") + Mono> usersMono(); } @@ -314,6 +346,9 @@ interface TestReactiveXService { @RequestLine("GET /users/{username}") Flowable user(@Param("username") String username); + @RequestLine("GET /users") + Flowable> users(); + @RequestLine("GET /users/search") Flowable search(@QueryMap SearchQuery query); } diff --git a/reactive/src/test/java/feign/reactive/examples/ConsoleLogger.java b/reactive/src/test/java/feign/reactive/examples/ConsoleLogger.java new file mode 100644 index 000000000..1323b2e54 --- /dev/null +++ b/reactive/src/test/java/feign/reactive/examples/ConsoleLogger.java @@ -0,0 +1,23 @@ +/* + * Copyright 2012-2023 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.examples; + +import feign.Logger; + +public 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/examples/ReactorGitHubExample.java b/reactive/src/test/java/feign/reactive/examples/ReactorGitHubExample.java new file mode 100644 index 000000000..9fce26a3a --- /dev/null +++ b/reactive/src/test/java/feign/reactive/examples/ReactorGitHubExample.java @@ -0,0 +1,76 @@ +/* + * Copyright 2012-2023 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.examples; + +import feign.Logger; +import feign.Param; +import feign.RequestLine; +import feign.jackson.JacksonDecoder; +import feign.reactive.ReactorDecoder; +import feign.reactive.ReactorFeign; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.util.List; + +/** + * adapted from {@code com.example.retrofit.GitHubClient} + */ +public class ReactorGitHubExample { + + public static void main(String... args) { + GitHub github = ReactorFeign.builder() + .decoder(new ReactorDecoder(new JacksonDecoder())) + .logger(new ConsoleLogger()) + .logLevel(Logger.Level.FULL) + .target(GitHub.class, "https://api.github.com"); + + System.out.println("Let's fetch and print a list of the contributors to this library (Using Flux)."); + List contributorsFromFlux = github.contributorsFlux("OpenFeign", "feign") + .collectList() + .block(); + for (Contributor contributor : contributorsFromFlux) { + System.out.println(contributor.login + " (" + contributor.contributions + ")"); + } + + System.out.println("Let's fetch and print a list of the contributors to this library (Using Mono)."); + List contributorsFromMono = github.contributorsMono("OpenFeign", "feign") + .block(); + for (Contributor contributor : contributorsFromMono) { + System.out.println(contributor.login + " (" + contributor.contributions + ")"); + } + } + + interface GitHub { + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Flux contributorsFlux(@Param("owner") String owner, @Param("repo") String repo); + + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Mono> contributorsMono(@Param("owner") String owner, @Param("repo") String repo); + } + + static class Contributor { + + private String login; + private int contributions; + + void setLogin(String login) { + this.login = login; + } + + void setContributions(int contributions) { + this.contributions = contributions; + } + } +} diff --git a/reactive/src/test/java/feign/reactive/examples/RxJavaGitHubExample.java b/reactive/src/test/java/feign/reactive/examples/RxJavaGitHubExample.java new file mode 100644 index 000000000..59f866a94 --- /dev/null +++ b/reactive/src/test/java/feign/reactive/examples/RxJavaGitHubExample.java @@ -0,0 +1,65 @@ +/* + * Copyright 2012-2023 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.examples; + +import feign.Logger; +import feign.Param; +import feign.RequestLine; +import feign.jackson.JacksonDecoder; +import feign.reactive.RxJavaDecoder; +import feign.reactive.RxJavaFeign; +import io.reactivex.Flowable; + +import java.util.List; + +/** + * adapted from {@code com.example.retrofit.GitHubClient} + */ +public class RxJavaGitHubExample { + + public static void main(String... args) { + GitHub github = RxJavaFeign.builder() + .decoder(new RxJavaDecoder(new JacksonDecoder())) + .logger(new ConsoleLogger()) + .logLevel(Logger.Level.FULL) + .target(GitHub.class, "https://api.github.com"); + + System.out.println("Let's fetch and print a list of the contributors to this library."); + List contributorsFromFlux = github.contributors("OpenFeign", "feign") + .blockingLast(); + for (Contributor contributor : contributorsFromFlux) { + System.out.println(contributor.login + " (" + contributor.contributions + ")"); + } + + } + + interface GitHub { + @RequestLine("GET /repos/{owner}/{repo}/contributors") + Flowable> contributors(@Param("owner") String owner, @Param("repo") String repo); + } + + static class Contributor { + + private String login; + private int contributions; + + void setLogin(String login) { + this.login = login; + } + + void setContributions(int contributions) { + this.contributions = contributions; + } + } +}