Stream decoder optimize

core/src/main/java/feign/stream/StreamDecoder.java

@@ -16,6 +16,7 @@
 import feign.FeignException;
 import feign.Response;
 import feign.codec.Decoder;
+import java.io.BufferedReader;
 import java.io.Closeable;
 import java.io.IOException;
 import java.lang.reflect.ParameterizedType;
@@ -24,6 +25,7 @@
 import java.util.Spliterators;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
+import static feign.Util.UTF_8;
 import static feign.Util.ensureClosed;
 
 /**
@@ -43,27 +45,33 @@
  *   Stream<Contributor> contributors(@Param("owner") String owner, @Param("repo") String repo);
  * }</code>
  * </pre>
+ * 
+ * @author mroccyen
  */
 public final class StreamDecoder implements Decoder {
 
   private final Decoder iteratorDecoder;
+  private final Decoder delegateDecoder;
 
-  StreamDecoder(Decoder iteratorDecoder) {
+  StreamDecoder(Decoder iteratorDecoder, Decoder delegateDecoder) {
     this.iteratorDecoder = iteratorDecoder;
+    this.delegateDecoder = 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 == null) {
+        throw new IllegalArgumentException("StreamDecoder supports types other than stream. " +
+            "When type is not stream, the delegate decoder needs to be setting.");
+      } else {
+        return delegateDecoder.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 +84,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 {
@@ -103,4 +123,17 @@ public Type getOwnerType() {
       return null;
     }
   }
+
+  /**
+   * Default implementation of Decodes an http response into an Iterator
+   */
+  public static class LineToIteratorDecoder implements Decoder {
+
+    @Override
+    public Object decode(Response response, Type type) throws IOException {
+      BufferedReader bufferedReader = new BufferedReader(response.body().asReader(UTF_8));
+      return bufferedReader.lines().iterator();
+    }
+
+  }
 }

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,36 @@ 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(new StreamDecoder.LineToIteratorDecoder()))
+        .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(new StreamDecoder.LineToIteratorDecoder(), (r, t) -> "str"))
+        // .decoder(StreamDecoder.create(new StreamDecoder.LineToIteratorDecoder()))
+        .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 +119,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 {