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,36 @@
  *   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;
 
-  StreamDecoder(Decoder iteratorDecoder) {
+  public StreamDecoder(Decoder iteratorDecoder) {
     this.iteratorDecoder = iteratorDecoder;
   }
 
   @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)) {
+      return iteratorDecoder.decode(response, type);
     }
     ParameterizedType streamType = (ParameterizedType) type;
-    if (!Stream.class.equals(streamType.getRawType())) {
-      throw new IllegalArgumentException("StreamDecoder supports only stream: unknown " + type);
+    Object result = iteratorDecoder
+        .decode(response, new IteratorParameterizedType(streamType));
+
+    Iterator<?> iterator;
+    if (result instanceof Iterator<?>) {
+      iterator = (Iterator<?>) result;
+    } else {
+      // use default iterator decoder handle
+      Object defaultResult = new LineToIteratorDecoder()
+          .decode(response, new IteratorParameterizedType(streamType));
+      iterator = (Iterator<?>) defaultResult;
     }
-    Iterator<?> iterator =
-        (Iterator) iteratorDecoder.decode(response, new IteratorParameterizedType(streamType));
 
     return StreamSupport.stream(
         Spliterators.spliteratorUnknownSize(iterator, 0), false)
@@ -76,6 +87,14 @@ 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);
   }
@@ -103,4 +122,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,23 +14,22 @@
 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 feign.codec.DecodeException;
+import feign.codec.Decoder;
+import okhttp3.mockwebserver.MockResponse;
+import okhttp3.mockwebserver.MockWebServer;
+import org.junit.Test;
 import java.io.BufferedReader;
 import java.io.Closeable;
 import java.io.IOException;
+import java.lang.reflect.Type;
 import java.util.Arrays;
 import java.util.Collections;
 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;
 
@@ -79,6 +78,21 @@ 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) -> new Object()))
+        .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 shouldCloseIteratorWhenStreamClosed() throws IOException {
     Response response = Response.builder()
@@ -93,7 +107,7 @@ public void shouldCloseIteratorWhenStreamClosed() throws IOException {
     StreamDecoder decoder = new StreamDecoder((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 {