CancellationScope improvements and documentation (#377)

Added newCancellationScope with function parameter and updated CancellationScope documentation

Author: mfateev

src/main/java/com/uber/cadence/internal/sync/CancellationScopeImpl.java

@@ -19,6 +19,7 @@
 
 import com.uber.cadence.workflow.CancellationScope;
 import com.uber.cadence.workflow.CompletablePromise;
+import com.uber.cadence.workflow.Functions;
 import com.uber.cadence.workflow.Workflow;
 import java.util.ArrayDeque;
 import java.util.Deque;
@@ -73,6 +74,13 @@ private static void popCurrent(CancellationScopeImpl expected) {
     setParent(parent);
   }
 
+  public CancellationScopeImpl(
+      boolean ignoreParentCancellation, Functions.Proc1<CancellationScope> proc) {
+    this.detached = ignoreParentCancellation;
+    this.runnable = () -> proc.apply(this);
+    setParent(current());
+  }
+
   private void setParent(CancellationScopeImpl parent) {
     if (parent == null) {
       detached = true;

src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java

@@ -296,6 +296,11 @@ public static CancellationScope newCancellationScope(boolean detached, Runnable
     return new CancellationScopeImpl(detached, runnable);
   }
 
+  public static CancellationScope newCancellationScope(
+      boolean detached, Functions.Proc1<CancellationScope> proc) {
+    return new CancellationScopeImpl(detached, proc);
+  }
+
   public static CancellationScopeImpl currentCancellationScope() {
     return CancellationScopeImpl.current();
   }

src/main/java/com/uber/cadence/workflow/CancellationScope.java

@@ -23,7 +23,8 @@
 /**
  * Handle to a cancellation scope created through {@link Workflow#newCancellationScope(Runnable)} or
  * {@link Workflow#newDetachedCancellationScope(Runnable)}. Supports explicit cancelling of the code
- * a cancellation scope wraps.
+ * a cancellation scope wraps. The code in the CancellationScope has to be executed using {@link
+ * Runnable#run()} method.
  */
 public interface CancellationScope extends Runnable {
 

src/main/java/com/uber/cadence/workflow/Workflow.java

@@ -225,7 +225,8 @@
  *                         activities.deleteLocalFile(processedName);
  *                     }
  *                 }
- *             });
+ *             }
+ *          ).run();
  *     }
  * }
  * }</pre>
@@ -581,6 +582,21 @@ public static WorkflowInfo getWorkflowInfo() {
    * blocking after the current scope is cancelled use a scope created through {@link
    * #newDetachedCancellationScope(Runnable)}.
    *
+   * <p>Example of running activities in parallel and cancelling them after a specified timeout.
+   *
+   * <pre><code>
+   *     List<Promise<String>> results = new ArrayList<>();
+   *     CancellationScope scope = Workflow.newDetachedCancellationScope(() -> {
+   *        Async.function(activities::a1);
+   *        Async.function(activities::a2);
+   *     });
+   *     scope.run(); // returns immediately as the activities are invoked asynchronously
+   *     Workflow.sleep(Duration.ofHours(1));
+   *     // Cancels any activity in the scope that is still running
+   *     scope.cancel("one hour passed");
+   *
+   * </code></pre>
+   *
    * @param runnable parameter to wrap in a cancellation scope.
    * @return wrapped parameter.
    */
@@ -589,7 +605,52 @@ public static CancellationScope newCancellationScope(Runnable runnable) {
   }
 
   /**
-   * Creates a CancellationScope that is not linked to a parent scope.
+   * Wraps a procedure in a CancellationScope. The procedure receives the wrapping CancellationScope
+   * as a parameter. Useful when cancellation is requested from within the wrapped code. The
+   * following example cancels the sibling activity on any failure.
+   *
+   * <pre><code>
+   *               Workflow.newCancellationScope(
+   *                   (scope) -> {
+   *                     Promise<Void> p1 = Async.proc(activities::a1).exceptionally(ex->
+   *                        {
+   *                           scope.cancel("a1 failed");
+   *                           return null;
+   *                        });
+   *
+   *                     Promise<Void> p2 = Async.proc(activities::a2).exceptionally(ex->
+   *                        {
+   *                           scope.cancel("a2 failed");
+   *                           return null;
+   *                        });
+   *                     Promise.allOf(p1, p2).get();
+   *                   })
+   *               .run();
+   * </code></pre>
+   *
+   * @param proc code to wrap in the cancellation scope
+   * @return wrapped proc
+   */
+  public static CancellationScope newCancellationScope(Functions.Proc1<CancellationScope> proc) {
+    return WorkflowInternal.newCancellationScope(false, proc);
+  }
+
+  /**
+   * Creates a CancellationScope that is not linked to a parent scope. {@link
+   * CancellationScope#run()} must be called to execute the code the scope wraps. The detached scope
+   * is needed to execute cleanup code after a workflow is cancelled which cancels the root scope
+   * that wraps the @WorkflowMethod invocation. Here is an example usage:
+   *
+   * <pre><code>
+   *  try {
+   *     // workflow logic
+   *  } catch (CancellationException e) {
+   *     CancellationScope detached = Workflow.newDetachedCancellationScope(() -> {
+   *         // cleanup logic
+   *     });
+   *     detached.run();
+   *  }
+   * </code></pre>
    *
    * @param runnable parameter to wrap in a cancellation scope.
    * @return wrapped parameter.

src/test/java/com/uber/cadence/internal/sync/DeterministicRunnerTest.java

@@ -468,6 +468,64 @@ public void testExplicitThreadCancellation() throws Throwable {
     trace.setExpected(expected);
   }
 
+  @Test
+  public void testExplicitCancellationOnFailure() throws Throwable {
+    trace.add("init");
+    DeterministicRunner d =
+        new DeterministicRunnerImpl(
+            () -> {
+              trace.add("root started");
+              Workflow.newCancellationScope(
+                      (scope) -> {
+                        Promise<Void> p1 =
+                            Async.procedure(
+                                () -> {
+                                  trace.add("thread1 started");
+                                  try {
+                                    Workflow.sleep(Duration.ofHours(1));
+                                  } catch (Exception e) {
+                                    trace.add("thread1 exception: " + e.getClass().getSimpleName());
+                                  } finally {
+                                    trace.add("thread1 done");
+                                  }
+                                });
+                        Promise<Void> p2 =
+                            Async.procedure(
+                                    () -> {
+                                      trace.add("thread2 started");
+                                      try {
+                                        throw new RuntimeException("simulated");
+                                      } finally {
+                                        trace.add("thread2 done");
+                                      }
+                                    })
+                                .exceptionally(
+                                    ex -> {
+                                      scope.cancel();
+                                      return null;
+                                    });
+                        Promise.allOf(p1, p2).get();
+                      })
+                  .run();
+              trace.add("root done");
+            });
+
+    d.runUntilAllBlocked();
+    assertTrue(d.stackTrace(), d.isDone());
+    String[] expected =
+        new String[] {
+          "init",
+          "root started",
+          "thread1 started",
+          "thread2 started",
+          "thread2 done",
+          "thread1 exception: CancellationException",
+          "thread1 done",
+          "root done"
+        };
+    trace.setExpected(expected);
+  }
+
   @Test
   public void testDetachedCancellation() throws Throwable {
     trace.add("init");