OpenFeign · codefromthecrypt · Jul 16, 2013 · Jul 6, 2013 · Jul 12, 2013 · Jul 12, 2013
diff --git a/CHANGES.md b/CHANGES.md
@@ -1,4 +1,5 @@
 ### Version 3.0
+* Added support for asynchronous callbacks via `IncrementalCallback<T>` and `IncrementalDecoder.TextStream<T>`.
 * 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<T>`

diff --git a/README.md b/README.md
@@ -68,6 +68,59 @@ The generic parameter of `Decoder.TextStream<T>` designates which The type param
   return new SAXDecoder<ZoneList>(handlers){};
 }
 ```
+### Asynchronous Incremental Callbacks
+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.
+
+Here's how one looks:
+```java
+IncrementalCallback<Contributor> printlnObserver = new IncrementalCallback<Contributor>() {
+
+  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<T>`, you'll need to configure an `IncrementalDecoderi.TextStream<T>` or a general one for all types (`IncrementalDecoder.TextStream<Object>`).
+
+Here's how to wire in a reflective incremental json decoder:
+```java
+@Provides(type = SET) IncrementalDecoder incrementalDecoder(final Gson gson) {
+  return new IncrementalDecoder.TextStream<Object>() {
+
+    @Override
+    public void decode(Reader reader, Type type, IncrementalCallback<? super Object> 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<T>` (default `HardCodedTarget<T>`), 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
@@ -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<MethodMetadata> 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<X> or IncrementalCallback<? super X> 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
@@ -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<Decoder> noDecoders() {
       return Collections.emptySet();
     }
+
+    @Provides Set<IncrementalDecoder> 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<Object> modulesForGraph(Object... modules) {
     return modulesForGraph;
   }
 
-  Feign() {
+  private final Lazy<Executor> httpExecutor;
 
+  Feign(Lazy<Executor> 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
@@ -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.
+ * <br>
+ * {@link #onSuccess() onSuccess} or {@link #onFailure(Throwable)} onFailure}
+ * will be called when the response is finished, but not both.
+ * <br>
+ * {@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.
+ * <br>
+ * <br>
+ * Here's an example of implementing {@code IncrementalCallback}:
+ * <br>
+ * <pre>
+ * IncrementalCallback<Contributor> counter = new IncrementalCallback<Contributor>() {
+ *
+ *   public int count;
+ *
+ *   &#064;Override public void onNext(Contributor element) {
+ *     count++;
+ *   }
+ *
+ *   &#064;Override public void onSuccess() {
+ *     System.out.println("found " + count + " contributors");
+ *   }
+ *
+ *   &#064;Override public void onFailure(Throwable cause) {
+ *     System.err.println("sad face after contributor " + count);
+ *   }
+ * };
+ * github.contributors("netflix", "feign", counter);
+ * </pre>
+ *
+ * @param <T> expected value to decode
+ */
+public interface IncrementalCallback<T> {
+  /**
+   * 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.
+   * <br>
+   * 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);
+}