Adds Call.handleError to address remote resource not found (#1717)

To complete an http span store adapter, we need to be able to coerce
404 to empty. This does that like this:

```java
call.handleError((error, callback) -> {
  if (error instanceof HttpException && ((HttpException) error).code == 404) {
    callback.onSuccess(Collections.emptyList());
  } else {
    callback.onError(error);
  }
});
```

zipkin/src/main/java/zipkin/internal/v2/Call.java

@@ -16,6 +16,8 @@
 import java.io.IOException;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nullable;
 
 /**
  * This captures a (usually remote) request and can be used once, either {@link #execute()
@@ -45,9 +47,18 @@
  * <p>This type owes its design to {@code retrofit2.Call}, which is nearly the same, except limited
  * to HTTP transports.
  *
- * @param <V> the success type, only null when {@code V} is {@linkplain Void}.
+ * @param <V> the success type, typically not null except when {@code V} is {@linkplain Void}.
  */
 public abstract class Call<V> implements Cloneable {
+  /**
+   * Returns a completed call which has the supplied value. This is useful when input parameters
+   * imply there's no call needed. For example, an empty input might always result in an empty
+   * output.
+   */
+  public static <V> Call<V> create(V v) {
+    return new Constant<>(v);
+  }
+
   @SuppressWarnings("unchecked")
   public static <T> Call<List<T>> emptyList() {
     return Call.create(Collections.emptyList());
@@ -98,13 +109,28 @@ public final <R> Call<R> flatMap(FlatMapper<V, R> flatMapper) {
     return new FlatMapping<>(flatMapper, this);
   }
 
+  public interface ErrorHandler<V> {
+    /** Attempts to resolve an error. The user must call the callback. */
+    void onErrorReturn(Throwable error, Callback<V> callback);
+  }
+
   /**
-   * Returns a completed call which has the supplied value. This is useful when input parameters
-   * imply there's no call needed. For example, an empty input might always result in an empty
-   * output.
+   * Returns a call which can attempt to resolve an exception. This is useful when a remote call
+   * returns an error when a resource is not found.
+   *
+   * <p>Here's an example of coercing 404 to empty:
+   * <pre>{@code
+   * call.handleError((error, callback) -> {
+   *   if (error instanceof HttpException && ((HttpException) error).code == 404) {
+   *     callback.onSuccess(Collections.emptyList());
+   *   } else {
+   *     callback.onError(error);
+   *   }
+   * });
+   * }</pre>
    */
-  public static <V> Call<V> create(V v) {
-    return new Constant<>(v);
+  public final Call<V> handleError(ErrorHandler<V> errorHandler) {
+    return new ErrorHandling<>(errorHandler, this);
   }
 
   /**
@@ -246,6 +272,60 @@ static final class FlatMapping<R, V> extends Base<R> {
     }
   }
 
+  static final class ErrorHandling<V> extends Base<V> {
+    static final Object SENTINEL = new Object(); // to differentiate from null
+    final ErrorHandler<V> errorHandler;
+    final Call<V> delegate;
+
+    ErrorHandling(ErrorHandler<V> errorHandler, Call<V> delegate) {
+      this.errorHandler = errorHandler;
+      this.delegate = delegate;
+    }
+
+    @Override public V doExecute() throws IOException {
+      try {
+        return delegate.execute();
+      } catch (IOException | RuntimeException | Error e) {
+        final AtomicReference ref = new AtomicReference(SENTINEL);
+        errorHandler.onErrorReturn(e, new Callback<V>() {
+          @Override public void onSuccess(@Nullable V value) {
+            ref.set(value);
+          }
+
+          @Override public void onError(Throwable t) {
+          }
+        });
+        Object result = ref.get();
+        if (SENTINEL == result) throw e;
+        return (V) result;
+      }
+    }
+
+    @Override public void doEnqueue(final Callback<V> callback) {
+      delegate.enqueue(new Callback<V>() {
+        @Override public void onSuccess(V value) {
+          callback.onSuccess(value);
+        }
+
+        @Override public void onError(Throwable t) {
+          errorHandler.onErrorReturn(t, callback);
+        }
+      });
+    }
+
+    @Override public void cancel() {
+      delegate.cancel();
+    }
+
+    @Override public boolean isCanceled() {
+      return delegate.isCanceled();
+    }
+
+    @Override public Call<V> clone() {
+      return new ErrorHandling<>(errorHandler, delegate.clone());
+    }
+  }
+
   static abstract class Base<V> extends Call<V> {
     volatile boolean canceled;
     boolean executed; // guarded by this

zipkin/src/test/java/zipkin/internal/v2/CallTest.java

@@ -15,7 +15,9 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.NoSuchElementException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -218,19 +220,7 @@ public void concurrent_executesOrSubmitsOnce() throws InterruptedException {
   @Test public void flatMap_enqueue_callException() throws Exception {
     IllegalArgumentException error = new IllegalArgumentException();
     Call<String> fooCall = Call.create("foo");
-    Call<String> exceptionCall = new Call.Base<String>() {
-      @Override String doExecute() throws IOException {
-        throw new AssertionError();
-      }
-
-      @Override void doEnqueue(Callback<String> callback) {
-        callback.onError(error);
-      }
-
-      @Override public Call<String> clone() {
-        throw new AssertionError();
-      }
-    };
+    Call<String> exceptionCall = errorCall(error);
 
     Call<String> fooBarCall = fooCall.flatMap(foo -> exceptionCall);
 
@@ -251,4 +241,100 @@ public void concurrent_executesOrSubmitsOnce() throws InterruptedException {
     assertThat(fooCall.isCanceled()).isTrue();
     assertThat(barCall.isCanceled()).isTrue();
   }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void onErrorReturn_execute_onError() throws Exception {
+    IllegalArgumentException exception = new IllegalArgumentException();
+    Call<String> errorCall = errorCall(exception);
+
+    Call<String> resolvedCall = errorCall.handleError(
+      (error, callback) -> callback.onError(error)
+    );
+
+    resolvedCall.execute();
+  }
+
+  @Test public void onErrorReturn_execute_onSuccess() throws Exception {
+    IllegalArgumentException exception = new IllegalArgumentException();
+    Call<String> errorCall = errorCall(exception);
+
+    Call<String> resolvedCall = errorCall.handleError(
+      (error, callback) -> callback.onSuccess("foo")
+    );
+
+    assertThat(resolvedCall.execute())
+      .isEqualTo("foo");
+  }
+
+  @Test public void onErrorReturn_execute_onSuccess_null() throws Exception {
+    IllegalArgumentException exception = new IllegalArgumentException();
+    Call<String> errorCall = errorCall(exception);
+
+    Call<String> resolvedCall = errorCall.handleError(
+      (error, callback) -> callback.onSuccess(null)
+    );
+
+    assertThat(resolvedCall.execute())
+      .isNull();
+  }
+
+  @Test public void onErrorReturn_enqueue_onError() throws Exception {
+    IllegalArgumentException exception = new IllegalArgumentException();
+    Call<String> errorCall = errorCall(exception);
+
+    Call<String> resolvedCall = errorCall.handleError(
+      (error, callback) -> callback.onError(error)
+    );
+
+    resolvedCall.enqueue(callback);
+
+    verify(callback).onError(exception);
+  }
+
+  @Test public void onErrorReturn_enqueue_onSuccess() throws Exception {
+    IllegalArgumentException exception = new IllegalArgumentException();
+    Call<String> errorCall = errorCall(exception);
+
+    Call<String> resolvedCall = errorCall.handleError(
+      (error, callback) -> callback.onSuccess("foo")
+    );
+
+    resolvedCall.enqueue(callback);
+
+    verify(callback).onSuccess("foo");
+  }
+
+  @Test public void onErrorReturn_enqueue_onSuccess_null() throws Exception {
+    NoSuchElementException exception = new NoSuchElementException();
+    Call<List<String>> call = errorCall(exception);
+
+    Call<List<String>> resolvedCall = call.handleError((error, callback) -> {
+        if (error instanceof NoSuchElementException) {
+          callback.onSuccess(Collections.emptyList());
+        } else {
+          callback.onError(error);
+        }
+      }
+    );
+
+    resolvedCall.enqueue(callback);
+
+    verify(callback).onSuccess(Collections.emptyList());
+  }
+
+  static <T> Call<T> errorCall(RuntimeException error) {
+    return new Call.Base<T>() {
+      @Override T doExecute() throws IOException {
+        throw error;
+      }
+
+      @Override void doEnqueue(Callback<T> callback) {
+        callback.onError(error);
+      }
+
+      @Override public Call<T> clone() {
+        throw new AssertionError();
+      }
+    };
+  }
 }