Skip to content

Commit 8ff681f

Browse files
trumpetinctrumpetinc
committed
Add support for streaming decoding for InputStream and Reader return
types
1 parent b216d52 commit 8ff681f

5 files changed

Lines changed: 301 additions & 19 deletions

File tree

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

Lines changed: 28 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import feign.codec.DecodeException;
2222
import feign.codec.Decoder;
2323
import feign.codec.ErrorDecoder;
24+
25+
import java.io.Closeable;
2426
import java.io.IOException;
2527
import java.lang.reflect.Type;
2628

@@ -68,9 +70,11 @@ public Response response() {
6870

6971
public Object proceed() throws Exception {
7072
if (returnType == Response.class) {
71-
return disconnectResponseBodyIfNeeded(response);
73+
return response;
7274
}
7375

76+
boolean noClose = false;
77+
7478
try {
7579
final boolean shouldDecodeResponseBody =
7680
(response.status() >= 200 && response.status() < 300)
@@ -86,35 +90,40 @@ public Object proceed() throws Exception {
8690
}
8791

8892
Class<?> rawType = Types.getRawType(returnType);
93+
94+
if(Closeable.class.isAssignableFrom(rawType)) {
95+
noClose = true;
96+
}
97+
8998
if (TypedResponse.class.isAssignableFrom(rawType)) {
9099
Type bodyType = Types.resolveLastTypeParameter(returnType, TypedResponse.class);
91100
return TypedResponse.builder(response).body(decode(response, bodyType)).build();
92101
}
93102

94103
return decode(response, returnType);
95104
} finally {
96-
if (closeAfterDecode) {
105+
if (closeAfterDecode && !noClose) {
97106
ensureClosed(response.body());
98107
}
99108
}
100109
}
101-
102-
private static Response disconnectResponseBodyIfNeeded(Response response) throws IOException {
103-
final boolean shouldDisconnectResponseBody =
104-
response.body() != null
105-
&& response.body().length() != null
106-
&& response.body().length() <= MAX_RESPONSE_BUFFER_SIZE;
107-
if (!shouldDisconnectResponseBody) {
108-
return response;
109-
}
110-
111-
try {
112-
final byte[] bodyData = Util.toByteArray(response.body().asInputStream());
113-
return response.toBuilder().body(bodyData).build();
114-
} finally {
115-
ensureClosed(response.body());
116-
}
117-
}
110+
//
111+
// private static Response disconnectResponseBodyIfNeeded(Response response) throws IOException {
112+
// final boolean shouldDisconnectResponseBody =
113+
// response.body() != null
114+
// && response.body().length() != null
115+
// && response.body().length() <= MAX_RESPONSE_BUFFER_SIZE;
116+
// if (!shouldDisconnectResponseBody) {
117+
// return response;
118+
// }
119+
//
120+
// try {
121+
// final byte[] bodyData = Util.toByteArray(response.body().asInputStream());
122+
// return response.toBuilder().body(bodyData).build();
123+
// } finally {
124+
// ensureClosed(response.body());
125+
// }
126+
// }
118127

119128
private Object decode(Response response, Type returnType) {
120129
try {

‎api/src/main/java/feign/Util.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,9 @@ public class Util {
5959
/** The HTTP Content-Length header field name. */
6060
public static final String CONTENT_LENGTH = "Content-Length";
6161

62+
/** The HTTP Content-Type header field name. */
63+
public static final String CONTENT_TYPE = "Content-Type";
64+
6265
/** The HTTP Content-Encoding header field name. */
6366
public static final String CONTENT_ENCODING = "Content-Encoding";
6467

‎api/src/main/java/feign/utils/ContentTypeParser.java‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
/*
2+
* ====================================================================
3+
* Licensed to the Apache Software Foundation (ASF) under one
4+
* or more contributor license agreements. See the NOTICE file
5+
* distributed with this work for additional information
6+
* regarding copyright ownership. The ASF licenses this file
7+
* to you under the Apache License, Version 2.0 (the
8+
* "License"); you may not use this file except in compliance
9+
* with the License. You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing,
14+
* software distributed under the License is distributed on an
15+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
* KIND, either express or implied. See the License for the
17+
* specific language governing permissions and limitations
18+
* under the License.
19+
* ====================================================================
20+
*
21+
* This software consists of voluntary contributions made by many
22+
* individuals on behalf of the Apache Software Foundation. For more
23+
* information on the Apache Software Foundation, please see
24+
* <http://www.apache.org/>.
25+
*
26+
*/
27+
package feign.utils;
28+
29+
import java.nio.charset.Charset;
30+
import java.util.Collection;
31+
import java.util.Collections;
32+
import java.util.Map;
33+
import java.util.Optional;
34+
35+
import feign.Util;
36+
37+
public final class ContentTypeParser {
38+
39+
private ContentTypeParser() {
40+
}
41+
42+
public static ContentTypeResult parseContentTypeFromHeaders(Map<String, Collection<String>> headers, String ifMissing) {
43+
// The header map *should* be a case insensitive treemap
44+
for (String val : headers.getOrDefault(Util.CONTENT_TYPE, Collections.emptyList())) {
45+
return parseContentTypeHeader(val);
46+
}
47+
48+
return new ContentTypeResult(ifMissing, null);
49+
}
50+
51+
public static ContentTypeResult parseContentTypeHeader(String contentTypeHeader) {
52+
53+
String[] contentTypeParmeters = contentTypeHeader.split(";");
54+
String contentType = contentTypeParmeters[0];
55+
String charsetString = "";
56+
if (contentTypeParmeters.length > 1) {
57+
String[] charsetParts = contentTypeParmeters[1].split("=");
58+
if (charsetParts.length == 2 && "charset".equalsIgnoreCase(charsetParts[0].trim())) {
59+
// TODO: 20260727 - this doesn't implement the full parser definition for the content-type header (esp related to quoted strings, etc...) - see https://www.w3.org/Protocols/rfc1341/4_Content-Type.html
60+
charsetString = charsetParts[1].trim();
61+
if (charsetString.length() > 1 && charsetString.startsWith("\"") && charsetString.endsWith("\""))
62+
charsetString = charsetString.substring(1, charsetString.length()-1);
63+
}
64+
}
65+
66+
return new ContentTypeResult(contentType, Charset.forName(charsetString, null));
67+
}
68+
69+
public static class ContentTypeResult{
70+
public static final ContentTypeResult MISSING = new ContentTypeResult("", null);
71+
72+
private String contentType;
73+
private Optional<Charset> charset;
74+
75+
public ContentTypeResult(String contentType, Charset charset) {
76+
this.contentType = contentType;
77+
this.charset = Optional.ofNullable(charset);
78+
}
79+
80+
public String getContentType() {
81+
return contentType;
82+
}
83+
84+
public Optional<Charset> getCharset() {
85+
return charset;
86+
}
87+
}
88+
}

‎core/src/test/java/feign/core/codec/InputStreamAndReaderDecoder.java‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
package feign.core.codec;
2+
3+
import feign.FeignException;
4+
import feign.Response;
5+
import feign.Util;
6+
import feign.codec.DecodeException;
7+
import feign.codec.Decoder;
8+
import feign.utils.ContentTypeParser;
9+
import java.io.IOException;
10+
import java.io.InputStream;
11+
import java.io.Reader;
12+
import java.lang.reflect.Type;
13+
14+
public class InputStreamAndReaderDecoder implements Decoder {
15+
private final Decoder delegateDecoder;
16+
17+
public InputStreamAndReaderDecoder(Decoder delegate) {
18+
this.delegateDecoder = delegate;
19+
}
20+
21+
@Override
22+
public Object decode(Response response, Type type)
23+
throws IOException, DecodeException, FeignException {
24+
25+
if (InputStream.class.equals(type)) return response.body().asInputStream();
26+
27+
if (Reader.class.equals(type))
28+
return response
29+
.body()
30+
.asReader(
31+
ContentTypeParser.parseContentTypeFromHeaders(response.headers(), "UTF-8")
32+
.getCharset()
33+
.orElse(Util.UTF_8));
34+
35+
if (delegateDecoder == null) return null;
36+
37+
return delegateDecoder.decode(response, type);
38+
}
39+
}

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

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
/*
2+
* ====================================================================
3+
* Licensed to the Apache Software Foundation (ASF) under one
4+
* or more contributor license agreements. See the NOTICE file
5+
* distributed with this work for additional information
6+
* regarding copyright ownership. The ASF licenses this file
7+
* to you under the Apache License, Version 2.0 (the
8+
* "License"); you may not use this file except in compliance
9+
* with the License. You may obtain a copy of the License at
10+
*
11+
* http://www.apache.org/licenses/LICENSE-2.0
12+
*
13+
* Unless required by applicable law or agreed to in writing,
14+
* software distributed under the License is distributed on an
15+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
16+
* KIND, either express or implied. See the License for the
17+
* specific language governing permissions and limitations
18+
* under the License.
19+
* ====================================================================
20+
*
21+
* This software consists of voluntary contributions made by many
22+
* individuals on behalf of the Apache Software Foundation. For more
23+
* information on the Apache Software Foundation, please see
24+
* <http://www.apache.org/>.
25+
*
26+
*/
27+
package feign.core.codec;
28+
29+
import static org.assertj.core.api.Assertions.assertThat;
30+
31+
import feign.Feign;
32+
import feign.RequestLine;
33+
import feign.Util;
34+
import java.io.InputStream;
35+
import java.io.Reader;
36+
import java.nio.charset.StandardCharsets;
37+
import java.util.Random;
38+
import mockwebserver3.MockResponse;
39+
import mockwebserver3.MockWebServer;
40+
import mockwebserver3.internal.BufferMockResponseBody;
41+
import okio.Buffer;
42+
import org.junit.jupiter.api.Test;
43+
44+
public class InputStreamAndReaderDecoderTest {
45+
public final MockWebServer server = new MockWebServer();
46+
47+
interface LargeStreamTestInterface {
48+
49+
@RequestLine("GET /")
50+
InputStream getLargeStream();
51+
52+
@RequestLine("GET /")
53+
Reader getLargeReader();
54+
}
55+
56+
@Test
57+
void streamingResponse() throws Exception {
58+
59+
server.start();
60+
61+
byte[] expectedResponse = new byte[16184];
62+
new Random().nextBytes(expectedResponse);
63+
server.enqueue(
64+
new MockResponse.Builder()
65+
.body(new BufferMockResponseBody(new Buffer().write(expectedResponse)))
66+
.build());
67+
68+
LargeStreamTestInterface api =
69+
Feign.builder()
70+
.decoder(new InputStreamAndReaderDecoder(null))
71+
.target(LargeStreamTestInterface.class, "http://localhost:" + server.getPort());
72+
73+
try (InputStream is = api.getLargeStream()) {
74+
byte[] out = is.readAllBytes();
75+
assertThat(out.length).isEqualTo(expectedResponse.length);
76+
assertThat(out).isEqualTo(expectedResponse);
77+
}
78+
}
79+
80+
@Test
81+
void streamingReaderResponse() throws Exception {
82+
83+
server.start();
84+
85+
String expectedResponse =
86+
new Random()
87+
.ints(1, 1500 + 1)
88+
.limit(16184)
89+
.collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
90+
.toString();
91+
92+
server.enqueue(
93+
new MockResponse.Builder()
94+
.body(
95+
new BufferMockResponseBody(
96+
new Buffer().write(expectedResponse.getBytes(StandardCharsets.UTF_16))))
97+
.addHeader("content-type", "text/plan; charset=utf-16")
98+
.build());
99+
100+
LargeStreamTestInterface api =
101+
Feign.builder()
102+
.decoder(new InputStreamAndReaderDecoder(null))
103+
.target(LargeStreamTestInterface.class, "http://localhost:" + server.getPort());
104+
105+
try (Reader r = api.getLargeReader()) {
106+
String out = Util.toString(r);
107+
assertThat(out.length()).isEqualTo(expectedResponse.length());
108+
assertThat(out).isEqualTo(expectedResponse);
109+
}
110+
}
111+
112+
@Test
113+
void streamingReaderResponseWithNoCharset() throws Exception {
114+
115+
server.start();
116+
117+
String expectedResponse =
118+
new Random()
119+
.ints(1, 1500 + 1)
120+
.limit(16184)
121+
.collect(StringBuilder::new, StringBuilder::appendCodePoint, StringBuilder::append)
122+
.toString();
123+
124+
server.enqueue(
125+
new MockResponse.Builder()
126+
.body(
127+
new BufferMockResponseBody(
128+
new Buffer().write(expectedResponse.getBytes(Util.UTF_8))))
129+
.addHeader("content-type", "text/plan")
130+
.build());
131+
132+
LargeStreamTestInterface api =
133+
Feign.builder()
134+
.decoder(new InputStreamAndReaderDecoder(null))
135+
.target(LargeStreamTestInterface.class, "http://localhost:" + server.getPort());
136+
137+
try (Reader r = api.getLargeReader()) {
138+
String out = Util.toString(r);
139+
assertThat(out.length()).isEqualTo(expectedResponse.length());
140+
assertThat(out).isEqualTo(expectedResponse);
141+
}
142+
}
143+
}

0 commit comments

Comments
 (0)