Add interop between Guava's ListenableFuture and Parseq Task (#306)

Co-authored-by: Karthik Ramgopal <kramgopa@linkedin.com>

Author: karthikrg

CHANGELOG.md

@@ -1,3 +1,7 @@
+v5.1.4
+------
+* Add interop between Guava's ListenableFuture and Parseq Task
+
 v5.1.3
 ------
 * Fix for multiple javadoc warnings coming from Task.java

gradle.properties

@@ -1,3 +1,3 @@
-version=5.1.3
+version=5.1.4
 group=com.linkedin.parseq
 org.gradle.parallel=true

settings.gradle

@@ -7,6 +7,7 @@ def modules = [ /* the name of the modules to use */
     'parseq-benchmark',
     'parseq-examples',
     'parseq-exec',
+    'parseq-guava-interop',
     'parseq-http-client',
     'parseq-lambda-names', // shadow jar
     'parseq-legacy-examples',

subprojects/parseq-guava-interop/build.gradle

@@ -0,0 +1,10 @@
+ext {
+  description = """Interop with Guava's ListenableFuture"""
+}
+
+dependencies {
+  compile project(":parseq")
+  compile "com.google.guava:guava:30.1.1-jre"
+
+  testCompile "org.testng:testng:6.9.9"
+}

subprojects/parseq-guava-interop/src/main/java/com/linkedin/parseq/guava/ListenableFutureUtil.java

@@ -0,0 +1,133 @@
+package com.linkedin.parseq.guava;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.AbstractFuture;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.MoreExecutors;
+import com.linkedin.parseq.BaseTask;
+import com.linkedin.parseq.Context;
+import com.linkedin.parseq.Task;
+import com.linkedin.parseq.promise.Promise;
+import com.linkedin.parseq.promise.SettablePromise;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+
+
+/**
+ * Utility methods to convert between Parseq {@link Task} and Guava's {@link ListenableFuture}.
+ */
+public class ListenableFutureUtil {
+
+  private ListenableFutureUtil() {
+    // Prevent instantiation.
+  }
+
+  public static <T> Task<T> fromListenableFuture(ListenableFuture<T> future) {
+    // Setup cancellation propagation from Task -> ListenableFuture.
+    final SettableTask<T> task =
+        new SettableTask<T>("fromListenableFuture: " + Task._taskDescriptor.getDescription(future.getClass().getName())) {
+          @Override
+          public boolean cancel(Exception rootReason) {
+            if (future.isCancelled()) {
+              return super.cancel(rootReason);
+            }
+
+            return super.cancel(rootReason) && future.cancel(true);
+          }
+        };
+
+    final SettablePromise<T> promise = task.getSettableDelegate();
+
+    // Setup forward event propagation ListenableFuture -> Task.
+    Runnable callbackRunnable = () -> {
+      try {
+        final T value = future.get();
+        promise.done(value);
+      } catch (CancellationException ex) {
+        task.cancel(ex);
+      } catch (ExecutionException ex) {
+        promise.fail(ex.getCause());
+      } catch (Exception | Error ex) {
+        promise.fail(ex);
+      }
+    };
+    future.addListener(callbackRunnable, MoreExecutors.directExecutor());
+
+    return task;
+  }
+
+  public static <T> ListenableFuture<T> toListenableFuture(Task<T> task) {
+    // Setup cancellation propagation from ListenableFuture -> Task.
+    SettableFuture<T> listenableFuture = new SettableFuture<T>() {
+      @Override
+      public boolean cancel(boolean mayInterruptIfRunning) {
+        return super.cancel(mayInterruptIfRunning) && task.cancel(new CancellationException());
+      }
+
+      @Override
+      public boolean setException(Throwable ex) {
+        if (!task.isDone() && ex instanceof CancellationException) {
+          task.cancel((CancellationException) ex);
+        }
+        return super.setException(ex);
+      }
+    };
+
+    // Setup forward event propagation Task -> ListenableFuture.
+    task.addListener(promise -> {
+      if (!promise.isFailed()) {
+        listenableFuture.set(promise.get());
+      }
+      else {
+        if (promise.getError() instanceof com.linkedin.parseq.CancellationException) {
+          listenableFuture.cancel(true);
+        } else {
+          listenableFuture.setException(promise.getError());
+        }
+      }
+    });
+
+    return listenableFuture;
+  }
+
+  /**
+   * A private helper class to assist toListenableFuture(), by overriding some methods to make them public.
+   *
+   * @param <T> The Settable future's type.
+   */
+  @VisibleForTesting
+  static class SettableFuture<T> extends AbstractFuture<T> {
+    @Override
+    public boolean set(T value) {
+      return super.set(value);
+    }
+
+    @Override
+    public boolean setException(Throwable throwable) {
+      return super.setException(throwable);
+    }
+  }
+
+  /**
+   * A private helper class to assist fromListenableFuture(), by overriding some methods to make them public.
+   *
+   * @param <T> The Settable task's type.
+   */
+  @VisibleForTesting
+  static class SettableTask<T> extends BaseTask<T> {
+
+    public SettableTask(String name) {
+      super(name);
+    }
+
+    @Override
+    protected Promise<? extends T> run(Context context) throws Throwable {
+      return getDelegate();
+    }
+
+    @Override
+    public SettablePromise<T> getSettableDelegate() {
+      return super.getSettableDelegate();
+    }
+  }
+}

subprojects/parseq-guava-interop/src/test/java/com/linkedin/parseq/guava/ListenableFutureUtilTest.java

@@ -0,0 +1,103 @@
+package com.linkedin.parseq.guava;
+
+import com.google.common.util.concurrent.ListenableFuture;
+import com.linkedin.parseq.Task;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+
+/**
+ * Unit tests for {@link ListenableFutureUtil}
+ */
+public class ListenableFutureUtilTest {
+
+  @Test
+  public void testFromListenableFuture() {
+    ListenableFutureUtil.SettableFuture<String> listenableFuture = new ListenableFutureUtil.SettableFuture<>();
+    Task<String> task = ListenableFutureUtil.fromListenableFuture(listenableFuture);
+
+    // Test cancel propagation from Task to ListenableFuture
+    task.cancel(new RuntimeException());
+    Assert.assertTrue(listenableFuture.isCancelled());
+
+    listenableFuture = new ListenableFutureUtil.SettableFuture<>();
+    task = ListenableFutureUtil.fromListenableFuture(listenableFuture);
+
+    // Test successful completion of ListenableFuture.
+    listenableFuture.set("COMPLETED");
+    Assert.assertTrue(task.isDone());
+    Assert.assertFalse(task.isFailed());
+    Assert.assertEquals(task.get(), "COMPLETED");
+
+    listenableFuture = new ListenableFutureUtil.SettableFuture<>();
+    task = ListenableFutureUtil.fromListenableFuture(listenableFuture);
+
+    // Test exceptional completion of ListenableFuture.
+    listenableFuture.setException(new RuntimeException("Test"));
+    Assert.assertTrue(task.isDone());
+    Assert.assertTrue(task.isFailed());
+    Assert.assertEquals(task.getError().getClass(), RuntimeException.class);
+    Assert.assertEquals(task.getError().getMessage(), "Test");
+
+    listenableFuture = new ListenableFutureUtil.SettableFuture<>();
+    task = ListenableFutureUtil.fromListenableFuture(listenableFuture);
+
+    // Test cancellation of ListenableFuture.
+    listenableFuture.cancel(true);
+    Assert.assertTrue(task.isDone());
+    Assert.assertTrue(task.isFailed());
+    Assert.assertEquals(task.getError().getCause().getClass(), CancellationException.class);
+  }
+
+  @Test
+  public void testToListenableFuture() throws Exception {
+    ListenableFutureUtil.SettableTask<String> task = new ListenableFutureUtil.SettableTask<>("test");
+    ListenableFuture<String> future = ListenableFutureUtil.toListenableFuture(task);
+
+    // Test cancel propagation from ListenableFuture to task
+    future.cancel(true);
+    Assert.assertTrue(task.isDone());
+    Assert.assertTrue(task.isFailed());
+    Assert.assertEquals(task.getError().getCause().getClass(), CancellationException.class);
+
+    task = new ListenableFutureUtil.SettableTask<>("test");
+    future = ListenableFutureUtil.toListenableFuture(task);
+
+    // Test successful completion of task.
+    task.getSettableDelegate().done("COMPLETED");
+    Assert.assertTrue(future.isDone());
+    Assert.assertEquals(future.get(), "COMPLETED");
+
+    task = new ListenableFutureUtil.SettableTask<>("test");
+    future = ListenableFutureUtil.toListenableFuture(task);
+
+    // Test exceptional completion of task.
+    task.getSettableDelegate().fail(new RuntimeException("Test"));
+    Assert.assertTrue(future.isDone());
+    try {
+      future.get();
+      Assert.fail("ExecutionException not thrown");
+    } catch (ExecutionException e) {
+      Assert.assertEquals(e.getCause().getClass(), RuntimeException.class);
+      Assert.assertEquals(e.getCause().getMessage(), "Test");
+    }
+
+    task = new ListenableFutureUtil.SettableTask<>("test");
+    future = ListenableFutureUtil.toListenableFuture(task);
+
+    // Test cancellation of task.
+    task.cancel(new RuntimeException("Cancelled"));
+    Assert.assertTrue(future.isDone());
+    Assert.assertTrue(future.isCancelled());
+    try {
+      future.get();
+      Assert.fail("Cancellation Exception not thrown");
+    } catch (CancellationException e) {
+      // Ignored since we expected a cancellation exception!
+    } catch (Throwable e) {
+      Assert.fail("Unexpected Exception thrown", e);
+    }
+  }
+}

subprojects/parseq/src/main/java/com/linkedin/parseq/CancellationException.java

@@ -1,6 +1,6 @@
 package com.linkedin.parseq;
 
-class CancellationException extends Exception {
+public class CancellationException extends Exception {
 
   private static final long serialVersionUID = 1L;