Streaming decoder support2

api/src/main/java/feign/InvocationContext.java

@@ -21,6 +21,7 @@
 import feign.codec.DecodeException;
 import feign.codec.Decoder;
 import feign.codec.ErrorDecoder;
+import java.io.Closeable;
 import java.io.IOException;
 import java.lang.reflect.Type;
 
@@ -71,6 +72,8 @@ public Object proceed() throws Exception {
       return disconnectResponseBodyIfNeeded(response);
     }
 
+    boolean noClose = false;
+
     try {
       final boolean shouldDecodeResponseBody =
           (response.status() >= 200 && response.status() < 300)
@@ -86,14 +89,19 @@ public Object proceed() throws Exception {
       }
 
       Class<?> rawType = Types.getRawType(returnType);
+
+      if (Closeable.class.isAssignableFrom(rawType)) {
+        noClose = true;
+      }
+
       if (TypedResponse.class.isAssignableFrom(rawType)) {
         Type bodyType = Types.resolveLastTypeParameter(returnType, TypedResponse.class);
         return TypedResponse.builder(response).body(decode(response, bodyType)).build();
       }
 
       return decode(response, returnType);
     } finally {
-      if (closeAfterDecode) {
+      if (closeAfterDecode && !noClose) {
         ensureClosed(response.body());
       }
     }

core/src/main/java/feign/core/codec/InputStreamAndReaderDecoder.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright © 2012 The Feign Authors (feign@commonhaus.dev)
+ *
+ * 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.core.codec;
+
+import static java.lang.String.format;
+
+import feign.FeignException;
+import feign.Response;
+import feign.codec.DecodeException;
+import feign.codec.PredicatedDecoder;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Reader;
+import java.lang.reflect.Type;
+
+public class InputStreamAndReaderDecoder implements PredicatedDecoder {
+
+  @Override
+  public Object decode(Response response, Type type)
+      throws IOException, DecodeException, FeignException {
+
+    if (InputStream.class.equals(type)) return response.body().asInputStream();
+
+    if (Reader.class.equals(type)) return response.body().asReader(response.charset());
+
+    throw new DecodeException(
+        response.status(),
+        format("%s is not a type supported by this decoder.", type),
+        response.request());
+  }
+
+  @Override
+  public boolean canDecode(Response response, Type type) {
+    if (InputStream.class.equals(type)) return true;
+    if (Reader.class.equals(type)) return true;
+
+    return false;
+  }
+}

core/src/test/java/feign/core/codec/InputStreamAndReaderDecoderTest.java

@@ -0,0 +1,139 @@
+/*
+ * Copyright © 2012 The Feign Authors (feign@commonhaus.dev)
+ *
+ * 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.core.codec;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import feign.Feign;
+import feign.RequestLine;
+import feign.Util;
+import java.io.InputStream;
+import java.io.Reader;
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+import mockwebserver3.MockResponse;
+import mockwebserver3.MockWebServer;
+import mockwebserver3.internal.BufferMockResponseBody;
+import okio.Buffer;
+import org.junit.jupiter.api.Test;
+
+public class InputStreamAndReaderDecoderTest {
+
+  interface LargeStreamTestInterface {
+
+    @RequestLine("GET /")
+    InputStream getLargeStream();
+
+    @RequestLine("GET /")
+    Reader getLargeReader();
+  }
+
+  @Test
+  void streamingResponse() throws Exception {
+
+    try (MockWebServer server = new MockWebServer()) {
+
+      server.start();
+
+      byte[] expectedResponse = new byte[16184];
+      new Random().nextBytes(expectedResponse);
+      server.enqueue(
+          new MockResponse.Builder()
+              .body(new BufferMockResponseBody(new Buffer().write(expectedResponse)))
+              .build());
+
+      LargeStreamTestInterface api =
+          Feign.builder()
+              .decoder(new InputStreamAndReaderDecoder())
+              .target(LargeStreamTestInterface.class, "http://localhost:" + server.getPort());
+
+      try (InputStream is = api.getLargeStream()) {
+        byte[] out = is.readAllBytes();
+        assertThat(out.length).isEqualTo(expectedResponse.length);
+        assertThat(out).isEqualTo(expectedResponse);
+      }
+    }
+  }
+
+  @Test
+  void streamingReaderResponse() throws Exception {
+    try (MockWebServer server = new MockWebServer()) {
+
+      server.start();
+
+      String expectedResponse =
+          new Random()
+              .ints(1, 1500 + 1)
+              .limit(16184)
+              .collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
+              .toString();
+
+      server.enqueue(
+          new MockResponse.Builder()
+              .body(
+                  new BufferMockResponseBody(
+                      new Buffer().write(expectedResponse.getBytes(StandardCharsets.UTF_16))))
+              .addHeader("content-type", "text/plan; charset=utf-16")
+              .build());
+
+      LargeStreamTestInterface api =
+          Feign.builder()
+              .decoder(new InputStreamAndReaderDecoder())
+              .target(LargeStreamTestInterface.class, "http://localhost:" + server.getPort());
+
+      try (Reader r = api.getLargeReader()) {
+        String out = Util.toString(r);
+        assertThat(out.length()).isEqualTo(expectedResponse.length());
+        assertThat(out).isEqualTo(expectedResponse);
+      }
+    }
+  }
+
+  @Test
+  void streamingReaderResponseWithNoCharset() throws Exception {
+
+    try (MockWebServer server = new MockWebServer()) {
+
+      server.start();
+
+      String expectedResponse =
+          new Random()
+              .ints(1, 1500 + 1)
+              .limit(16184)
+              .collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
+              .toString();
+
+      server.enqueue(
+          new MockResponse.Builder()
+              .body(
+                  new BufferMockResponseBody(
+                      new Buffer().write(expectedResponse.getBytes(Util.UTF_8))))
+              .addHeader("content-type", "text/plan")
+              .build());
+
+      LargeStreamTestInterface api =
+          Feign.builder()
+              .decoder(new InputStreamAndReaderDecoder())
+              .target(LargeStreamTestInterface.class, "http://localhost:" + server.getPort());
+
+      try (Reader r = api.getLargeReader()) {
+        String out = Util.toString(r);
+        assertThat(out.length()).isEqualTo(expectedResponse.length());
+        assertThat(out).isEqualTo(expectedResponse);
+      }
+    }
+  }
+}