Added WorkflowLocal and WorkflowThreadLocal (#135)

* Added WorkflowLocal and WorkflowThreadLocal

* Implemented unit test for WorkflowLocal and WorkflowThreadLocal

Author: mfateev

src/main/java/com/uber/cadence/client/WorkflowException.java

@@ -40,14 +40,22 @@ protected WorkflowException(
   private static String getMessage(
       String message, WorkflowExecution execution, Optional<String> workflowType) {
     StringBuilder result = new StringBuilder();
-    result.append(message);
+    if (message != null) {
+      result.append(message);
+      result.append(", ");
+    }
     if (workflowType.isPresent()) {
-      result.append(", WorkflowType=\"");
+      result.append("WorkflowType=\"");
       result.append(workflowType.get());
+    }
+    if (execution != null) {
+      if (result.length() > 0) {
+        result.append("\", ");
+      }
+      result.append("WorkflowExecution=\"");
+      result.append(execution);
       result.append("\"");
     }
-    result.append(", WorkflowExecution=\"");
-    result.append(execution);
     return result.toString();
   }
 

src/main/java/com/uber/cadence/client/WorkflowServiceException.java

@@ -18,26 +18,13 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import com.uber.cadence.internal.common.CheckedExceptionWrapper;
 import java.util.Optional;
 
 public final class WorkflowServiceException extends WorkflowException {
 
   public WorkflowServiceException(
       WorkflowExecution execution, Optional<String> workflowType, Throwable failure) {
-    super(getMessage(execution, workflowType), execution, workflowType, failure);
-  }
-
-  private static String getMessage(WorkflowExecution execution, Optional<String> workflowType) {
-    StringBuilder result = new StringBuilder();
-    if (workflowType.isPresent()) {
-      result.append("WorkflowType=\"");
-      result.append(workflowType.get());
-      result.append("\", ");
-    }
-    result.append("WorkflowID=\"");
-    result.append(execution.getWorkflowId());
-    result.append("\", RunID=\"");
-    result.append(execution.getRunId());
-    return result.toString();
+    super(null, execution, workflowType, CheckedExceptionWrapper.unwrap(failure));
   }
 }

src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java

@@ -93,7 +93,7 @@ public WorkflowExecution startWorkflow(StartWorkflowExecutionParameters startPar
     } catch (WorkflowExecutionAlreadyStartedError e) {
       throw e;
     } catch (TException e) {
-      throw new RuntimeException(e);
+      throw CheckedExceptionWrapper.wrap(e);
     }
     WorkflowExecution execution = new WorkflowExecution();
     execution.setRunId(result.getRunId());
@@ -116,7 +116,7 @@ public void signalWorkflowExecution(SignalExternalWorkflowParameters signalParam
     try {
       service.SignalWorkflowExecution(request);
     } catch (TException e) {
-      throw new RuntimeException(e);
+      throw CheckedExceptionWrapper.wrap(e);
     }
   }
 
@@ -128,7 +128,7 @@ public void requestCancelWorkflowExecution(WorkflowExecution execution) {
     try {
       service.RequestCancelWorkflowExecution(request);
     } catch (TException e) {
-      throw new RuntimeException(e);
+      throw CheckedExceptionWrapper.wrap(e);
     }
   }
 
@@ -169,7 +169,7 @@ public void terminateWorkflowExecution(TerminateWorkflowExecutionParameters term
     try {
       service.TerminateWorkflowExecution(request);
     } catch (TException e) {
-      throw new RuntimeException(e);
+      throw CheckedExceptionWrapper.wrap(e);
     }
   }
 }

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

@@ -23,9 +23,12 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Deque;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.SynchronousQueue;
@@ -51,13 +54,16 @@ private NamedRunnable(String name, Runnable runnable) {
   }
 
   private static final Logger log = LoggerFactory.getLogger(DeterministicRunnerImpl.class);
-  public static final String WORKFLOW_ROOT_THREAD_NAME = "workflow-root";
+  static final String WORKFLOW_ROOT_THREAD_NAME = "workflow-root";
   private static final ThreadLocal<WorkflowThread> currentThreadThreadLocal = new ThreadLocal<>();
 
   private final Lock lock = new ReentrantLock();
   private final ExecutorService threadPool;
   private final SyncDecisionContext decisionContext;
   private final Deque<WorkflowThread> threads = new ArrayDeque<>(); // protected by lock
+  // Values from RunnerLocalInternal
+  private final Map<RunnerLocalInternal<?>, Object> runnerLocalMap = new HashMap<>();
+
   private final List<WorkflowThread> threadsToAdd = Collections.synchronizedList(new ArrayList<>());
   private final List<NamedRunnable> toExecuteInWorkflowThread = new ArrayList<>();
   private final Supplier<Long> clock;
@@ -128,7 +134,7 @@ static void setCurrentThreadInternal(WorkflowThread coroutine) {
     rootWorkflowThread.start();
   }
 
-  public SyncDecisionContext getDecisionContext() {
+  SyncDecisionContext getDecisionContext() {
     return decisionContext;
   }
 
@@ -347,12 +353,12 @@ Lock getLock() {
   }
 
   /** Register a promise that had failed but wasn't accessed yet. */
-  public void registerFailedPromise(Promise promise) {
+  void registerFailedPromise(Promise promise) {
     failedPromises.add(promise);
   }
 
   /** Forget a failed promise as it was accessed. */
-  public void forgetFailedPromise(Promise promise) {
+  void forgetFailedPromise(Promise promise) {
     failedPromises.remove(promise);
   }
 
@@ -368,4 +374,16 @@ private void checkWorkflowThreadOnly() {
       throw new Error("called from non workflow thread");
     }
   }
+
+  @SuppressWarnings("unchecked")
+  <T> Optional<T> getRunnerLocal(RunnerLocalInternal<T> key) {
+    if (!runnerLocalMap.containsKey(key)) {
+      return Optional.empty();
+    }
+    return Optional.of((T) runnerLocalMap.get(key));
+  }
+
+  <T> void setRunnerLocal(RunnerLocalInternal<T> key, T value) {
+    runnerLocalMap.put(key, value);
+  }
 }

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

@@ -0,0 +1,34 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.sync;
+
+import java.util.Optional;
+import java.util.function.Supplier;
+
+public final class RunnerLocalInternal<T> {
+
+  public T get(Supplier<? extends T> supplier) {
+    Optional<T> result =
+        DeterministicRunnerImpl.currentThreadInternal().getRunner().getRunnerLocal(this);
+    return result.orElse(supplier.get());
+  }
+
+  public void set(T value) {
+    DeterministicRunnerImpl.currentThreadInternal().getRunner().setRunnerLocal(this, value);
+  }
+}

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

@@ -94,7 +94,11 @@ public void signal(String signalName, Object... input) {
     // TODO: Deal with signaling started workflow only, when requested
     // Commented out to support signaling workflows that called continue as new.
     //        p.setRunId(execution.getRunId());
-    genericClient.signalWorkflowExecution(p);
+    try {
+      genericClient.signalWorkflowExecution(p);
+    } catch (Exception e) {
+      throw new WorkflowServiceException(execution.get(), workflowType, e);
+    }
   }
 
   private WorkflowExecution startWithOptions(WorkflowOptions o, Object... args) {
@@ -125,6 +129,8 @@ private WorkflowExecution startWithOptions(WorkflowOptions o, Object... args) {
       WorkflowExecution execution =
           new WorkflowExecution().setWorkflowId(p.getWorkflowId()).setRunId(e.getRunId());
       throw new DuplicateWorkflowException(execution, workflowType.get(), e.getMessage());
+    } catch (Exception e) {
+      throw new WorkflowServiceException(execution.get(), workflowType, e);
     }
     return execution.get();
   }

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

@@ -20,6 +20,7 @@
 import static com.uber.cadence.internal.sync.DeterministicRunnerImpl.currentThreadInternal;
 
 import com.uber.cadence.workflow.CancellationScope;
+import java.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.function.Supplier;
 
@@ -108,4 +109,8 @@ static <R> void exit(R value) {
   }
 
   <R> void exitThread(R value);
+
+  <T> void setThreadLocal(WorkflowThreadLocalInternal<T> key, T value);
+
+  <T> Optional<T> getThreadLocal(WorkflowThreadLocalInternal<T> key);
 }

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

@@ -20,6 +20,9 @@
 import com.uber.cadence.workflow.Promise;
 import java.io.PrintWriter;
 import java.io.StringWriter;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -136,6 +139,7 @@ public void setName(String name) {
   private final RunnableWrapper task;
   private Thread thread;
   private Future<?> taskFuture;
+  private final Map<WorkflowThreadLocalInternal<?>, Object> threadLocalMap = new HashMap<>();
 
   /**
    * If not 0 then thread is blocked on a sleep (or on an operation with a timeout). The value is
@@ -363,6 +367,20 @@ public <R> void exitThread(R value) {
     throw new DestroyWorkflowThreadError("exit");
   }
 
+  @Override
+  public <T> void setThreadLocal(WorkflowThreadLocalInternal<T> key, T value) {
+    threadLocalMap.put(key, value);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public <T> Optional<T> getThreadLocal(WorkflowThreadLocalInternal<T> key) {
+    if (!threadLocalMap.containsKey(key)) {
+      return Optional.empty();
+    }
+    return Optional.of((T) threadLocalMap.get(key));
+  }
+
   /** @return stack trace of the coroutine thread */
   @Override
   public String getStackTrace() {

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

@@ -0,0 +1,33 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.internal.sync;
+
+import java.util.Optional;
+import java.util.function.Supplier;
+
+public final class WorkflowThreadLocalInternal<T> {
+
+  public T get(Supplier<? extends T> supplier) {
+    Optional<T> result = DeterministicRunnerImpl.currentThreadInternal().getThreadLocal(this);
+    return result.orElse(supplier.get());
+  }
+
+  public void set(T value) {
+    DeterministicRunnerImpl.currentThreadInternal().setThreadLocal(this, value);
+  }
+}

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

@@ -0,0 +1,71 @@
+/*
+ *  Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ *  Modifications copyright (C) 2017 Uber Technologies, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"). You may not
+ *  use this file except in compliance with the License. A copy of the License is
+ *  located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ *  or in the "license" file accompanying this file. This file is distributed on
+ *  an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ *  express or implied. See the License for the specific language governing
+ *  permissions and limitations under the License.
+ */
+
+package com.uber.cadence.workflow;
+
+import com.uber.cadence.internal.sync.RunnerLocalInternal;
+import java.util.Objects;
+import java.util.function.Supplier;
+
+/**
+ * A value that is local to a single workflow execution. So it can act as a <i>global</i> variable
+ * for the workflow code. For example the {@code Workflow.isSignalled()} static method returns the
+ * correct value even if there are multiple workflows executing on the same machine simultaneously.
+ * It would be invalid if the {@code signaled} was a {@code static boolean} variable.
+ *
+ * <pre>{@code
+ * public class Workflow {
+ *
+ *   private static final WorkflowLocal<Boolean> signaled = WorkflowLocal.withInitial(() -> false);
+ *
+ *   public static boolean isSignalled() {
+ *     return signaled.get();
+ *   }
+ *
+ *   public void signal() {
+ *     signaled.set(true);
+ *   }
+ * }
+ * }</pre>
+ *
+ * @see WorkflowThreadLocal for thread local that can be used inside workflow code.
+ */
+public final class WorkflowLocal<T> {
+
+  private final RunnerLocalInternal<T> impl = new RunnerLocalInternal<>();
+  private Supplier<? extends T> supplier;
+
+  private WorkflowLocal(Supplier<? extends T> supplier) {
+    this.supplier = Objects.requireNonNull(supplier);
+  }
+
+  public WorkflowLocal() {
+    this.supplier = () -> null;
+  }
+
+  public static <S> WorkflowLocal<S> withInitial(Supplier<? extends S> supplier) {
+    return new WorkflowLocal<>(supplier);
+  }
+
+  public T get() {
+    return impl.get(supplier);
+  }
+
+  public void set(T value) {
+    impl.set(value);
+  }
+}