Skip to content

Commit 31c0b2c

Browse files
author
Adrian Cole
committed
Merge pull request OpenFeign#10 from Netflix/observer
Implement IncrementalCallback and IncrementalDecoder
2 parents 8780a0c + 0d7a69b commit 31c0b2c

17 files changed

Lines changed: 825 additions & 107 deletions

‎CHANGES.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
### Version 3.0
2+
* Added support for asynchronous callbacks via `IncrementalCallback<T>` and `IncrementalDecoder.TextStream<T>`.
23
* Wire is now Logger, with configurable Logger.Level.
34
* changed codec to be similar to [WebSocket JSR 356](http://docs.oracle.com/javaee/7/api/javax/websocket/package-summary.html)
45
* Decoder is now `Decoder.TextStream<T>`

‎README.md‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,59 @@ The generic parameter of `Decoder.TextStream<T>` designates which The type param
6868
return new SAXDecoder<ZoneList>(handlers){};
6969
}
7070
```
71+
### Asynchronous Incremental Callbacks
72+
If specified as the last argument of a method `IncrementalCallback<T>` fires a background task to add new elements to the callback as they are decoded. Think of `IncrementalCallback<T>` as an asynchronous equivalent to a lazy sequence.
73+
74+
Here's how one looks:
75+
```java
76+
IncrementalCallback<Contributor> printlnObserver = new IncrementalCallback<Contributor>() {
77+
78+
public int count;
79+
80+
@Override public void onNext(Contributor element) {
81+
count++;
82+
}
83+
84+
@Override public void onSuccess() {
85+
System.out.println("found " + count + " contributors");
86+
}
87+
88+
@Override public void onFailure(Throwable cause) {
89+
cause.printStackTrace();
90+
}
91+
};
92+
github.contributors("netflix", "feign", printlnObserver);
93+
```
94+
#### Incremental Decoding
95+
When using an `IncrementalCallback<T>`, you'll need to configure an `IncrementalDecoderi.TextStream<T>` or a general one for all types (`IncrementalDecoder.TextStream<Object>`).
96+
97+
Here's how to wire in a reflective incremental json decoder:
98+
```java
99+
@Provides(type = SET) IncrementalDecoder incrementalDecoder(final Gson gson) {
100+
return new IncrementalDecoder.TextStream<Object>() {
101+
102+
@Override
103+
public void decode(Reader reader, Type type, IncrementalCallback<? super Object> incrementalCallback) throws IOException {
104+
JsonReader jsonReader = new JsonReader(reader);
105+
jsonReader.beginArray();
106+
while (jsonReader.hasNext()) {
107+
try {
108+
incrementalCallback.onNext(gson.fromJson(jsonReader, type));
109+
} catch (JsonIOException e) {
110+
if (e.getCause() != null && e.getCause() instanceof IOException) {
111+
throw IOException.class.cast(e.getCause());
112+
}
113+
throw e;
114+
}
115+
}
116+
jsonReader.endArray();
117+
}
118+
};
119+
}
120+
```
121+
122+
123+
71124
### Multiple Interfaces
72125
Feign can produce multiple api interfaces. These are defined as `Target<T>` (default `HardCodedTarget<T>`), which allow for dynamic discovery and decoration of requests prior to execution.
73126

‎feign-core/src/main/java/feign/Contract.java‎

Lines changed: 14 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -15,17 +15,18 @@
1515
*/
1616
package feign;
1717

18+
import javax.inject.Named;
1819
import java.lang.annotation.Annotation;
1920
import java.lang.reflect.Method;
21+
import java.lang.reflect.Type;
2022
import java.net.URI;
2123
import java.util.ArrayList;
2224
import java.util.Collection;
2325
import java.util.List;
2426

25-
import javax.inject.Named;
26-
2727
import static feign.Util.checkState;
2828
import static feign.Util.emptyToNull;
29+
import static feign.Util.resolveLastTypeParameter;
2930

3031
/**
3132
* Defines what annotations and values are valid on interfaces.
@@ -50,7 +51,7 @@ public List<MethodMetadata> parseAndValidatateMetadata(Class<?> declaring) {
5051
*/
5152
public MethodMetadata parseAndValidatateMetadata(Method method) {
5253
MethodMetadata data = new MethodMetadata();
53-
data.returnType(method.getGenericReturnType());
54+
data.decodeInto(method.getGenericReturnType());
5455
data.configKey(Feign.configKey(method));
5556

5657
for (Annotation methodAnnotation : method.getAnnotations()) {
@@ -69,8 +70,17 @@ public MethodMetadata parseAndValidatateMetadata(Method method) {
6970
}
7071
if (parameterTypes[i] == URI.class) {
7172
data.urlIndex(i);
73+
} else if (IncrementalCallback.class.isAssignableFrom(parameterTypes[i])) {
74+
checkState(method.getReturnType() == void.class, "IncrementalCallback methods must return void: %s", method);
75+
checkState(i == count - 1, "IncrementalCallback must be the last parameter: %s", method);
76+
Type context = method.getGenericParameterTypes()[i];
77+
Type incrementalCallbackType = resolveLastTypeParameter(context, IncrementalCallback.class);
78+
data.decodeInto(incrementalCallbackType);
79+
data.incrementalCallbackIndex(i);
80+
checkState(incrementalCallbackType != null, "Expected param %s to be IncrementalCallback<X> or IncrementalCallback<? super X> or a subtype",
81+
context, incrementalCallbackType);
7282
} else if (!isHttpAnnotation) {
73-
checkState(data.formParams().isEmpty(), "Body parameters cannot be used with @FormParam parameters.");
83+
checkState(data.formParams().isEmpty(), "Body parameters cannot be used with form parameters.");
7484
checkState(data.bodyIndex() == null, "Method has too many Body parameters: %s", method);
7585
data.bodyIndex(i);
7686
data.bodyType(method.getGenericParameterTypes()[i]);

‎feign-core/src/main/java/feign/Feign.java‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@
1515
*/
1616
package feign;
1717

18+
19+
import dagger.Lazy;
1820
import dagger.ObjectGraph;
1921
import dagger.Provides;
2022
import feign.Logger.NoOpLogger;
@@ -23,13 +25,23 @@
2325
import feign.codec.Decoder;
2426
import feign.codec.Encoder;
2527
import feign.codec.ErrorDecoder;
28+
import feign.codec.IncrementalDecoder;
2629

30+
import javax.inject.Named;
31+
import javax.inject.Singleton;
2732
import javax.net.ssl.SSLSocketFactory;
33+
import java.io.Closeable;
2834
import java.lang.reflect.Method;
2935
import java.util.ArrayList;
3036
import java.util.Collections;
3137
import java.util.List;
3238
import java.util.Set;
39+
import java.util.concurrent.Executor;
40+
import java.util.concurrent.ExecutorService;
41+
import java.util.concurrent.Executors;
42+
import java.util.concurrent.ThreadFactory;
43+
44+
import static java.lang.Thread.MIN_PRIORITY;
3345

3446
/**
3547
* Feign's purpose is to ease development against http apis that feign
@@ -38,7 +50,7 @@
3850
* In implementation, Feign is a {@link Feign#newInstance factory} for
3951
* generating {@link Target targeted} http apis.
4052
*/
41-
public abstract class Feign {
53+
public abstract class Feign implements Closeable {
4254

4355
/**
4456
* Returns a new instance of an HTTP API, defined by annotations in the
@@ -119,6 +131,26 @@ public static class Defaults {
119131
@Provides Set<Decoder> noDecoders() {
120132
return Collections.emptySet();
121133
}
134+
135+
@Provides Set<IncrementalDecoder> noIncrementalDecoders() {
136+
return Collections.emptySet();
137+
}
138+
139+
/**
140+
* Used for both http invocation and decoding when incrementalCallbacks are used.
141+
*/
142+
@Provides @Singleton @Named("http") Executor httpExecutor() {
143+
return Executors.newCachedThreadPool(new ThreadFactory() {
144+
@Override public Thread newThread(final Runnable r) {
145+
return new Thread(new Runnable() {
146+
@Override public void run() {
147+
Thread.currentThread().setPriority(MIN_PRIORITY);
148+
r.run();
149+
}
150+
}, MethodHandler.IDLE_THREAD_NAME);
151+
}
152+
});
153+
}
122154
}
123155

124156
/**
@@ -162,7 +194,16 @@ private static List<Object> modulesForGraph(Object... modules) {
162194
return modulesForGraph;
163195
}
164196

165-
Feign() {
197+
private final Lazy<Executor> httpExecutor;
166198

199+
Feign(Lazy<Executor> httpExecutor) {
200+
this.httpExecutor = httpExecutor;
201+
}
202+
203+
@Override public void close() {
204+
Executor e = httpExecutor.get();
205+
if (e instanceof ExecutorService) {
206+
ExecutorService.class.cast(e).shutdownNow();
207+
}
167208
}
168209
}

‎feign-core/src/main/java/feign/IncrementalCallback.java‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
package feign;
2+
3+
/**
4+
* Communicates results as they are {@link feign.codec.Decoder decoded} from
5+
* an {@link Response.Body http response body}. {@link #onNext(Object) onNext}
6+
* will be called for each incremental value of type {@code T}, or not at all
7+
* when there are no values present in the response. Methods that accept
8+
* {@code IncrementalCallback} are asynchronous, which implies background
9+
* processing.
10+
* <br>
11+
* {@link #onSuccess() onSuccess} or {@link #onFailure(Throwable)} onFailure}
12+
* will be called when the response is finished, but not both.
13+
* <br>
14+
* {@code IncrementalCallback} can be used as an asynchronous alternative to a
15+
* {@code Collection}, or any other use where iterative response parsing is
16+
* worth the additional effort to implement this interface.
17+
* <br>
18+
* <br>
19+
* Here's an example of implementing {@code IncrementalCallback}:
20+
* <br>
21+
* <pre>
22+
* IncrementalCallback<Contributor> counter = new IncrementalCallback<Contributor>() {
23+
*
24+
* public int count;
25+
*
26+
* &#064;Override public void onNext(Contributor element) {
27+
* count++;
28+
* }
29+
*
30+
* &#064;Override public void onSuccess() {
31+
* System.out.println("found " + count + " contributors");
32+
* }
33+
*
34+
* &#064;Override public void onFailure(Throwable cause) {
35+
* System.err.println("sad face after contributor " + count);
36+
* }
37+
* };
38+
* github.contributors("netflix", "feign", counter);
39+
* </pre>
40+
*
41+
* @param <T> expected value to decode
42+
*/
43+
public interface IncrementalCallback<T> {
44+
/**
45+
* Invoked as soon as new data is available. Could be invoked many times or
46+
* not at all.
47+
*
48+
* @param element next decoded element.
49+
*/
50+
void onNext(T element);
51+
52+
/**
53+
* Called when response processing completed successfully.
54+
*/
55+
void onSuccess();
56+
57+
/**
58+
* Called when response processing failed for any reason.
59+
* <br>
60+
* Common failure cases include {@link FeignException},
61+
* {@link java.io.IOException}, and {@link feign.codec.DecodeException}.
62+
* However, the cause could be a {@code Throwable} of any kind.
63+
*
64+
* @param cause the reason for the failure
65+
*/
66+
void onFailure(Throwable cause);
67+
}

0 commit comments

Comments
 (0)