Stream decoder optimize (#1590)

* Optimize StreamDecoder * Optimize StreamDecoder * add license header * add license header * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * add license header * add license header * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * add license header * add license header * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * add license header * Optimize StreamDecoder * Optimize StreamDecoder * Optimize StreamDecoder * add some example * Optimize StreamDecoder * add a section of README for stream decoder * Update StreamDecoder.java Co-authored-by: Marvin Froeder <[email protected]>
OpenFeign · Apr 29, 2022 · 7230f7c · 7230f7c
1 parent e1a7028
commit 7230f7c
Show file tree

Hide file tree

Showing 3 changed files with 107 additions and 18 deletions.
diff --git a/README.md b/README.md
@@ -568,6 +568,39 @@ public class Example {
 }
 ```
 
+If any methods in your interface return type `Stream`, you'll need to configure a `StreamDecoder`.
+
+Here's how to configure Stream decoder without delegate decoder:
+
+```java
+public class Example {
+  public static void main(String[] args) {
+    GitHub github = Feign.builder()
+            .decoder(StreamDecoder.create((r, t) -> {
+              BufferedReader bufferedReader = new BufferedReader(r.body().asReader(UTF_8));
+              return bufferedReader.lines().iterator();
+            }))
+            .target(GitHub.class, "https://api.github.com");
+  }
+}
+``` 
+
+Here's how to configure Stream decoder with delegate decoder:
+
+```java
+
+public class Example {
+  public static void main(String[] args) {
+    GitHub github = Feign.builder()
+            .decoder(StreamDecoder.create((r, t) -> {
+              BufferedReader bufferedReader = new BufferedReader(r.body().asReader(UTF_8));
+              return bufferedReader.lines().iterator();
+            }, (r, t) -> "this is delegate decoder"))
+            .target(GitHub.class, "https://api.github.com");
+  }
+}
+```
+
 ### Encoders
 The simplest way to send a request body to a server is to define a `POST` method that has a `String` or `byte[]` parameter without any annotations on it. You will likely need to add a `Content-Type` header.
 

diff --git a/core/src/main/java/feign/stream/StreamDecoder.java b/core/src/main/java/feign/stream/StreamDecoder.java
@@ -21,6 +21,7 @@
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
 import java.util.Iterator;
+import java.util.Optional;
 import java.util.Spliterators;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
@@ -38,6 +39,11 @@
  *   .decoder(StreamDecoder.create(JacksonIteratorDecoder.create()))
  *   .doNotCloseAfterDecode() // Required for streaming
  *   .target(GitHub.class, "https://api.github.com");
+ * or
+ * Feign.builder()
+ *   .decoder(StreamDecoder.create(JacksonIteratorDecoder.create(), (r, t) -> "hello world")))
+ *   .doNotCloseAfterDecode() // Required for streaming
+ *   .target(GitHub.class, "https://api.github.com");
  * interface GitHub {
  *  {@literal @}RequestLine("GET /repos/{owner}/{repo}/contributors")
  *   Stream<Contributor> contributors(@Param("owner") String owner, @Param("repo") String repo);
@@ -47,23 +53,27 @@
 public final class StreamDecoder implements Decoder {
 
   private final Decoder iteratorDecoder;
+  private final Optional<Decoder> delegateDecoder;
 
-  StreamDecoder(Decoder iteratorDecoder) {
+  StreamDecoder(Decoder iteratorDecoder, Decoder delegateDecoder) {
     this.iteratorDecoder = iteratorDecoder;
+    this.delegateDecoder = Optional.ofNullable(delegateDecoder);
   }
 
   @Override
   public Object decode(Response response, Type type)
       throws IOException, FeignException {
-    if (!(type instanceof ParameterizedType)) {
-      throw new IllegalArgumentException("StreamDecoder supports only stream: unknown " + type);
+    if (!isStream(type)) {
+      if (!delegateDecoder.isPresent()) {
+        throw new IllegalArgumentException("StreamDecoder supports types other than stream. " +
+            "When type is not stream, the delegate decoder needs to be setting.");
+      } else {
+        return delegateDecoder.get().decode(response, type);
+      }
     }
     ParameterizedType streamType = (ParameterizedType) type;
-    if (!Stream.class.equals(streamType.getRawType())) {
-      throw new IllegalArgumentException("StreamDecoder supports only stream: unknown " + type);
-    }
     Iterator<?> iterator =
-        (Iterator) iteratorDecoder.decode(response, new IteratorParameterizedType(streamType));
+        (Iterator<?>) iteratorDecoder.decode(response, new IteratorParameterizedType(streamType));
 
     return StreamSupport.stream(
         Spliterators.spliteratorUnknownSize(iterator, 0), false)
@@ -76,8 +86,20 @@ public Object decode(Response response, Type type)
         });
   }
 
+  public static boolean isStream(Type type) {
+    if (!(type instanceof ParameterizedType)) {
+      return false;
+    }
+    ParameterizedType parameterizedType = (ParameterizedType) type;
+    return parameterizedType.getRawType().equals(Stream.class);
+  }
+
   public static StreamDecoder create(Decoder iteratorDecoder) {
-    return new StreamDecoder(iteratorDecoder);
+    return new StreamDecoder(iteratorDecoder, null);
+  }
+
+  public static StreamDecoder create(Decoder iteratorDecoder, Decoder delegateDecoder) {
+    return new StreamDecoder(iteratorDecoder, delegateDecoder);
   }
 
   static final class IteratorParameterizedType implements ParameterizedType {

diff --git a/core/src/test/java/feign/stream/StreamDecoderTest.java b/core/src/test/java/feign/stream/StreamDecoderTest.java
@@ -14,12 +14,11 @@
 package feign.stream;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import feign.Feign;
-import feign.Request;
+import feign.*;
 import feign.Request.HttpMethod;
-import feign.RequestLine;
-import feign.Response;
-import feign.Util;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.Test;
 import java.io.BufferedReader;
 import java.io.Closeable;
 import java.io.IOException;
@@ -28,9 +27,6 @@
 import java.util.Iterator;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
-import okhttp3.mockwebserver.MockResponse;
-import okhttp3.mockwebserver.MockWebServer;
-import org.junit.Test;
 import static feign.Util.UTF_8;
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -41,6 +37,9 @@ interface StreamInterface {
     @RequestLine("GET /")
     Stream<String> get();
 
+    @RequestLine("GET /str")
+    String str();
+
     @RequestLine("GET /cars")
     Stream<Car> getCars();
 
@@ -79,6 +78,41 @@ public void simpleStreamTest() {
     }
   }
 
+  @Test
+  public void simpleDefaultStreamTest() {
+    MockWebServer server = new MockWebServer();
+    server.enqueue(new MockResponse().setBody("foo\nbar"));
+
+    StreamInterface api = Feign.builder()
+        .decoder(StreamDecoder.create((r, t) -> {
+          BufferedReader bufferedReader = new BufferedReader(r.body().asReader(UTF_8));
+          return bufferedReader.lines().iterator();
+        }))
+        .doNotCloseAfterDecode()
+        .target(StreamInterface.class, server.url("/").toString());
+
+    try (Stream<String> stream = api.get()) {
+      assertThat(stream.collect(Collectors.toList())).isEqualTo(Arrays.asList("foo", "bar"));
+    }
+  }
+
+  @Test
+  public void simpleDeleteDecoderTest() {
+    MockWebServer server = new MockWebServer();
+    server.enqueue(new MockResponse().setBody("foo\nbar"));
+
+    StreamInterface api = Feign.builder()
+        .decoder(StreamDecoder.create((r, t) -> {
+          BufferedReader bufferedReader = new BufferedReader(r.body().asReader(UTF_8));
+          return bufferedReader.lines().iterator();
+        }, (r, t) -> "str"))
+        .doNotCloseAfterDecode()
+        .target(StreamInterface.class, server.url("/").toString());
+
+    String str = api.str();
+    assertThat(str).isEqualTo("str");
+  }
+
   @Test
   public void shouldCloseIteratorWhenStreamClosed() throws IOException {
     Response response = Response.builder()
@@ -90,10 +124,10 @@ public void shouldCloseIteratorWhenStreamClosed() throws IOException {
         .build();
 
     TestCloseableIterator it = new TestCloseableIterator();
-    StreamDecoder decoder = new StreamDecoder((r, t) -> it);
+    StreamDecoder decoder = StreamDecoder.create((r, t) -> it);
 
     try (Stream<?> stream =
-        (Stream) decoder.decode(response, new TypeReference<Stream<String>>() {}.getType())) {
+        (Stream<?>) decoder.decode(response, new TypeReference<Stream<String>>() {}.getType())) {
       assertThat(stream.collect(Collectors.toList())).hasSize(1);
       assertThat(it.called).isTrue();
     } finally {