Flux type response should be corresponding to List (#2199)

* Refactoring: Move resolveLastTypeParameter from Util to Types

* Add ReactiveDecoder

* Update code as per suggestions

* Update code as per suggestions

* Refactoring

* Add tests

core/src/main/java/feign/Types.java

@@ -23,6 +23,8 @@ import java.lang.reflect.WildcardType;
 import java.util.Arrays;
 import java.util.NoSuchElementException;
 
+import static feign.Util.checkState;
+
 /**
  * Static methods for working with types.
  *
@@ -325,6 +327,40 @@ public final class Types {
     return baseType;
   }
 
+  /**
+   * Resolves the last type parameter of the parameterized {@code supertype}, based on the {@code
+   * genericContext}, into its upper bounds.
+   * <p/>
+   * 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;

core/src/main/java/feign/Util.java

@@ -215,33 +215,12 @@ public class Util {
   }
 
   /**
-   * Resolves the last type parameter of the parameterized {@code supertype}, based on the {@code
-   * genericContext}, into its upper bounds.
-   * <p/>
-   * 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);
   }
 
   /**

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<Contributor> contributors(@Param("owner") String owner, @Param("repo") String repo);
+  Flux<Contributor> contributorsFlux(@Param("owner") String owner, @Param("repo") String repo);
+      
+  @RequestLine("GET /repos/{owner}/{repo}/contributors")
+  Mono<List<Contributor>> 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<Contributor> contributors = gitHub.contributors("OpenFeign", "feign")
-      .collect(Collectors.toList())
+    List<GitHubReactor.Contributor> contributorsFromFlux = gitHub.contributorsFlux("OpenFeign", "feign")
+      .collectList()
+      .block();
+    List<GitHubReactor.Contributor> 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<Contributor> 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<List<Contributor>> 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<List<Contributor>> contributors = gitHub.contributors("OpenFeign", "feign");
-    Flux<Contributor> contributorFlux = Flux.fromIterable(contributors.block());
-  }
-}
-```

reactive/src/main/java/feign/reactive/ReactiveDelegatingContract.java

@@ -55,7 +55,7 @@ public class ReactiveDelegatingContract implements Contract {
           throw new IllegalArgumentException(
               "Streams are not supported when using Reactive Wrappers");
         }
-        metadata.returnType(actualTypes[0]);
+        metadata.returnType(type);
       }
     }
 

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);
+    }
+}

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);
     }

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);
+    }
+}

reactive/src/test/java/feign/reactive/ReactiveFeignIntegrationTest.java

@@ -47,6 +47,7 @@ import java.lang.reflect.Type;
 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 class ReactiveFeignIntegrationTest {
   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 class ReactiveFeignIntegrationTest {
         .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 class ReactiveFeignIntegrationTest {
         .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 class ReactiveFeignIntegrationTest {
         .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 class ReactiveFeignIntegrationTest {
     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 class ReactiveFeignIntegrationTest {
     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 class ReactiveFeignIntegrationTest {
     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 class ReactiveFeignIntegrationTest {
     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 class ReactiveFeignIntegrationTest {
 
     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 class ReactiveFeignIntegrationTest {
 
     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 @@ public class ReactiveFeignIntegrationTest {
     Mono<String> version();
 
     @RequestLine("GET /users/{username}")
-    Flux<User> user(@Param("username") String username);
+    Mono<User> user(@Param("username") String username);
+
+    @RequestLine("GET /users")
+    Flux<User> usersFlux();
+
+    @RequestLine("GET /users")
+    Mono<List<User>> usersMono();
   }
 
 
@@ -314,6 +346,9 @@ public class ReactiveFeignIntegrationTest {
     @RequestLine("GET /users/{username}")
     Flowable<User> user(@Param("username") String username);
 
+    @RequestLine("GET /users")
+    Flowable<List<User>> users();
+
     @RequestLine("GET /users/search")
     Flowable<String> search(@QueryMap SearchQuery query);
   }

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));
+    }
+}

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<Contributor> 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<Contributor> 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<Contributor> contributorsFlux(@Param("owner") String owner, @Param("repo") String repo);
+
+    @RequestLine("GET /repos/{owner}/{repo}/contributors")
+    Mono<List<Contributor>> 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;
+    }
+  }
+}

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<Contributor> 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<List<Contributor>> 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;
+    }
+  }
+}