diff --git a/src/main/java/com/google/api/core/ApiFutures.java b/src/main/java/com/google/api/core/ApiFutures.java index c9caee980..115cdf7a9 100644 --- a/src/main/java/com/google/api/core/ApiFutures.java +++ b/src/main/java/com/google/api/core/ApiFutures.java @@ -100,6 +100,17 @@ public static ApiFuture transform( listenableFutureForApiFuture(input), new GaxFunctionToGuavaFunction(function))); } + public static ApiFuture transform( + ApiFuture input, + final ApiFunction function, + Executor executor) { + return new ListenableFutureToApiFuture<>( + Futures.transform( + listenableFutureForApiFuture(input), + new GaxFunctionToGuavaFunction(function), + executor)); + } + public static ApiFuture> allAsList( Iterable> futures) { return new ListenableFutureToApiFuture<>( @@ -128,6 +139,22 @@ public ListenableFuture apply(I input) throws Exception { return new ListenableFutureToApiFuture<>(listenableOutput); } + public static ApiFuture transformAsync( + ApiFuture input, final ApiAsyncFunction function, Executor executor) { + ListenableFuture listenableInput = listenableFutureForApiFuture(input); + ListenableFuture listenableOutput = + Futures.transformAsync( + listenableInput, + new AsyncFunction() { + @Override + public ListenableFuture apply(I input) throws Exception { + return listenableFutureForApiFuture(function.apply(input)); + } + }, + executor); + return new ListenableFutureToApiFuture<>(listenableOutput); + } + private static ListenableFuture listenableFutureForApiFuture(ApiFuture apiFuture) { ListenableFuture listenableFuture; if (apiFuture instanceof AbstractApiFuture) { diff --git a/src/test/java/com/google/api/core/ApiFuturesTest.java b/src/test/java/com/google/api/core/ApiFuturesTest.java index ab134f6f3..c97ff694a 100644 --- a/src/test/java/com/google/api/core/ApiFuturesTest.java +++ b/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 inputFuture = SettableApiFuture.create(); + final AtomicInteger counter = new AtomicInteger(0); + ApiFuture transformedFuture = + ApiFutures.transform( + inputFuture, + new ApiFunction() { + @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 inputFuture1 = SettableApiFuture.create(); @@ -121,6 +147,30 @@ public ApiFuture apply(Integer input) { assertThat(outputFuture.get()).isEqualTo(1); } + @Test + public void testTransformAsyncWithExecutor() throws Exception { + ApiFuture inputFuture = ApiFutures.immediateFuture(0); + final AtomicInteger counter = new AtomicInteger(0); + ApiFuture outputFuture = + ApiFutures.transformAsync( + inputFuture, + new ApiAsyncFunction() { + @Override + public ApiFuture 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 future =