Added transform and transformAsync overrides that accept an Executor (#38)

hiranya911 · garrettjonesgoogle · commit c2485baf7e91 · 2018-01-17T08:33:08.000-08:00
diff --git a/sdk-platform-java/api-common-java/src/main/java/com/google/api/core/ApiFutures.java b/sdk-platform-java/api-common-java/src/main/java/com/google/api/core/ApiFutures.java
@@ -100,6 +100,17 @@ public static <V, X> ApiFuture<X> transform(
             listenableFutureForApiFuture(input), new GaxFunctionToGuavaFunction<V, X>(function)));
   }
 
+  public static <V, X> ApiFuture<X> transform(
+      ApiFuture<? extends V> input,
+      final ApiFunction<? super V, ? extends X> function,
+      Executor executor) {
+    return new ListenableFutureToApiFuture<>(
+        Futures.transform(
+            listenableFutureForApiFuture(input),
+            new GaxFunctionToGuavaFunction<V, X>(function),
+            executor));
+  }
+
   public static <V> ApiFuture<List<V>> allAsList(
       Iterable<? extends ApiFuture<? extends V>> futures) {
     return new ListenableFutureToApiFuture<>(
@@ -128,6 +139,22 @@ public ListenableFuture<O> apply(I input) throws Exception {
     return new ListenableFutureToApiFuture<>(listenableOutput);
   }
 
+  public static <I, O> ApiFuture<O> transformAsync(
+      ApiFuture<I> input, final ApiAsyncFunction<I, O> function, Executor executor) {
+    ListenableFuture<I> listenableInput = listenableFutureForApiFuture(input);
+    ListenableFuture<O> listenableOutput =
+        Futures.transformAsync(
+            listenableInput,
+            new AsyncFunction<I, O>() {
+              @Override
+              public ListenableFuture<O> apply(I input) throws Exception {
+                return listenableFutureForApiFuture(function.apply(input));
+              }
+            },
+            executor);
+    return new ListenableFutureToApiFuture<>(listenableOutput);
+  }
+
   private static <V> ListenableFuture<V> listenableFutureForApiFuture(ApiFuture<V> apiFuture) {
     ListenableFuture<V> listenableFuture;
     if (apiFuture instanceof AbstractApiFuture) {
diff --git a/sdk-platform-java/api-common-java/src/test/java/com/google/api/core/ApiFuturesTest.java b/sdk-platform-java/api-common-java/src/test/java/com/google/api/core/ApiFuturesTest.java
@@ -36,6 +36,7 @@
 import java.util.List;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 
@@ -95,6 +96,31 @@ public String apply(Integer input) {
     assertThat(transformedFuture.get()).isEqualTo("6");
   }
 
+  @Test
+  public void testTransformWithExecutor() throws Exception {
+    SettableApiFuture<Integer> inputFuture = SettableApiFuture.<Integer>create();
+    final AtomicInteger counter = new AtomicInteger(0);
+    ApiFuture<String> transformedFuture =
+        ApiFutures.transform(
+            inputFuture,
+            new ApiFunction<Integer, String>() {
+              @Override
+              public String apply(Integer input) {
+                return input.toString();
+              }
+            },
+            new Executor() {
+              @Override
+              public void execute(Runnable command) {
+                counter.incrementAndGet();
+                command.run();
+              }
+            });
+    inputFuture.set(6);
+    assertThat(transformedFuture.get()).isEqualTo("6");
+    assertThat(counter.get()).isEqualTo(1);
+  }
+
   @Test
   public void testAllAsList() throws Exception {
     SettableApiFuture<Integer> inputFuture1 = SettableApiFuture.<Integer>create();
@@ -121,6 +147,30 @@ public ApiFuture<Integer> apply(Integer input) {
     assertThat(outputFuture.get()).isEqualTo(1);
   }
 
+  @Test
+  public void testTransformAsyncWithExecutor() throws Exception {
+    ApiFuture<Integer> inputFuture = ApiFutures.immediateFuture(0);
+    final AtomicInteger counter = new AtomicInteger(0);
+    ApiFuture<Integer> outputFuture =
+        ApiFutures.transformAsync(
+            inputFuture,
+            new ApiAsyncFunction<Integer, Integer>() {
+              @Override
+              public ApiFuture<Integer> apply(Integer input) {
+                return ApiFutures.immediateFuture(input + 1);
+              }
+            },
+            new Executor() {
+              @Override
+              public void execute(Runnable command) {
+                counter.incrementAndGet();
+                command.run();
+              }
+            });
+    assertThat(outputFuture.get()).isEqualTo(1);
+    assertThat(counter.get()).isEqualTo(1);
+  }
+
   @Test
   public void testImmediateFailedFuture() throws InterruptedException {
     ApiFuture<String> future =
