feat: add @BetaApi ApiFutures.catchingAsync (#117)

Add new ApiFuture.catchingAsync method mirroring ApiFuture.catching, except allowing for an ApiAsyncFunction to be passed instead of ApiFunction.

I tried to match the existing signature that allows the callback to return '? extends V' but this does not seem to be possible (see transformAsync).

src/main/java/com/google/api/core/ApiFutures.java

@@ -104,6 +104,27 @@ public static <V, X extends Throwable> ApiFuture<V> catching(
     return new ListenableFutureToApiFuture<V>(catchingFuture);
   }
 
+  @BetaApi
+  public static <V, X extends Throwable> ApiFuture<V> catchingAsync(
+      ApiFuture<V> input,
+      Class<X> exceptionType,
+      final ApiAsyncFunction<? super X, V> callback,
+      Executor executor) {
+    ListenableFuture<V> catchingFuture =
+        Futures.catchingAsync(
+            listenableFutureForApiFuture(input),
+            exceptionType,
+            new AsyncFunction<X, V>() {
+              @Override
+              public ListenableFuture<V> apply(X exception) throws Exception {
+                ApiFuture<V> result = callback.apply(exception);
+                return listenableFutureForApiFuture(result);
+              }
+            },
+            executor);
+    return new ListenableFutureToApiFuture<>(catchingFuture);
+  }
+
   public static <V> ApiFuture<V> immediateFuture(V value) {
     return new ListenableFutureToApiFuture<>(Futures.<V>immediateFuture(value));
   }

src/test/java/com/google/api/core/ApiFuturesTest.java

@@ -83,6 +83,24 @@ public Integer apply(Exception ex) {
     assertThat(fallback.get()).isEqualTo(42);
   }
 
+  @Test
+  public void testCatchAsync() throws Exception {
+    SettableApiFuture<Integer> future = SettableApiFuture.<Integer>create();
+    ApiFuture<Integer> fallback =
+        ApiFutures.catchingAsync(
+            future,
+            Exception.class,
+            new ApiAsyncFunction<Exception, Integer>() {
+              @Override
+              public ApiFuture<Integer> apply(Exception ex) {
+                return ApiFutures.immediateFuture(42);
+              }
+            },
+            directExecutor());
+    future.setException(new Exception());
+    assertThat(fallback.get()).isEqualTo(42);
+  }
+
   @Test
   public void testTransform() throws Exception {
     SettableApiFuture<Integer> inputFuture = SettableApiFuture.<Integer>create();