diff --git a/CHANGES.md b/CHANGES.md index bf2eb4e20d..8c6102d972 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,4 +1,5 @@ ### Version 3.0 +* Added support for asynchronous callbacks via `IncrementalCallback` and `IncrementalDecoder.TextStream`. * Wire is now Logger, with configurable Logger.Level. * changed codec to be similar to [WebSocket JSR 356](http://docs.oracle.com/javaee/7/api/javax/websocket/package-summary.html) * Decoder is now `Decoder.TextStream` diff --git a/README.md b/README.md index b60ba16e4c..95195ce31d 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,59 @@ The generic parameter of `Decoder.TextStream` designates which The type param return new SAXDecoder(handlers){}; } ``` +### Asynchronous Incremental Callbacks +If specified as the last argument of a method `IncrementalCallback` fires a background task to add new elements to the callback as they are decoded. Think of `IncrementalCallback` as an asynchronous equivalent to a lazy sequence. + +Here's how one looks: +```java +IncrementalCallback printlnObserver = new IncrementalCallback() { + + public int count; + + @Override public void onNext(Contributor element) { + count++; + } + + @Override public void onSuccess() { + System.out.println("found " + count + " contributors"); + } + + @Override public void onFailure(Throwable cause) { + cause.printStackTrace(); + } +}; +github.contributors("netflix", "feign", printlnObserver); +``` +#### Incremental Decoding +When using an `IncrementalCallback`, you'll need to configure an `IncrementalDecoderi.TextStream` or a general one for all types (`IncrementalDecoder.TextStream`). + +Here's how to wire in a reflective incremental json decoder: +```java +@Provides(type = SET) IncrementalDecoder incrementalDecoder(final Gson gson) { + return new IncrementalDecoder.TextStream() { + + @Override + public void decode(Reader reader, Type type, IncrementalCallback incrementalCallback) throws IOException { + JsonReader jsonReader = new JsonReader(reader); + jsonReader.beginArray(); + while (jsonReader.hasNext()) { + try { + incrementalCallback.onNext(gson.fromJson(jsonReader, type)); + } catch (JsonIOException e) { + if (e.getCause() != null && e.getCause() instanceof IOException) { + throw IOException.class.cast(e.getCause()); + } + throw e; + } + } + jsonReader.endArray(); + } + }; +} +``` + + + ### Multiple Interfaces Feign can produce multiple api interfaces. These are defined as `Target` (default `HardCodedTarget`), which allow for dynamic discovery and decoration of requests prior to execution. diff --git a/feign-core/src/main/java/feign/Contract.java b/feign-core/src/main/java/feign/Contract.java index 7669fb9549..f9b6d7d29b 100644 --- a/feign-core/src/main/java/feign/Contract.java +++ b/feign-core/src/main/java/feign/Contract.java @@ -15,17 +15,18 @@ */ package feign; +import javax.inject.Named; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.lang.reflect.Type; import java.net.URI; import java.util.ArrayList; import java.util.Collection; import java.util.List; -import javax.inject.Named; - import static feign.Util.checkState; import static feign.Util.emptyToNull; +import static feign.Util.resolveLastTypeParameter; /** * Defines what annotations and values are valid on interfaces. @@ -50,7 +51,7 @@ public List parseAndValidatateMetadata(Class declaring) { */ public MethodMetadata parseAndValidatateMetadata(Method method) { MethodMetadata data = new MethodMetadata(); - data.returnType(method.getGenericReturnType()); + data.decodeInto(method.getGenericReturnType()); data.configKey(Feign.configKey(method)); for (Annotation methodAnnotation : method.getAnnotations()) { @@ -69,8 +70,17 @@ public MethodMetadata parseAndValidatateMetadata(Method method) { } if (parameterTypes[i] == URI.class) { data.urlIndex(i); + } else if (IncrementalCallback.class.isAssignableFrom(parameterTypes[i])) { + checkState(method.getReturnType() == void.class, "IncrementalCallback methods must return void: %s", method); + checkState(i == count - 1, "IncrementalCallback must be the last parameter: %s", method); + Type context = method.getGenericParameterTypes()[i]; + Type incrementalCallbackType = resolveLastTypeParameter(context, IncrementalCallback.class); + data.decodeInto(incrementalCallbackType); + data.incrementalCallbackIndex(i); + checkState(incrementalCallbackType != null, "Expected param %s to be IncrementalCallback or IncrementalCallback or a subtype", + context, incrementalCallbackType); } else if (!isHttpAnnotation) { - checkState(data.formParams().isEmpty(), "Body parameters cannot be used with @FormParam parameters."); + checkState(data.formParams().isEmpty(), "Body parameters cannot be used with form parameters."); checkState(data.bodyIndex() == null, "Method has too many Body parameters: %s", method); data.bodyIndex(i); data.bodyType(method.getGenericParameterTypes()[i]); diff --git a/feign-core/src/main/java/feign/Feign.java b/feign-core/src/main/java/feign/Feign.java index b5de31cf68..171116f4dc 100644 --- a/feign-core/src/main/java/feign/Feign.java +++ b/feign-core/src/main/java/feign/Feign.java @@ -15,6 +15,8 @@ */ package feign; + +import dagger.Lazy; import dagger.ObjectGraph; import dagger.Provides; import feign.Logger.NoOpLogger; @@ -23,13 +25,23 @@ import feign.codec.Decoder; import feign.codec.Encoder; import feign.codec.ErrorDecoder; +import feign.codec.IncrementalDecoder; +import javax.inject.Named; +import javax.inject.Singleton; import javax.net.ssl.SSLSocketFactory; +import java.io.Closeable; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; + +import static java.lang.Thread.MIN_PRIORITY; /** * Feign's purpose is to ease development against http apis that feign @@ -38,7 +50,7 @@ * In implementation, Feign is a {@link Feign#newInstance factory} for * generating {@link Target targeted} http apis. */ -public abstract class Feign { +public abstract class Feign implements Closeable { /** * Returns a new instance of an HTTP API, defined by annotations in the @@ -119,6 +131,26 @@ public static class Defaults { @Provides Set noDecoders() { return Collections.emptySet(); } + + @Provides Set noIncrementalDecoders() { + return Collections.emptySet(); + } + + /** + * Used for both http invocation and decoding when incrementalCallbacks are used. + */ + @Provides @Singleton @Named("http") Executor httpExecutor() { + return Executors.newCachedThreadPool(new ThreadFactory() { + @Override public Thread newThread(final Runnable r) { + return new Thread(new Runnable() { + @Override public void run() { + Thread.currentThread().setPriority(MIN_PRIORITY); + r.run(); + } + }, MethodHandler.IDLE_THREAD_NAME); + } + }); + } } /** @@ -162,7 +194,16 @@ private static List modulesForGraph(Object... modules) { return modulesForGraph; } - Feign() { + private final Lazy httpExecutor; + Feign(Lazy httpExecutor) { + this.httpExecutor = httpExecutor; + } + + @Override public void close() { + Executor e = httpExecutor.get(); + if (e instanceof ExecutorService) { + ExecutorService.class.cast(e).shutdownNow(); + } } } diff --git a/feign-core/src/main/java/feign/IncrementalCallback.java b/feign-core/src/main/java/feign/IncrementalCallback.java new file mode 100644 index 0000000000..90173be566 --- /dev/null +++ b/feign-core/src/main/java/feign/IncrementalCallback.java @@ -0,0 +1,67 @@ +package feign; + +/** + * Communicates results as they are {@link feign.codec.Decoder decoded} from + * an {@link Response.Body http response body}. {@link #onNext(Object) onNext} + * will be called for each incremental value of type {@code T}, or not at all + * when there are no values present in the response. Methods that accept + * {@code IncrementalCallback} are asynchronous, which implies background + * processing. + *
+ * {@link #onSuccess() onSuccess} or {@link #onFailure(Throwable)} onFailure} + * will be called when the response is finished, but not both. + *
+ * {@code IncrementalCallback} can be used as an asynchronous alternative to a + * {@code Collection}, or any other use where iterative response parsing is + * worth the additional effort to implement this interface. + *
+ *
+ * Here's an example of implementing {@code IncrementalCallback}: + *
+ * 
+ * IncrementalCallback counter = new IncrementalCallback() {
+ *
+ *   public int count;
+ *
+ *   @Override public void onNext(Contributor element) {
+ *     count++;
+ *   }
+ *
+ *   @Override public void onSuccess() {
+ *     System.out.println("found " + count + " contributors");
+ *   }
+ *
+ *   @Override public void onFailure(Throwable cause) {
+ *     System.err.println("sad face after contributor " + count);
+ *   }
+ * };
+ * github.contributors("netflix", "feign", counter);
+ *
+ * + * @param expected value to decode + */ +public interface IncrementalCallback { + /** + * Invoked as soon as new data is available. Could be invoked many times or + * not at all. + * + * @param element next decoded element. + */ + void onNext(T element); + + /** + * Called when response processing completed successfully. + */ + void onSuccess(); + + /** + * Called when response processing failed for any reason. + *
+ * Common failure cases include {@link FeignException}, + * {@link java.io.IOException}, and {@link feign.codec.DecodeException}. + * However, the cause could be a {@code Throwable} of any kind. + * + * @param cause the reason for the failure + */ + void onFailure(Throwable cause); +} diff --git a/feign-core/src/main/java/feign/MethodHandler.java b/feign-core/src/main/java/feign/MethodHandler.java index 0cef816e9d..42076ff5af 100644 --- a/feign-core/src/main/java/feign/MethodHandler.java +++ b/feign-core/src/main/java/feign/MethodHandler.java @@ -15,14 +15,19 @@ */ package feign; +import dagger.Lazy; import feign.Request.Options; import feign.codec.DecodeException; import feign.codec.Decoder; import feign.codec.ErrorDecoder; +import feign.codec.IncrementalDecoder; import javax.inject.Inject; +import javax.inject.Named; import javax.inject.Provider; import java.io.IOException; +import java.io.Reader; +import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; import static feign.FeignException.errorExecuting; @@ -32,6 +37,12 @@ abstract class MethodHandler { + /** + * same approach as retrofit: temporarily rename threads + */ + static final String THREAD_PREFIX = "Feign-"; + static final String IDLE_THREAD_NAME = THREAD_PREFIX + "Idle"; + /** * Those using guava will implement as {@code Function}. */ @@ -42,12 +53,15 @@ static interface BuildTemplateFromArgs { static class Factory { private final Client client; + private final Lazy httpExecutor; private final Provider retryer; private final Logger logger; private final Logger.Level logLevel; - @Inject Factory(Client client, Provider retryer, Logger logger, Logger.Level logLevel) { + @Inject Factory(Client client, @Named("http") Lazy httpExecutor, Provider retryer, Logger logger, + Logger.Level logLevel) { this.client = checkNotNull(client, "client"); + this.httpExecutor = checkNotNull(httpExecutor, "httpExecutor"); this.retryer = checkNotNull(retryer, "retryer"); this.logger = checkNotNull(logger, "logger"); this.logLevel = checkNotNull(logLevel, "logLevel"); @@ -58,6 +72,78 @@ public MethodHandler create(Target target, MethodMetadata md, BuildTemplateFr return new SynchronousMethodHandler(target, client, retryer, logger, logLevel, md, buildTemplateFromArgs, options, decoder, errorDecoder); } + + public MethodHandler create(Target target, MethodMetadata md, BuildTemplateFromArgs buildTemplateFromArgs, + Options options, IncrementalDecoder.TextStream incrementalCallbackDecoder, + ErrorDecoder errorDecoder) { + return new IncrementalCallbackMethodHandler(target, client, retryer, logger, logLevel, md, buildTemplateFromArgs, + options, incrementalCallbackDecoder, errorDecoder, httpExecutor); + } + } + + static final class IncrementalCallbackMethodHandler extends MethodHandler { + private final Lazy httpExecutor; + private final IncrementalDecoder.TextStream incDecoder; + + private IncrementalCallbackMethodHandler(Target target, Client client, Provider retryer, Logger logger, + Logger.Level logLevel, MethodMetadata metadata, + BuildTemplateFromArgs buildTemplateFromArgs, Options options, + IncrementalDecoder.TextStream incDecoder, ErrorDecoder errorDecoder, + Lazy httpExecutor) { + super(target, client, retryer, logger, logLevel, metadata, buildTemplateFromArgs, options, errorDecoder); + this.httpExecutor = checkNotNull(httpExecutor, "httpExecutor for %s", target); + this.incDecoder = checkNotNull(incDecoder, "incrementalCallbackDecoder for %s", target); + } + + @Override public Object invoke(final Object[] argv) throws Throwable { + httpExecutor.get().execute(new Runnable() { + @Override public void run() { + Error error = null; + Object arg = argv[metadata.incrementalCallbackIndex()]; + IncrementalCallback incrementalCallback = IncrementalCallback.class.cast(arg); + try { + IncrementalCallbackMethodHandler.super.invoke(argv); + incrementalCallback.onSuccess(); + } catch (Error cause) { + // assign to a variable in case .onFailure throws a RTE + error = cause; + incrementalCallback.onFailure(cause); + } catch (Throwable cause) { + incrementalCallback.onFailure(cause); + } finally { + Thread.currentThread().setName(IDLE_THREAD_NAME); + if (error != null) + throw error; + } + } + }); + return null; // void. + } + + @Override protected Object decode(Object[] argv, Response response) throws Throwable { + Object arg = argv[metadata.incrementalCallbackIndex()]; + IncrementalCallback incrementalCallback = IncrementalCallback.class.cast(arg); + if (metadata.decodeInto().equals(Response.class)) { + incrementalCallback.onNext(response); + } else if (metadata.decodeInto() != Void.class) { + Response.Body body = response.body(); + if (body == null) + return null; + Reader reader = body.asReader(); + try { + incDecoder.decode(reader, metadata.decodeInto(), incrementalCallback); + } finally { + ensureClosed(body); + } + } + return null; // void + } + + @Override protected Request targetRequest(RequestTemplate template) { + Request request = super.targetRequest(template); + Thread.currentThread().setName(THREAD_PREFIX + metadata.configKey()); + return request; + } } static final class SynchronousMethodHandler extends MethodHandler { @@ -72,13 +158,13 @@ private SynchronousMethodHandler(Target target, Client client, Provider Logger.Level.NONE.ordinal()) { logger.logRequest(target, logLevel, request); @@ -159,5 +243,9 @@ public Object executeAndDecode(Object[] argv, RequestTemplate template) } } + protected Request targetRequest(RequestTemplate template) { + return target.apply(new RequestTemplate(template)); + } + protected abstract Object decode(Object[] argv, Response response) throws Throwable; } diff --git a/feign-core/src/main/java/feign/MethodMetadata.java b/feign-core/src/main/java/feign/MethodMetadata.java index 05a61e8f0c..5463af09b9 100644 --- a/feign-core/src/main/java/feign/MethodMetadata.java +++ b/feign-core/src/main/java/feign/MethodMetadata.java @@ -24,12 +24,14 @@ import java.util.Map; public final class MethodMetadata implements Serializable { + MethodMetadata() { } private String configKey; - private transient Type returnType; + private transient Type decodeInto; private Integer urlIndex; + private Integer incrementalCallbackIndex; private Integer bodyIndex; private transient Type bodyType; private RequestTemplate template = new RequestTemplate(); @@ -48,12 +50,16 @@ MethodMetadata configKey(String configKey) { return this; } - public Type returnType() { - return returnType; + /** + * Method return type unless there is an {@link IncrementalCallback} arg. In this case, it is the type parameter of the + * incrementalCallback. + */ + public Type decodeInto() { + return decodeInto; } - MethodMetadata returnType(Type returnType) { - this.returnType = returnType; + MethodMetadata decodeInto(Type decodeInto) { + this.decodeInto = decodeInto; return this; } @@ -66,6 +72,15 @@ MethodMetadata urlIndex(Integer urlIndex) { return this; } + public Integer incrementalCallbackIndex() { + return incrementalCallbackIndex; + } + + MethodMetadata incrementalCallbackIndex(Integer incrementalCallbackIndex) { + this.incrementalCallbackIndex = incrementalCallbackIndex; + return this; + } + public Integer bodyIndex() { return bodyIndex; } @@ -97,4 +112,5 @@ public Map> indexToName() { } private static final long serialVersionUID = 1L; + } diff --git a/feign-core/src/main/java/feign/ReflectiveFeign.java b/feign-core/src/main/java/feign/ReflectiveFeign.java index 50289e16e8..d4e227dc74 100644 --- a/feign-core/src/main/java/feign/ReflectiveFeign.java +++ b/feign-core/src/main/java/feign/ReflectiveFeign.java @@ -15,15 +15,19 @@ */ package feign; +import dagger.Lazy; import dagger.Provides; import feign.Request.Options; import feign.codec.Decoder; import feign.codec.EncodeException; import feign.codec.Encoder; import feign.codec.ErrorDecoder; +import feign.codec.IncrementalDecoder; import feign.codec.StringDecoder; +import feign.codec.StringIncrementalDecoder; import javax.inject.Inject; +import javax.inject.Named; import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy; @@ -35,6 +39,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import java.util.concurrent.Executor; import static feign.Util.checkArgument; import static feign.Util.checkNotNull; @@ -47,7 +52,8 @@ public class ReflectiveFeign extends Feign { private final ParseHandlersByName targetToHandlersByName; - @Inject ReflectiveFeign(ParseHandlersByName targetToHandlersByName) { + @Inject ReflectiveFeign(@Named("http") Lazy httpExecutor, ParseHandlersByName targetToHandlersByName) { + super(httpExecutor); this.targetToHandlersByName = targetToHandlersByName; } @@ -118,12 +124,15 @@ static final class ParseHandlersByName { private final Map> encoders = new HashMap>(); private final Encoder.Text> formEncoder; private final Map> decoders = new HashMap>(); + private final Map> incrementalDecoders = + new HashMap>(); private final ErrorDecoder errorDecoder; private final MethodHandler.Factory factory; @SuppressWarnings("unchecked") @Inject ParseHandlersByName(Contract contract, Options options, Set encoders, Set decoders, - ErrorDecoder errorDecoder, MethodHandler.Factory factory) { + Set incrementalDecoders, ErrorDecoder errorDecoder, + MethodHandler.Factory factory) { this.contract = contract; this.options = options; this.factory = factory; @@ -155,20 +164,22 @@ static final class ParseHandlersByName { Type type = resolveLastTypeParameter(decoder.getClass(), Decoder.class); this.decoders.put(type, Decoder.TextStream.class.cast(decoder)); } + StringIncrementalDecoder stringIncrementalDecoder = new StringIncrementalDecoder(); + this.incrementalDecoders.put(Void.class, stringIncrementalDecoder); + this.incrementalDecoders.put(Response.class, stringIncrementalDecoder); + this.incrementalDecoders.put(String.class, stringIncrementalDecoder); + for (IncrementalDecoder incrementalDecoder : incrementalDecoders) { + checkState(incrementalDecoder instanceof IncrementalDecoder.TextStream, + "Currently, only IncrementalDecoder.TextStream is supported. Found: ", incrementalDecoder); + Type type = resolveLastTypeParameter(incrementalDecoder.getClass(), IncrementalDecoder.class); + this.incrementalDecoders.put(type, IncrementalDecoder.TextStream.class.cast(incrementalDecoder)); + } } public Map apply(Target key) { List metadata = contract.parseAndValidatateMetadata(key.type()); Map result = new LinkedHashMap(); for (MethodMetadata md : metadata) { - Decoder.TextStream decoder = decoders.get(md.returnType()); - if (decoder == null) { - decoder = decoders.get(Object.class); - } - if (decoder == null) { - throw new IllegalStateException(format("%s needs @Provides(type = Set) Decoder decoder()" + - "{ // Decoder.TextStream<%s> or Decoder.TextStream}", md.configKey(), md.returnType())); - } BuildTemplateByResolvingArgs buildTemplate; if (!md.formParams().isEmpty() && md.template().bodyTemplate() == null) { if (formEncoder == null) { @@ -183,13 +194,33 @@ public Map apply(Target key) { } if (encoder == null) { throw new IllegalStateException(format("%s needs @Provides(type = Set) Encoder encoder()" + - "{ // Encoder.Text<%s> or Encoder.Text}", md.bodyType(), md.returnType())); + "{ // Encoder.Text<%s> or Encoder.Text}", md.bodyType(), md.decodeInto())); } buildTemplate = new BuildEncodedTemplateFromArgs(md, encoder); } else { buildTemplate = new BuildTemplateByResolvingArgs(md); } - result.put(md.configKey(), factory.create(key, md, buildTemplate, options, decoder, errorDecoder)); + if (md.incrementalCallbackIndex() != null) { + IncrementalDecoder.TextStream incrementalDecoder = incrementalDecoders.get(md.decodeInto()); + if (incrementalDecoder == null) { + incrementalDecoder = incrementalDecoders.get(Object.class); + } + if (incrementalDecoder == null) { + throw new IllegalStateException(format("%s needs @Provides(type = Set) IncrementalDecoder incrementalDecoder()" + + "{ // IncrementalDecoder.TextStream<%s> or IncrementalDecoder.TextStream}", md.configKey(), md.decodeInto())); + } + result.put(md.configKey(), factory.create(key, md, buildTemplate, options, incrementalDecoder, errorDecoder)); + } else { + Decoder.TextStream decoder = decoders.get(md.decodeInto()); + if (decoder == null) { + decoder = decoders.get(Object.class); + } + if (decoder == null) { + throw new IllegalStateException(format("%s needs @Provides(type = Set) Decoder decoder()" + + "{ // Decoder.TextStream<%s> or Decoder.TextStream}", md.configKey(), md.decodeInto())); + } + result.put(md.configKey(), factory.create(key, md, buildTemplate, options, decoder, errorDecoder)); + } } return result; } diff --git a/feign-core/src/main/java/feign/codec/Decoder.java b/feign-core/src/main/java/feign/codec/Decoder.java index afcc6406ee..8492d143b4 100644 --- a/feign-core/src/main/java/feign/codec/Decoder.java +++ b/feign-core/src/main/java/feign/codec/Decoder.java @@ -15,41 +15,30 @@ */ package feign.codec; +import feign.FeignException; +import feign.Response; + import java.io.IOException; import java.io.Reader; import java.lang.reflect.Type; -import feign.FeignException; -import feign.Response; - /** * Decodes an HTTP response into a given type. Invoked when * {@link Response#status()} is in the 2xx range. Like * {@code javax.websocket.Decoder}, except that the decode method is passed the * generic type of the target.
- *
- *
- * Error handling
- *
- * Responses where {@link Response#status()} is not in the 2xx range are - * classified as errors, addressed by the {@link ErrorDecoder}. That said, - * certain RPC apis return errors defined in the {@link Response#body()} even on - * a 200 status. For example, in the DynECT api, a job still running condition - * is returned with a 200 status, encoded in json. When scenarios like this - * occur, you should raise an application-specific exception (which may be - * {@link feign.RetryableException retryable}). * * @param input that can be derived from {@link feign.Response.Body}. * @param widest type an instance of this can decode. */ public interface Decoder { /** - * Implement this to decode a resource to an object of the specified type. + * Implement this to decode a resource to an object into a single object. * If you need to wrap exceptions, please do so via {@link DecodeException}. * * @param input if {@code Closeable}, no need to close this, as the caller - * manages resources. - * @param type Target object type. + * manages resources. + * @param type Target object type. * @return instance of {@code type} * @throws IOException will be propagated safely to the caller. * @throws DecodeException when decoding failed due to a checked exception diff --git a/feign-core/src/main/java/feign/codec/ErrorDecoder.java b/feign-core/src/main/java/feign/codec/ErrorDecoder.java index d9982360e5..273202d400 100644 --- a/feign-core/src/main/java/feign/codec/ErrorDecoder.java +++ b/feign-core/src/main/java/feign/codec/ErrorDecoder.java @@ -51,6 +51,16 @@ * * } * + *
+ * Error handling
+ *
+ * Responses where {@link Response#status()} is not in the 2xx range are + * classified as errors, addressed by the {@link ErrorDecoder}. That said, + * certain RPC apis return errors defined in the {@link Response#body()} even on + * a 200 status. For example, in the DynECT api, a job still running condition + * is returned with a 200 status, encoded in json. When scenarios like this + * occur, you should raise an application-specific exception (which may be + * {@link feign.RetryableException retryable}). */ public interface ErrorDecoder { diff --git a/feign-core/src/main/java/feign/codec/IncrementalDecoder.java b/feign-core/src/main/java/feign/codec/IncrementalDecoder.java new file mode 100644 index 0000000000..30f27a04bf --- /dev/null +++ b/feign-core/src/main/java/feign/codec/IncrementalDecoder.java @@ -0,0 +1,114 @@ +/* + * Copyright 2013 Netflix, Inc. + * + * 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.codec; + +import feign.FeignException; +import feign.IncrementalCallback; + +import java.io.IOException; +import java.io.Reader; +import java.lang.reflect.Type; + +/** + * Decodes an HTTP response incrementally into an {@link IncrementalCallback} + * via a series of {@link IncrementalCallback#onNext(Object) onNext} calls. + *

+ * Invoked when {@link feign.Response#status()} is in the 2xx range. + * + * @param input that can be derived from {@link feign.Response.Body}. + * @param widest type an instance of this can decode. + */ +public interface IncrementalDecoder { + /** + * Implement this to decode a resource to an object into a single object. + * If you need to wrap exceptions, please do so via {@link feign.codec.DecodeException}. + *
+ * Do not call {@link feign.IncrementalCallback#onSuccess() onSuccess} or + * {@link feign.IncrementalCallback#onFailure onFailure}. + * + * @param input if {@code Closeable}, no need to close this, as the caller + * manages resources. + * @param type type parameter of {@link feign.IncrementalCallback#onNext}. + * @param incrementalCallback call {@link feign.IncrementalCallback#onNext onNext} + * each time an object of {@code type} is decoded + * from the response. + * @throws java.io.IOException will be propagated safely to the caller. + * @throws feign.codec.DecodeException when decoding failed due to a checked exception + * besides IOException. + * @throws feign.FeignException when decoding succeeds, but conveys the operation + * failed. + */ + void decode(I input, Type type, IncrementalCallback incrementalCallback) + throws IOException, DecodeException, FeignException; + + /** + * Used for text-based apis, follows + * {@link feign.codec.IncrementalDecoder#decode(Object, java.lang.reflect.Type, IncrementalCallback)} + * semantics, applied to inputs of type {@link java.io.Reader}.
+ * Ex.
+ *

+ * 

+   * public class GsonDecoder implements Decoder.TextStream<Object> {
+   *   private final Gson gson;
+   *
+   *   public GsonDecoder(Gson gson) {
+   *     this.gson = gson;
+   *   }
+   *
+   *   @Override
+   *   public Object decode(Reader reader, Type type) throws IOException {
+   *     try {
+   *       return gson.fromJson(reader, type);
+   *     } catch (JsonIOException e) {
+   *       if (e.getCause() != null &&
+   *           e.getCause() instanceof IOException) {
+   *         throw IOException.class.cast(e.getCause());
+   *       }
+   *       throw e;
+   *     }
+   *   }
+   * }
+   *
+ * 
+   * public class GsonIncrementalDecoder implements IncrementalDecoder {
+   *   private final Gson gson;
+   *
+   *   public GsonIncrementalDecoder(Gson gson) {
+   *     this.gson = gson;
+   *   }
+   *
+   *   @Override public void decode(Reader reader, Type type, IncrementalCallback incrementalCallback) throws Exception {
+   *     JsonReader jsonReader = new JsonReader(reader);
+   *     jsonReader.beginArray();
+   *     while (jsonReader.hasNext()) {
+   *       try {
+   *          incrementalCallback.onNext(gson.fromJson(jsonReader, type));
+   *       } catch (JsonIOException e) {
+   *         if (e.getCause() != null &&
+   *             e.getCause() instanceof IOException) {
+   *           throw IOException.class.cast(e.getCause());
+   *         }
+   *         throw e;
+   *       }
+   *     }
+   *     jsonReader.endArray();
+   *   }
+   * }
+   * 
+   */
+  public interface TextStream extends IncrementalDecoder {
+  }
+}
diff --git a/feign-core/src/main/java/feign/codec/StringIncrementalDecoder.java b/feign-core/src/main/java/feign/codec/StringIncrementalDecoder.java
new file mode 100644
index 0000000000..3e9dc8e005
--- /dev/null
+++ b/feign-core/src/main/java/feign/codec/StringIncrementalDecoder.java
@@ -0,0 +1,32 @@
+/*
+ * Copyright 2013 Netflix, Inc.
+ *
+ * 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.codec;
+
+import feign.IncrementalCallback;
+
+import java.io.IOException;
+import java.io.Reader;
+import java.lang.reflect.Type;
+
+public class StringIncrementalDecoder implements IncrementalDecoder.TextStream {
+  private static final StringDecoder STRING_DECODER = new StringDecoder();
+
+  @Override
+  public void decode(Reader reader, Type type, IncrementalCallback incrementalCallback)
+      throws IOException {
+    incrementalCallback.onNext(STRING_DECODER.decode(reader, type));
+  }
+}
diff --git a/feign-core/src/test/java/feign/DefaultContractTest.java b/feign-core/src/test/java/feign/DefaultContractTest.java
index dc66330073..958a7785fc 100644
--- a/feign-core/src/test/java/feign/DefaultContractTest.java
+++ b/feign-core/src/test/java/feign/DefaultContractTest.java
@@ -21,6 +21,7 @@
 import org.testng.annotations.Test;
 
 import javax.inject.Named;
+import java.lang.reflect.Type;
 import java.net.URI;
 import java.util.List;
 
@@ -237,4 +238,45 @@ interface HeaderParams {
     assertEquals(md.template().headers().get("Auth-Token"), ImmutableSet.of("{Auth-Token}"));
     assertEquals(md.indexToName().get(0), ImmutableSet.of("Auth-Token"));
   }
+
+  interface WithIncrementalCallback {
+    @RequestLine("GET /") void valid(IncrementalCallback> one);
+
+    @RequestLine("GET /{path}") void badOrder(IncrementalCallback> one, @Named("path") String path);
+
+    @RequestLine("GET /") Response returnType(IncrementalCallback> one);
+
+    @RequestLine("GET /") void wildcardExtends(IncrementalCallback> one);
+
+    @RequestLine("GET /") void subtype(ParameterizedIncrementalCallback> one);
+  }
+
+  static final List listString = null;
+
+  interface ParameterizedIncrementalCallback> extends IncrementalCallback {
+  }
+
+  @Test public void methodCanHaveIncrementalCallbackParam() throws Exception {
+    contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("valid", IncrementalCallback.class));
+  }
+
+  @Test public void methodMetadataReturnTypeOnObservableMethodIsItsTypeParameter() throws Exception {
+    Type listStringType = getClass().getDeclaredField("listString").getGenericType();
+    MethodMetadata md = contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("valid", IncrementalCallback.class));
+    assertEquals(md.decodeInto(), listStringType);
+    md = contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("wildcardExtends", IncrementalCallback.class));
+    assertEquals(md.decodeInto(), listStringType);
+    md = contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("subtype", ParameterizedIncrementalCallback.class));
+    assertEquals(md.decodeInto(), listStringType);
+  }
+
+  @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp = ".*the last parameter.*")
+  public void incrementalCallbackParamMustBeLast() throws Exception {
+    contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("badOrder", IncrementalCallback.class, String.class));
+  }
+
+  @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp = ".*must return void.*")
+  public void incrementalCallbackMethodMustReturnVoid() throws Exception {
+    contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("returnType", IncrementalCallback.class));
+  }
 }
diff --git a/feign-core/src/test/java/feign/FeignTest.java b/feign-core/src/test/java/feign/FeignTest.java
index 3155e1549f..ac91708ba7 100644
--- a/feign-core/src/test/java/feign/FeignTest.java
+++ b/feign-core/src/test/java/feign/FeignTest.java
@@ -19,10 +19,10 @@
 import com.google.mockwebserver.MockResponse;
 import com.google.mockwebserver.MockWebServer;
 import com.google.mockwebserver.SocketPolicy;
+import dagger.Lazy;
 import dagger.Module;
 import dagger.Provides;
 import feign.codec.Decoder;
-import feign.codec.EncodeException;
 import feign.codec.Encoder;
 import feign.codec.ErrorDecoder;
 import feign.codec.StringDecoder;
@@ -38,14 +38,35 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static dagger.Provides.Type.SET;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
 
 @Test
 // unbound wildcards are not currently injectable in dagger.
 @SuppressWarnings("rawtypes")
 public class FeignTest {
+
+  @Test public void closeShutsdownExecutorService() throws IOException, InterruptedException {
+    final ExecutorService service = Executors.newCachedThreadPool();
+    new Feign(new Lazy() {
+      @Override public Executor get() {
+        return service;
+      }
+    }) {
+      @Override public  T newInstance(Target target) {
+        return null;
+      }
+    }.close();
+    assertTrue(service.isShutdown());
+  }
+
   interface TestInterface {
     @RequestLine("POST /") String post();
 
@@ -54,15 +75,19 @@ interface TestInterface {
     void login(
         @Named("customer_name") String customer, @Named("user_name") String user, @Named("password") String password);
 
-    @RequestLine("POST /")
-    void body(List contents);
+    @RequestLine("POST /") void body(List contents);
 
-    @RequestLine("POST /")
-    void form(
+    @RequestLine("POST /") void form(
         @Named("customer_name") String customer, @Named("user_name") String user, @Named("password") String password);
 
     @RequestLine("GET /{1}/{2}") Response uriParam(@Named("1") String one, URI endpoint, @Named("2") String two);
 
+    @RequestLine("POST /") void incrementVoid(IncrementalCallback incrementalCallback);
+
+    @RequestLine("POST /") void incrementString(IncrementalCallback incrementalCallback);
+
+    @RequestLine("POST /") void incrementResponse(IncrementalCallback incrementalCallback);
+
     @dagger.Module(overrides = true, library = true)
     static class Module {
       @Provides(type = SET) Encoder defaultEncoder() {
@@ -80,6 +105,117 @@ static class Module {
           }
         };
       }
+
+      // just run synchronously
+      @Provides @Singleton @Named("http") Executor httpExecutor() {
+        return new Executor() {
+          @Override public void execute(Runnable command) {
+            command.run();
+          }
+        };
+      }
+    }
+  }
+
+  @Test
+  public void incrementVoid() throws IOException, InterruptedException {
+    final MockWebServer server = new MockWebServer();
+    server.enqueue(new MockResponse().setResponseCode(200).setBody("foo"));
+    server.play();
+
+    try {
+      TestInterface api = Feign.create(TestInterface.class, server.getUrl("").toString(), new TestInterface.Module());
+
+      final AtomicBoolean success = new AtomicBoolean();
+
+      IncrementalCallback incrementalCallback = new IncrementalCallback() {
+
+        @Override public void onNext(Void element) {
+          fail("on next isn't valid for void");
+        }
+
+        @Override public void onSuccess() {
+          success.set(true);
+        }
+
+        @Override public void onFailure(Throwable cause) {
+          fail(cause.getMessage());
+        }
+      };
+      api.incrementVoid(incrementalCallback);
+
+      assertTrue(success.get());
+      assertEquals(server.getRequestCount(), 1);
+    } finally {
+      server.shutdown();
+    }
+  }
+
+  @Test
+  public void incrementResponse() throws IOException, InterruptedException {
+    final MockWebServer server = new MockWebServer();
+    server.enqueue(new MockResponse().setResponseCode(200).setBody("foo"));
+    server.play();
+
+    try {
+      TestInterface api = Feign.create(TestInterface.class, server.getUrl("").toString(), new TestInterface.Module());
+
+      final AtomicBoolean success = new AtomicBoolean();
+
+      IncrementalCallback incrementalCallback = new IncrementalCallback() {
+
+        @Override public void onNext(Response element) {
+          assertEquals(element.status(), 200);
+        }
+
+        @Override public void onSuccess() {
+          success.set(true);
+        }
+
+        @Override public void onFailure(Throwable cause) {
+          fail(cause.getMessage());
+        }
+      };
+      api.incrementResponse(incrementalCallback);
+
+      assertTrue(success.get());
+      assertEquals(server.getRequestCount(), 1);
+    } finally {
+      server.shutdown();
+    }
+  }
+
+  @Test
+  public void incrementString() throws IOException, InterruptedException {
+    final MockWebServer server = new MockWebServer();
+    server.enqueue(new MockResponse().setResponseCode(200).setBody("foo"));
+    server.play();
+
+    try {
+      TestInterface api = Feign.create(TestInterface.class, server.getUrl("").toString(), new TestInterface.Module());
+
+      final AtomicBoolean success = new AtomicBoolean();
+
+      IncrementalCallback incrementalCallback = new IncrementalCallback() {
+
+        @Override public void onNext(String element) {
+          assertEquals(element, "foo");
+        }
+
+        @Override public void onSuccess() {
+          success.set(true);
+        }
+
+        @Override public void onFailure(Throwable cause) {
+          fail(cause.getMessage());
+        }
+      };
+      api.incrementString(incrementalCallback);
+
+      assertTrue(success.get());
+      assertEquals(server.getRequestCount(), 1);
+    } finally {
+      server.shutdown();
     }
   }
 
diff --git a/feign-core/src/test/java/feign/examples/GitHubExample.java b/feign-core/src/test/java/feign/examples/GitHubExample.java
index 502f0f3e22..5ecc8cb1d4 100644
--- a/feign-core/src/test/java/feign/examples/GitHubExample.java
+++ b/feign-core/src/test/java/feign/examples/GitHubExample.java
@@ -15,24 +15,26 @@
  */
 package feign.examples;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.gson.Gson;
 import com.google.gson.JsonIOException;
+import com.google.gson.stream.JsonReader;
 import dagger.Module;
 import dagger.Provides;
 import feign.Feign;
+import feign.IncrementalCallback;
 import feign.RequestLine;
 import feign.codec.Decoder;
+import feign.codec.IncrementalDecoder;
 
+import javax.inject.Inject;
 import javax.inject.Named;
+import javax.inject.Singleton;
 import java.io.IOException;
 import java.io.Reader;
 import java.lang.reflect.Type;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 
-import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.ANY;
-import static com.fasterxml.jackson.annotation.PropertyAccessor.FIELD;
-import static com.fasterxml.jackson.databind.DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES;
 import static dagger.Provides.Type.SET;
 
 /**
@@ -43,6 +45,10 @@ public class GitHubExample {
   interface GitHub {
     @RequestLine("GET /repos/{owner}/{repo}/contributors")
     List contributors(@Named("owner") String owner, @Named("repo") String repo);
+
+    @RequestLine("GET /repos/{owner}/{repo}/contributors")
+    void contributors(@Named("owner") String owner, @Named("repo") String repo,
+                      IncrementalCallback contributors);
   }
 
   static class Contributor {
@@ -50,14 +56,46 @@ static class Contributor {
     int contributions;
   }
 
-  public static void main(String... args) {
+  public static void main(String... args) throws InterruptedException {
     GitHub github = Feign.create(GitHub.class, "https://api.github.com", new GsonModule());
 
-    // Fetch and print a list of the contributors to this library.
+    System.out.println("Let's fetch and print a list of the contributors to this library.");
     List contributors = github.contributors("netflix", "feign");
     for (Contributor contributor : contributors) {
       System.out.println(contributor.login + " (" + contributor.contributions + ")");
     }
+
+    final CountDownLatch latch = new CountDownLatch(1);
+
+    System.out.println("Now, let's do it as an incremental async task.");
+    IncrementalCallback task = new IncrementalCallback() {
+
+      public int count;
+
+      // parsed directly from the text stream without an intermediate collection.
+      @Override public void onNext(Contributor contributor) {
+        System.out.println(contributor.login + " (" + contributor.contributions + ")");
+        count++;
+      }
+
+      @Override public void onSuccess() {
+        System.out.println("found " + count + " contributors");
+        latch.countDown();
+      }
+
+      @Override public void onFailure(Throwable cause) {
+        cause.printStackTrace();
+        latch.countDown();
+      }
+    };
+
+    // fire a task in the background.
+    github.contributors("netflix", "feign", task);
+
+    // wait for the task to complete.
+    latch.await();
+
+    System.exit(0);
   }
 
   /**
@@ -65,37 +103,49 @@ public static void main(String... args) {
    */
   @Module(overrides = true, library = true)
   static class GsonModule {
-    @Provides(type = SET) Decoder decoder() {
-      return new Decoder.TextStream() {
-        Gson gson = new Gson();
-
-        @Override public Object decode(Reader reader, Type type) throws IOException {
-          try {
-            return gson.fromJson(reader, type);
-          } catch (JsonIOException e) {
-            if (e.getCause() != null && e.getCause() instanceof IOException) {
-              throw IOException.class.cast(e.getCause());
-            }
-            throw e;
-          }
-        }
-      };
+    @Provides @Singleton Gson gson() {
+      return new Gson();
+    }
+
+    @Provides(type = SET) Decoder decoder(GsonDecoder gsonDecoder) {
+      return gsonDecoder;
+    }
+
+    @Provides(type = SET) IncrementalDecoder incrementalDecoder(GsonDecoder gsonDecoder) {
+      return gsonDecoder;
     }
   }
 
-  /**
-   * Here's how to wire jackson deserialization.
-   */
-  @Module(overrides = true, library = true)
-  static class JacksonModule {
-    @Provides(type = SET) Decoder decoder() {
-      return new Decoder.TextStream() {
-        ObjectMapper mapper = new ObjectMapper().disable(FAIL_ON_UNKNOWN_PROPERTIES).setVisibility(FIELD, ANY);
+  static class GsonDecoder implements Decoder.TextStream, IncrementalDecoder.TextStream {
+    private final Gson gson;
+
+    @Inject GsonDecoder(Gson gson) {
+      this.gson = gson;
+    }
+
+    @Override public Object decode(Reader reader, Type type) throws IOException {
+      return fromJson(new JsonReader(reader), type);
+    }
+
+    @Override
+    public void decode(Reader reader, Type type, IncrementalCallback incrementalCallback) throws IOException {
+      JsonReader jsonReader = new JsonReader(reader);
+      jsonReader.beginArray();
+      while (jsonReader.hasNext()) {
+        incrementalCallback.onNext(fromJson(jsonReader, type));
+      }
+      jsonReader.endArray();
+    }
 
-        @Override public Object decode(Reader reader, final Type type) throws IOException {
-          return mapper.readValue(reader, mapper.constructType(type));
+    private Object fromJson(JsonReader jsonReader, Type type) throws IOException {
+      try {
+        return gson.fromJson(jsonReader, type);
+      } catch (JsonIOException e) {
+        if (e.getCause() != null && e.getCause() instanceof IOException) {
+          throw IOException.class.cast(e.getCause());
         }
-      };
+        throw e;
+      }
     }
   }
 }
diff --git a/feign-jaxrs/src/main/java/feign/jaxrs/JAXRSModule.java b/feign-jaxrs/src/main/java/feign/jaxrs/JAXRSModule.java
index 9e766387a1..e9e2a5dba2 100644
--- a/feign-jaxrs/src/main/java/feign/jaxrs/JAXRSModule.java
+++ b/feign-jaxrs/src/main/java/feign/jaxrs/JAXRSModule.java
@@ -15,9 +15,10 @@
  */
 package feign.jaxrs;
 
-import java.lang.annotation.Annotation;
-import java.lang.reflect.Method;
-import java.util.Collection;
+import dagger.Provides;
+import feign.Body;
+import feign.Contract;
+import feign.MethodMetadata;
 
 import javax.ws.rs.Consumes;
 import javax.ws.rs.FormParam;
@@ -27,11 +28,9 @@
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
-
-import dagger.Provides;
-import feign.Body;
-import feign.Contract;
-import feign.MethodMetadata;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.util.Collection;
 
 import static feign.Util.checkState;
 
@@ -44,7 +43,7 @@ public final class JAXRSModule {
     return new JAXRSContract();
   }
 
-  static final class JAXRSContract extends Contract {
+  public static final class JAXRSContract extends Contract {
 
     @Override
     protected void processAnnotationOnMethod(MethodMetadata data, Annotation methodAnnotation, Method method) {
diff --git a/feign-jaxrs/src/test/java/feign/jaxrs/JAXRSContractTest.java b/feign-jaxrs/src/test/java/feign/jaxrs/JAXRSContractTest.java
index 64621840d8..36888cad83 100644
--- a/feign-jaxrs/src/test/java/feign/jaxrs/JAXRSContractTest.java
+++ b/feign-jaxrs/src/test/java/feign/jaxrs/JAXRSContractTest.java
@@ -17,18 +17,13 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
-
 import com.google.gson.reflect.TypeToken;
-import feign.RequestLine;
+import feign.Body;
+import feign.IncrementalCallback;
+import feign.MethodMetadata;
+import feign.Response;
 import org.testng.annotations.Test;
 
-import java.lang.annotation.ElementType;
-import java.lang.annotation.Retention;
-import java.lang.annotation.RetentionPolicy;
-import java.lang.annotation.Target;
-import java.net.URI;
-import java.util.List;
-
 import javax.ws.rs.DELETE;
 import javax.ws.rs.FormParam;
 import javax.ws.rs.GET;
@@ -40,10 +35,13 @@
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
 import javax.ws.rs.QueryParam;
-
-import feign.Body;
-import feign.MethodMetadata;
-import feign.Response;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import java.lang.reflect.Type;
+import java.net.URI;
+import java.util.List;
 
 import static feign.jaxrs.JAXRSModule.CONTENT_TYPE;
 import static javax.ws.rs.HttpMethod.DELETE;
@@ -264,4 +262,45 @@ interface HeaderParams {
     assertEquals(md.template().headers().get("Auth-Token"), ImmutableSet.of("{Auth-Token}"));
     assertEquals(md.indexToName().get(0), ImmutableSet.of("Auth-Token"));
   }
+
+  interface WithIncrementalCallback {
+    @GET @Path("/") void valid(IncrementalCallback> one);
+
+    @GET @Path("/{path}") void badOrder(IncrementalCallback> one, @PathParam("path") String path);
+
+    @GET @Path("/") Response returnType(IncrementalCallback> one);
+
+    @GET @Path("/") void wildcardExtends(IncrementalCallback> one);
+
+    @GET @Path("/") void subtype(ParameterizedIncrementalCallback> one);
+  }
+
+  static final List listString = null;
+
+  interface ParameterizedIncrementalCallback> extends IncrementalCallback {
+  }
+
+  @Test public void methodCanHaveIncrementalCallbackParam() throws Exception {
+    contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("valid", IncrementalCallback.class));
+  }
+
+  @Test public void methodMetadataReturnTypeOnObservableMethodIsItsTypeParameter() throws Exception {
+    Type listStringType = getClass().getDeclaredField("listString").getGenericType();
+    MethodMetadata md = contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("valid", IncrementalCallback.class));
+    assertEquals(md.decodeInto(), listStringType);
+    md = contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("wildcardExtends", IncrementalCallback.class));
+    assertEquals(md.decodeInto(), listStringType);
+    md = contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("subtype", ParameterizedIncrementalCallback.class));
+    assertEquals(md.decodeInto(), listStringType);
+  }
+
+  @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp = ".*the last parameter.*")
+  public void incrementalCallbackParamMustBeLast() throws Exception {
+    contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("badOrder", IncrementalCallback.class, String.class));
+  }
+
+  @Test(expectedExceptions = IllegalStateException.class, expectedExceptionsMessageRegExp = ".*must return void.*")
+  public void incrementalCallbackMethodMustReturnVoid() throws Exception {
+    contract.parseAndValidatateMetadata(WithIncrementalCallback.class.getDeclaredMethod("returnType", IncrementalCallback.class));
+  }
 }