Added List[Open|Closed]WorkflowExecutions to the TestWorkflowService (#145)

* Eliminate timer failure message on closed workflows

* Added TestWorkflowEnvironment.getDomain

* Implemented TestWorkflowService.listWorkflows

* Workflow.retry for procedure

* Fixed TestWorkflowEnvironemnt.close

Author: mfateev

src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java

@@ -452,7 +452,7 @@ public static WorkflowExecutionCloseStatus waitForWorkflowInstanceCompletion(
     return getCloseStatus(closeEvent);
   }
 
-  private static WorkflowExecutionCloseStatus getCloseStatus(HistoryEvent event) {
+  public static WorkflowExecutionCloseStatus getCloseStatus(HistoryEvent event) {
     switch (event.getEventType()) {
       case WorkflowExecutionCanceled:
         return WorkflowExecutionCloseStatus.CANCELED;
@@ -485,7 +485,7 @@ public static WorkflowExecutionCloseStatus waitForWorkflowInstanceCompletionAcro
       WorkflowExecution workflowExecution,
       long timeout,
       TimeUnit unit)
-      throws InterruptedException, TimeoutException, EntityNotExistsError {
+      throws TimeoutException, EntityNotExistsError {
 
     WorkflowExecution lastExecutionToRun = workflowExecution;
     long millisecondsAtFirstWait = System.currentTimeMillis();

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

@@ -157,6 +157,11 @@ public IWorkflowService getWorkflowService() {
     return service;
   }
 
+  @Override
+  public String getDomain() {
+    return testEnvironmentOptions.getDomain();
+  }
+
   @Override
   public String getDiagnostics() {
     StringBuilder result = new StringBuilder();
@@ -167,7 +172,11 @@ public String getDiagnostics() {
   @Override
   public void close() {
     for (Worker w : workers) {
-      w.shutdown(Duration.ofMillis(10));
+      if (w.isStarted()) {
+        w.shutdown(Duration.ofMillis(10));
+      } else {
+        log.warn("Worker was created, but never started for taskList: " + w.getTaskList());
+      }
     }
     service.close();
   }

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowMutableStateImpl.java

@@ -265,17 +265,6 @@ public void completeDecisionTask(int historySize, RespondDecisionTaskCompletedRe
           this.concurrentToDecision.clear();
           ctx.unlockTimer();
         });
-    lock.lock();
-    try {
-      {
-        if (decision != null && decision.getState() != StateMachines.State.INITIATED) {
-          throw new InternalServiceError(
-              "non null decision after the completion: " + decision.getState());
-        }
-      }
-    } finally {
-      lock.unlock();
-    }
   }
 
   private boolean hasCompleteDecision(List<Decision> decisions) {
@@ -650,9 +639,17 @@ private void processStartTimer(
   }
 
   private void fireTimer(String timerId) {
-    StateMachine<TimerData> timer = timers.get(timerId);
-    if (timer == null) {
-      return; // cancelled already
+    StateMachine<TimerData> timer;
+    lock.lock();
+    try {
+      {
+        timer = timers.get(timerId);
+        if (timer == null || workflow.getState() != State.STARTED) {
+          return; // cancelled already
+        }
+      }
+    } finally {
+      lock.unlock();
     }
     try {
       update(

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowService.java

@@ -67,10 +67,14 @@
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
 import com.uber.cadence.WorkflowExecutionContinuedAsNewEventAttributes;
+import com.uber.cadence.WorkflowExecutionFilter;
+import com.uber.cadence.WorkflowExecutionInfo;
 import com.uber.cadence.internal.testservice.TestWorkflowMutableStateImpl.QueryId;
+import com.uber.cadence.internal.testservice.TestWorkflowStore.WorkflowState;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import java.time.Duration;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -441,15 +445,36 @@ public ListOpenWorkflowExecutionsResponse ListOpenWorkflowExecutions(
       ListOpenWorkflowExecutionsRequest listRequest)
       throws BadRequestError, InternalServiceError, EntityNotExistsError, ServiceBusyError,
           TException {
-    throw new UnsupportedOperationException("not implemented");
+    Optional<String> workflowIdFilter;
+    WorkflowExecutionFilter executionFilter = listRequest.getExecutionFilter();
+    if (executionFilter != null
+        && executionFilter.isSetWorkflowId()
+        && !executionFilter.getWorkflowId().isEmpty()) {
+      workflowIdFilter = Optional.of(executionFilter.getWorkflowId());
+    } else {
+      workflowIdFilter = Optional.empty();
+    }
+    List<WorkflowExecutionInfo> result = store.listWorkflows(WorkflowState.OPEN, workflowIdFilter);
+    return new ListOpenWorkflowExecutionsResponse().setExecutions(result);
   }
 
   @Override
   public ListClosedWorkflowExecutionsResponse ListClosedWorkflowExecutions(
       ListClosedWorkflowExecutionsRequest listRequest)
       throws BadRequestError, InternalServiceError, EntityNotExistsError, ServiceBusyError,
           TException {
-    throw new UnsupportedOperationException("not implemented");
+    Optional<String> workflowIdFilter;
+    WorkflowExecutionFilter executionFilter = listRequest.getExecutionFilter();
+    if (executionFilter != null
+        && executionFilter.isSetWorkflowId()
+        && !executionFilter.getWorkflowId().isEmpty()) {
+      workflowIdFilter = Optional.of(executionFilter.getWorkflowId());
+    } else {
+      workflowIdFilter = Optional.empty();
+    }
+    List<WorkflowExecutionInfo> result =
+        store.listWorkflows(WorkflowState.CLOSED, workflowIdFilter);
+    return new ListClosedWorkflowExecutionsResponse().setExecutions(result);
   }
 
   @Override

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStore.java

@@ -25,11 +25,19 @@
 import com.uber.cadence.PollForActivityTaskResponse;
 import com.uber.cadence.PollForDecisionTaskRequest;
 import com.uber.cadence.PollForDecisionTaskResponse;
+import com.uber.cadence.WorkflowExecutionInfo;
 import java.time.Duration;
+import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 
 interface TestWorkflowStore {
 
+  enum WorkflowState {
+    OPEN,
+    CLOSED
+  }
+
   class TaskListId {
 
     private final String domain;
@@ -147,5 +155,7 @@ GetWorkflowExecutionHistoryResponse getWorkflowExecutionHistory(
 
   void getDiagnostics(StringBuilder result);
 
+  List<WorkflowExecutionInfo> listWorkflows(WorkflowState state, Optional<String> workflowId);
+
   void close();
 }

src/main/java/com/uber/cadence/internal/testservice/TestWorkflowStoreImpl.java

@@ -30,6 +30,7 @@
 import com.uber.cadence.PollForDecisionTaskRequest;
 import com.uber.cadence.PollForDecisionTaskResponse;
 import com.uber.cadence.WorkflowExecution;
+import com.uber.cadence.WorkflowExecutionInfo;
 import com.uber.cadence.internal.common.WorkflowExecutionUtils;
 import com.uber.cadence.internal.testservice.RequestContext.Timer;
 import java.time.Duration;
@@ -38,6 +39,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Optional;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.locks.Condition;
@@ -64,6 +66,14 @@ private HistoryStore(ExecutionId id, Lock lock) {
       this.newEventsCondition = lock.newCondition();
     }
 
+    public boolean isCompleted() {
+      return completed;
+    }
+
+    public List<HistoryEvent> getHistory() {
+      return history;
+    }
+
     private void checkNextEventId(long nextEventId) {
       if (nextEventId != history.size()) {
         throw new IllegalStateException(
@@ -346,6 +356,54 @@ public void getDiagnostics(StringBuilder result) {
     }
   }
 
+  @Override
+  public List<WorkflowExecutionInfo> listWorkflows(
+      WorkflowState state, Optional<String> filterWorkflowId) {
+    List<WorkflowExecutionInfo> result = new ArrayList<>();
+    for (Entry<ExecutionId, HistoryStore> entry : this.histories.entrySet()) {
+      if (state == WorkflowState.OPEN) {
+        if (entry.getValue().isCompleted()) {
+          continue;
+        }
+        ExecutionId executionId = entry.getKey();
+        String workflowId = executionId.getWorkflowId().getWorkflowId();
+        if (filterWorkflowId.isPresent() && !workflowId.equals(filterWorkflowId.get())) {
+          continue;
+        }
+        List<HistoryEvent> history = entry.getValue().getHistory();
+        WorkflowExecutionInfo info =
+            new WorkflowExecutionInfo()
+                .setExecution(executionId.getExecution())
+                .setHistoryLength(history.size())
+                .setStartTime(history.get(0).getTimestamp())
+                .setType(
+                    history.get(0).getWorkflowExecutionStartedEventAttributes().getWorkflowType());
+        result.add(info);
+      } else {
+        if (!entry.getValue().isCompleted()) {
+          continue;
+        }
+        ExecutionId executionId = entry.getKey();
+        String workflowId = executionId.getWorkflowId().getWorkflowId();
+        if (filterWorkflowId.isPresent() && !workflowId.equals(filterWorkflowId.get())) {
+          continue;
+        }
+        List<HistoryEvent> history = entry.getValue().getHistory();
+        WorkflowExecutionInfo info =
+            new WorkflowExecutionInfo()
+                .setExecution(executionId.getExecution())
+                .setHistoryLength(history.size())
+                .setStartTime(history.get(0).getTimestamp())
+                .setType(
+                    history.get(0).getWorkflowExecutionStartedEventAttributes().getWorkflowType())
+                .setCloseStatus(
+                    WorkflowExecutionUtils.getCloseStatus(history.get(history.size() - 1)));
+        result.add(info);
+      }
+    }
+    return result;
+  }
+
   @Override
   public void close() {
     timerService.shutdown();

src/main/java/com/uber/cadence/testing/TestWorkflowEnvironment.java

@@ -134,6 +134,8 @@ static TestWorkflowEnvironment newInstance(TestEnvironmentOptions options) {
   /** Returns the in-memory test Cadence service that is owned by this. */
   IWorkflowService getWorkflowService();
 
+  String getDomain();
+
   /**
    * Returns diagnostic data about the internal service state to the provided {@link StringBuilder}.
    * Currently prints histories of all workflow instances stored in the service. This is useful to

src/main/java/com/uber/cadence/worker/Worker.java

@@ -20,12 +20,15 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.client.WorkflowClient;
+import com.uber.cadence.converter.DataConverter;
 import com.uber.cadence.internal.sync.SyncActivityWorker;
 import com.uber.cadence.internal.sync.SyncWorkflowWorker;
 import com.uber.cadence.internal.worker.SingleWorkerOptions;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.cadence.serviceclient.WorkflowServiceTChannel;
+import com.uber.cadence.worker.WorkerOptions.Builder;
 import com.uber.cadence.workflow.Functions.Func;
+import com.uber.cadence.workflow.WorkflowMethod;
 import java.time.Duration;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
@@ -37,10 +40,11 @@
  */
 public final class Worker {
 
-  private final AtomicBoolean started = new AtomicBoolean();
   private final WorkerOptions options;
+  private final String taskList;
   private final SyncWorkflowWorker workflowWorker;
   private final SyncActivityWorker activityWorker;
+  private final AtomicBoolean started = new AtomicBoolean();
 
   /**
    * Creates worker that connects to the local instance of the Cadence Service that listens on a
@@ -61,8 +65,7 @@ public Worker(String domain, String taskList) {
    * @param domain domain that worker uses to poll.
    * @param taskList task list name worker uses to poll. It uses this name for both decision and
    *     activity task list polls.
-   * @param options Options (like {@link com.uber.cadence.converter.DataConverter} override) for
-   *     configuring worker.
+   * @param options Options (like {@link DataConverter} override) for configuring worker.
    */
   public Worker(String domain, String taskList, WorkerOptions options) {
     this(new WorkflowServiceTChannel(), domain, taskList, options);
@@ -89,8 +92,7 @@ public Worker(String host, int port, String domain, String taskList) {
    * @param domain domain that worker uses to poll.
    * @param taskList task list name worker uses to poll. It uses this name for both decision and
    *     activity task list polls.
-   * @param options Options (like {@link com.uber.cadence.converter.DataConverter} override) for
-   *     configuring worker.
+   * @param options Options (like {@link DataConverter} override) for configuring worker.
    */
   public Worker(String host, int port, String domain, String taskList, WorkerOptions options) {
     this(new WorkflowServiceTChannel(host, port), domain, taskList, options);
@@ -103,15 +105,14 @@ public Worker(String host, int port, String domain, String taskList, WorkerOptio
    * @param domain domain that worker uses to poll.
    * @param taskList task list name worker uses to poll. It uses this name for both decision and
    *     activity task list polls.
-   * @param options Options (like {@link com.uber.cadence.converter.DataConverter} override) for
-   *     configuring worker.
+   * @param options Options (like {@link DataConverter} override) for configuring worker.
    */
   public Worker(IWorkflowService service, String domain, String taskList, WorkerOptions options) {
     Objects.requireNonNull(service, "service");
     Objects.requireNonNull(domain, "domain");
-    Objects.requireNonNull(taskList, "taskList");
+    this.taskList = Objects.requireNonNull(taskList, "taskList");
     if (options == null) {
-      options = new WorkerOptions.Builder().build();
+      options = new Builder().build();
     }
     this.options = options;
     SingleWorkerOptions activityOptions = toActivityOptions(options);
@@ -160,8 +161,7 @@ private SingleWorkerOptions toWorkflowOptions(WorkerOptions options) {
   /**
    * Register workflow implementation classes with a worker. Overwrites previously registered types.
    * A workflow implementation class must implement at least one interface with a method annotated
-   * with {@link com.uber.cadence.workflow.WorkflowMethod}. That method becomes a workflow type that
-   * this worker supports.
+   * with {@link WorkflowMethod}. That method becomes a workflow type that this worker supports.
    *
    * <p>Implementations that share a worker must implement different interfaces as a workflow type
    * is identified by the workflow interface, not by the implementation.
@@ -239,6 +239,10 @@ public void start() {
     }
   }
 
+  public boolean isStarted() {
+    return started.get();
+  }
+
   /**
    * Shutdown a worker, waiting for activities to complete execution up to the specified timeout.
    */
@@ -287,4 +291,8 @@ public <R> R queryWorkflowExecution(
     }
     return workflowWorker.queryWorkflowExecution(execution, queryType, returnType, args);
   }
+
+  public String getTaskList() {
+    return taskList;
+  }
 }

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

@@ -307,6 +307,22 @@ public static <R> R retry(RetryOptions options, Functions.Func<R> fn) {
     return WorkflowInternal.retry(options, fn);
   }
 
+  /**
+   * Invokes function retrying in case of failures according to retry options. Synchronous variant.
+   * Use {@link Async#retry(RetryOptions, Functions.Func)} for asynchronous functions.
+   *
+   * @param options retry options that specify retry policy
+   * @param proc procedure to invoke and retry
+   */
+  public static void retry(RetryOptions options, Functions.Proc proc) {
+    WorkflowInternal.retry(
+        options,
+        () -> {
+          proc.apply();
+          return null;
+        });
+  }
+
   /**
    * If there is a need to return a checked exception from a workflow implementation do not add the
    * exception to a method signature but wrap it using this method before rethrowing. The library