Exposing CancellationScope.run method (#261)

* Added RunnableCancellationScope

* Got rid of RunnableCancellationScope

* Exposed run method on CancellationScope

Author: mfateev

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

@@ -87,7 +87,8 @@ private void setParent(CancellationScopeImpl parent) {
     }
   }
 
-  void run() {
+  @Override
+  public void run() {
     try {
       pushCurrent(this);
       runnable.run();

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

@@ -279,9 +279,7 @@ public static Promise<Object> promiseAnyOf(Promise<?>... promises) {
   }
 
   public static CancellationScope newCancellationScope(boolean detached, Runnable runnable) {
-    CancellationScopeImpl result = new CancellationScopeImpl(detached, runnable);
-    result.run();
-    return result;
+    return new CancellationScopeImpl(detached, runnable);
   }
 
   public static CancellationScopeImpl currentCancellationScope() {

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

@@ -189,6 +189,11 @@ public void setName(String name) {
             runnable);
   }
 
+  @Override
+  public void run() {
+    throw new UnsupportedOperationException("not used");
+  }
+
   @Override
   public boolean isDetached() {
     return task.cancellationScope.isDetached();

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

@@ -25,7 +25,7 @@
  * {@link Workflow#newDetachedCancellationScope(Runnable)}. Supports explicit cancelling of the code
  * a cancellation scope wraps.
  */
-public interface CancellationScope {
+public interface CancellationScope extends Runnable {
 
   /**
    * When set to false parent thread cancellation causes this one to get cancelled automatically.

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

@@ -537,10 +537,38 @@ public static WorkflowInfo getWorkflowInfo() {
     return WorkflowInternal.getWorkflowInfo();
   }
 
-  public static <R> CancellationScope newCancellationScope(Runnable runnable) {
+  /**
+   * Wraps the Runnable method argument in a {@link CancellationScope}. The {@link
+   * CancellationScope#run()} calls {@link Runnable#run()} on the wrapped Runnable. The returned
+   * CancellationScope can be used to cancel the wrapped code. The cancellation semantic depends on
+   * the operation the code is blocked on. For example activity or child workflow is first cancelled
+   * then throws a {@link CancellationException}. The same applies for {@link Workflow#sleep(long)}
+   * operation. When an activity or a child workflow is invoked asynchronously then they get
+   * cancelled and a {@link Promise} that contains their result will throw CancellationException
+   * when {@link Promise#get()} is called.
+   *
+   * <p>The new cancellation scope is linked to the parent one (available as {@link
+   * CancellationScope#current()}. If the parent one is cancelled then all the children scopes are
+   * cancelled automatically. The main workflow function (annotated with @{@link WorkflowMethod} is
+   * wrapped within a root cancellation scope which gets cancelled when a workflow is cancelled
+   * through the Cadence CancelWorkflowExecution API. To perform cleanup operations that require
+   * blocking after the current scope is cancelled use a scope created through {@link
+   * #newDetachedCancellationScope(Runnable)}.
+   *
+   * @param runnable parameter to wrap in a cancellation scope.
+   * @return wrapped parameter.
+   */
+  public static CancellationScope newCancellationScope(Runnable runnable) {
     return WorkflowInternal.newCancellationScope(false, runnable);
   }
 
+  /**
+   * Creates a CancellationScope that is not linked to a parent scope.
+   *
+   * @param runnable parameter to wrap in a cancellation scope.
+   * @return wrapped parameter.
+   * @see #newCancellationScope(Runnable)
+   */
   public static CancellationScope newDetachedCancellationScope(Runnable runnable) {
     return WorkflowInternal.newCancellationScope(true, runnable);
   }

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

@@ -344,6 +344,7 @@ public void testExplicitScopeCancellation() throws Throwable {
                         var.completeFrom(newTimer(300));
                         trace.add("scope done");
                       });
+              scope.run();
               trace.add("root before cancel");
               scope.cancel("from root");
               try {
@@ -385,6 +386,7 @@ public void testExplicitDetachedScopeCancellation() throws Throwable {
                         var.completeFrom(newTimer(300));
                         trace.add("scope done");
                       });
+              scope.run();
               trace.add("root before cancel");
               scope.cancel("from root");
               try {
@@ -446,6 +448,7 @@ public void testExplicitThreadCancellation() throws Throwable {
                               trace.add("thread done: " + cancellation.get());
                             });
                       });
+              scope.run();
               trace.add("root before cancel");
               scope.cancel("from root");
               threadDone.get();
@@ -475,21 +478,23 @@ public void testDetachedCancellation() throws Throwable {
               trace.add("root started");
               CompletablePromise<Void> done = Workflow.newPromise();
               Workflow.newDetachedCancellationScope(
-                  () -> {
-                    Async.procedure(
-                        () -> {
-                          trace.add("thread started");
-                          WorkflowThread.await(
-                              "reason1",
-                              () -> unblock1 || CancellationScope.current().isCancelRequested());
-                          if (CancellationScope.current().isCancelRequested()) {
-                            done.completeExceptionally(new CancellationException());
-                          } else {
-                            done.complete(null);
-                          }
-                          trace.add("await done");
-                        });
-                  });
+                      () -> {
+                        Async.procedure(
+                            () -> {
+                              trace.add("thread started");
+                              WorkflowThread.await(
+                                  "reason1",
+                                  () ->
+                                      unblock1 || CancellationScope.current().isCancelRequested());
+                              if (CancellationScope.current().isCancelRequested()) {
+                                done.completeExceptionally(new CancellationException());
+                              } else {
+                                done.complete(null);
+                              }
+                              trace.add("await done");
+                            });
+                      })
+                  .run();
               try {
                 done.get();
               } catch (CancellationException e) {

src/test/java/com/uber/cadence/workflow/WorkflowTest.java

@@ -867,21 +867,26 @@ public String execute(String taskList) {
         testActivities.activityWithDelay(100000, true);
         fail("unreachable");
       } catch (CancellationException e) {
-        Workflow.newDetachedCancellationScope(() -> assertEquals(1, testActivities.activity1(1)));
+        Workflow.newDetachedCancellationScope(() -> assertEquals(1, testActivities.activity1(1)))
+            .run();
       }
       try {
         Workflow.sleep(Duration.ofHours(1));
         fail("unreachable");
       } catch (CancellationException e) {
         Workflow.newDetachedCancellationScope(
-            () -> assertEquals("a12", testActivities.activity2("a1", 2)));
+                () -> assertEquals("a12", testActivities.activity2("a1", 2)))
+            .run();
+        ;
       }
       try {
         Workflow.newTimer(Duration.ofHours(1)).get();
         fail("unreachable");
       } catch (CancellationException e) {
         Workflow.newDetachedCancellationScope(
-            () -> assertEquals("a123", testActivities.activity3("a1", 2, 3)));
+                () -> assertEquals("a123", testActivities.activity3("a1", 2, 3)))
+            .run();
+        ;
       }
       return "result";
     }
@@ -2680,6 +2685,7 @@ public String execute(String taskList) {
       CancellationScope scope =
           Workflow.newCancellationScope(
               () -> signal.completeFrom(Async.procedure(workflow::signal1, "World")));
+      scope.run();
       scope.cancel();
       try {
         signal.get();