Skip to content

Commit 309d4b3

Browse files
author
adriancole
committed
integrated IncrementalCallback into methodhandler and added IncrementalDecoder
1 parent e88f2e5 commit 309d4b3

8 files changed

Lines changed: 481 additions & 40 deletions

File tree

‎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/MethodHandler.java‎

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

18+
import dagger.Lazy;
1819
import feign.Request.Options;
1920
import feign.codec.DecodeException;
2021
import feign.codec.Decoder;
2122
import feign.codec.ErrorDecoder;
23+
import feign.codec.IncrementalDecoder;
2224

2325
import javax.inject.Inject;
26+
import javax.inject.Named;
2427
import javax.inject.Provider;
2528
import java.io.IOException;
29+
import java.io.Reader;
30+
import java.util.concurrent.Executor;
2631
import java.util.concurrent.TimeUnit;
2732

2833
import static feign.FeignException.errorExecuting;
@@ -32,6 +37,12 @@
3237

3338
abstract class MethodHandler {
3439

40+
/**
41+
* same approach as retrofit: temporarily rename threads
42+
*/
43+
static final String THREAD_PREFIX = "Feign-";
44+
static final String IDLE_THREAD_NAME = THREAD_PREFIX + "Idle";
45+
3546
/**
3647
* Those using guava will implement as {@code Function<Object[], RequestTemplate>}.
3748
*/
@@ -42,12 +53,15 @@ static interface BuildTemplateFromArgs {
4253
static class Factory {
4354

4455
private final Client client;
56+
private final Lazy<Executor> httpExecutor;
4557
private final Provider<Retryer> retryer;
4658
private final Logger logger;
4759
private final Logger.Level logLevel;
4860

49-
@Inject Factory(Client client, Provider<Retryer> retryer, Logger logger, Logger.Level logLevel) {
61+
@Inject Factory(Client client, @Named("http") Lazy<Executor> httpExecutor, Provider<Retryer> retryer, Logger logger,
62+
Logger.Level logLevel) {
5063
this.client = checkNotNull(client, "client");
64+
this.httpExecutor = checkNotNull(httpExecutor, "httpExecutor");
5165
this.retryer = checkNotNull(retryer, "retryer");
5266
this.logger = checkNotNull(logger, "logger");
5367
this.logLevel = checkNotNull(logLevel, "logLevel");
@@ -58,6 +72,78 @@ public MethodHandler create(Target<?> target, MethodMetadata md, BuildTemplateFr
5872
return new SynchronousMethodHandler(target, client, retryer, logger, logLevel, md, buildTemplateFromArgs, options,
5973
decoder, errorDecoder);
6074
}
75+
76+
public MethodHandler create(Target<?> target, MethodMetadata md, BuildTemplateFromArgs buildTemplateFromArgs,
77+
Options options, IncrementalDecoder.TextStream<?> incrementalCallbackDecoder,
78+
ErrorDecoder errorDecoder) {
79+
return new IncrementalCallbackMethodHandler(target, client, retryer, logger, logLevel, md, buildTemplateFromArgs,
80+
options, incrementalCallbackDecoder, errorDecoder, httpExecutor);
81+
}
82+
}
83+
84+
static final class IncrementalCallbackMethodHandler extends MethodHandler {
85+
private final Lazy<Executor> httpExecutor;
86+
private final IncrementalDecoder.TextStream<?> incDecoder;
87+
88+
private IncrementalCallbackMethodHandler(Target<?> target, Client client, Provider<Retryer> retryer, Logger logger,
89+
Logger.Level logLevel, MethodMetadata metadata,
90+
BuildTemplateFromArgs buildTemplateFromArgs, Options options,
91+
IncrementalDecoder.TextStream<?> incDecoder, ErrorDecoder errorDecoder,
92+
Lazy<Executor> httpExecutor) {
93+
super(target, client, retryer, logger, logLevel, metadata, buildTemplateFromArgs, options, errorDecoder);
94+
this.httpExecutor = checkNotNull(httpExecutor, "httpExecutor for %s", target);
95+
this.incDecoder = checkNotNull(incDecoder, "incrementalCallbackDecoder for %s", target);
96+
}
97+
98+
@Override public Object invoke(final Object[] argv) throws Throwable {
99+
httpExecutor.get().execute(new Runnable() {
100+
@Override public void run() {
101+
Error error = null;
102+
Object arg = argv[metadata.incrementalCallbackIndex()];
103+
IncrementalCallback<Object> incrementalCallback = IncrementalCallback.class.cast(arg);
104+
try {
105+
IncrementalCallbackMethodHandler.super.invoke(argv);
106+
incrementalCallback.onSuccess();
107+
} catch (Error cause) {
108+
// assign to a variable in case .onFailure throws a RTE
109+
error = cause;
110+
incrementalCallback.onFailure(cause);
111+
} catch (Throwable cause) {
112+
incrementalCallback.onFailure(cause);
113+
} finally {
114+
Thread.currentThread().setName(IDLE_THREAD_NAME);
115+
if (error != null)
116+
throw error;
117+
}
118+
}
119+
});
120+
return null; // void.
121+
}
122+
123+
@Override protected Object decode(Object[] argv, Response response) throws Throwable {
124+
Object arg = argv[metadata.incrementalCallbackIndex()];
125+
IncrementalCallback<Object> incrementalCallback = IncrementalCallback.class.cast(arg);
126+
if (metadata.decodeInto().equals(Response.class)) {
127+
incrementalCallback.onNext(response);
128+
} else if (metadata.decodeInto() != Void.class) {
129+
Response.Body body = response.body();
130+
if (body == null)
131+
return null;
132+
Reader reader = body.asReader();
133+
try {
134+
incDecoder.decode(reader, metadata.decodeInto(), incrementalCallback);
135+
} finally {
136+
ensureClosed(body);
137+
}
138+
}
139+
return null; // void
140+
}
141+
142+
@Override protected Request targetRequest(RequestTemplate template) {
143+
Request request = super.targetRequest(template);
144+
Thread.currentThread().setName(THREAD_PREFIX + metadata.configKey());
145+
return request;
146+
}
61147
}
62148

63149
static final class SynchronousMethodHandler extends MethodHandler {
@@ -125,10 +211,8 @@ public Object invoke(Object[] argv) throws Throwable {
125211
}
126212
}
127213

128-
public Object executeAndDecode(Object[] argv, RequestTemplate template)
129-
throws Throwable {
130-
// create the request from a mutable copy of the input template.
131-
Request request = target.apply(new RequestTemplate(template));
214+
public Object executeAndDecode(Object[] argv, RequestTemplate template) throws Throwable {
215+
Request request = targetRequest(template);
132216

133217
if (logLevel.ordinal() > Logger.Level.NONE.ordinal()) {
134218
logger.logRequest(target, logLevel, request);
@@ -159,5 +243,9 @@ public Object executeAndDecode(Object[] argv, RequestTemplate template)
159243
}
160244
}
161245

246+
protected Request targetRequest(RequestTemplate template) {
247+
return target.apply(new RequestTemplate(template));
248+
}
249+
162250
protected abstract Object decode(Object[] argv, Response response) throws Throwable;
163251
}

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

Lines changed: 42 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,19 @@
1515
*/
1616
package feign;
1717

18+
import dagger.Lazy;
1819
import dagger.Provides;
1920
import feign.Request.Options;
2021
import feign.codec.Decoder;
2122
import feign.codec.EncodeException;
2223
import feign.codec.Encoder;
2324
import feign.codec.ErrorDecoder;
25+
import feign.codec.IncrementalDecoder;
2426
import feign.codec.StringDecoder;
27+
import feign.codec.StringIncrementalDecoder;
2528

2629
import javax.inject.Inject;
30+
import javax.inject.Named;
2731
import java.lang.reflect.InvocationHandler;
2832
import java.lang.reflect.Method;
2933
import java.lang.reflect.Proxy;
@@ -35,6 +39,7 @@
3539
import java.util.Map;
3640
import java.util.Map.Entry;
3741
import java.util.Set;
42+
import java.util.concurrent.Executor;
3843

3944
import static feign.Util.checkArgument;
4045
import static feign.Util.checkNotNull;
@@ -47,7 +52,8 @@ public class ReflectiveFeign extends Feign {
4752

4853
private final ParseHandlersByName targetToHandlersByName;
4954

50-
@Inject ReflectiveFeign(ParseHandlersByName targetToHandlersByName) {
55+
@Inject ReflectiveFeign(@Named("http") Lazy<Executor> httpExecutor, ParseHandlersByName targetToHandlersByName) {
56+
super(httpExecutor);
5157
this.targetToHandlersByName = targetToHandlersByName;
5258
}
5359

@@ -118,12 +124,15 @@ static final class ParseHandlersByName {
118124
private final Map<Type, Encoder.Text<? super Object>> encoders = new HashMap<Type, Encoder.Text<? super Object>>();
119125
private final Encoder.Text<Map<String, ?>> formEncoder;
120126
private final Map<Type, Decoder.TextStream<?>> decoders = new HashMap<Type, Decoder.TextStream<?>>();
127+
private final Map<Type, IncrementalDecoder.TextStream<?>> incrementalDecoders =
128+
new HashMap<Type, IncrementalDecoder.TextStream<?>>();
121129
private final ErrorDecoder errorDecoder;
122130
private final MethodHandler.Factory factory;
123131

124132
@SuppressWarnings("unchecked")
125133
@Inject ParseHandlersByName(Contract contract, Options options, Set<Encoder> encoders, Set<Decoder> decoders,
126-
ErrorDecoder errorDecoder, MethodHandler.Factory factory) {
134+
Set<IncrementalDecoder> incrementalDecoders, ErrorDecoder errorDecoder,
135+
MethodHandler.Factory factory) {
127136
this.contract = contract;
128137
this.options = options;
129138
this.factory = factory;
@@ -155,20 +164,22 @@ static final class ParseHandlersByName {
155164
Type type = resolveLastTypeParameter(decoder.getClass(), Decoder.class);
156165
this.decoders.put(type, Decoder.TextStream.class.cast(decoder));
157166
}
167+
StringIncrementalDecoder stringIncrementalDecoder = new StringIncrementalDecoder();
168+
this.incrementalDecoders.put(Void.class, stringIncrementalDecoder);
169+
this.incrementalDecoders.put(Response.class, stringIncrementalDecoder);
170+
this.incrementalDecoders.put(String.class, stringIncrementalDecoder);
171+
for (IncrementalDecoder incrementalDecoder : incrementalDecoders) {
172+
checkState(incrementalDecoder instanceof IncrementalDecoder.TextStream,
173+
"Currently, only IncrementalDecoder.TextStream is supported. Found: ", incrementalDecoder);
174+
Type type = resolveLastTypeParameter(incrementalDecoder.getClass(), IncrementalDecoder.class);
175+
this.incrementalDecoders.put(type, IncrementalDecoder.TextStream.class.cast(incrementalDecoder));
176+
}
158177
}
159178

160179
public Map<String, MethodHandler> apply(Target key) {
161180
List<MethodMetadata> metadata = contract.parseAndValidatateMetadata(key.type());
162181
Map<String, MethodHandler> result = new LinkedHashMap<String, MethodHandler>();
163182
for (MethodMetadata md : metadata) {
164-
Decoder.TextStream decoder = decoders.get(md.decodeInto());
165-
if (decoder == null) {
166-
decoder = decoders.get(Object.class);
167-
}
168-
if (decoder == null) {
169-
throw new IllegalStateException(format("%s needs @Provides(type = Set) Decoder decoder()" +
170-
"{ // Decoder.TextStream<%s> or Decoder.TextStream<Object>}", md.configKey(), md.decodeInto()));
171-
}
172183
BuildTemplateByResolvingArgs buildTemplate;
173184
if (!md.formParams().isEmpty() && md.template().bodyTemplate() == null) {
174185
if (formEncoder == null) {
@@ -189,7 +200,27 @@ public Map<String, MethodHandler> apply(Target key) {
189200
} else {
190201
buildTemplate = new BuildTemplateByResolvingArgs(md);
191202
}
192-
result.put(md.configKey(), factory.create(key, md, buildTemplate, options, decoder, errorDecoder));
203+
if (md.incrementalCallbackIndex() != null) {
204+
IncrementalDecoder.TextStream incrementalDecoder = incrementalDecoders.get(md.decodeInto());
205+
if (incrementalDecoder == null) {
206+
incrementalDecoder = incrementalDecoders.get(Object.class);
207+
}
208+
if (incrementalDecoder == null) {
209+
throw new IllegalStateException(format("%s needs @Provides(type = Set) IncrementalDecoder incrementalDecoder()" +
210+
"{ // IncrementalDecoder.TextStream<%s> or IncrementalDecoder.TextStream<Object>}", md.configKey(), md.decodeInto()));
211+
}
212+
result.put(md.configKey(), factory.create(key, md, buildTemplate, options, incrementalDecoder, errorDecoder));
213+
} else {
214+
Decoder.TextStream decoder = decoders.get(md.decodeInto());
215+
if (decoder == null) {
216+
decoder = decoders.get(Object.class);
217+
}
218+
if (decoder == null) {
219+
throw new IllegalStateException(format("%s needs @Provides(type = Set) Decoder decoder()" +
220+
"{ // Decoder.TextStream<%s> or Decoder.TextStream<Object>}", md.configKey(), md.decodeInto()));
221+
}
222+
result.put(md.configKey(), factory.create(key, md, buildTemplate, options, decoder, errorDecoder));
223+
}
193224
}
194225
return result;
195226
}

0 commit comments

Comments
 (0)