Browse Source

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 <velo@users.noreply.github.com>
pull/1616/head
mroccyen 2 years ago committed by GitHub
parent
commit
7230f7c6ee
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 33
      README.md
  2. 38
      core/src/main/java/feign/stream/StreamDecoder.java
  3. 54
      core/src/test/java/feign/stream/StreamDecoderTest.java

33
README.md

@ -568,6 +568,39 @@ public class Example { @@ -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.

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

@ -21,6 +21,7 @@ import java.io.IOException; @@ -21,6 +21,7 @@ import java.io.IOException;
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 @@ import static feign.Util.ensureClosed; @@ -38,6 +39,11 @@ import static feign.Util.ensureClosed;
* .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 @@ import static feign.Util.ensureClosed; @@ -47,23 +53,27 @@ import static feign.Util.ensureClosed;
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 final class StreamDecoder implements Decoder { @@ -76,8 +86,20 @@ public final class StreamDecoder implements Decoder {
});
}
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 {

54
core/src/test/java/feign/stream/StreamDecoderTest.java

@ -14,12 +14,11 @@ @@ -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.Collections; @@ -28,9 +27,6 @@ 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;
@ -41,6 +37,9 @@ public class StreamDecoderTest { @@ -41,6 +37,9 @@ public class StreamDecoderTest {
@RequestLine("GET /")
Stream<String> get();
@RequestLine("GET /str")
String str();
@RequestLine("GET /cars")
Stream<Car> getCars();
@ -79,6 +78,41 @@ public class StreamDecoderTest { @@ -79,6 +78,41 @@ public class StreamDecoderTest {
}
}
@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 class StreamDecoderTest { @@ -90,10 +124,10 @@ public class StreamDecoderTest {
.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 {

Loading…
Cancel
Save