Introduce Workerfactory (#188)

* Introduce Worker.Factory for worker creation
* Change Worker constructors, start and shutdown method to private

Author: halakaraki

build.gradle

@@ -50,6 +50,7 @@ dependencies {
     compile group: 'org.apache.thrift', name: 'libthrift', version: '0.9.3'
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.2'
     compile group: 'com.uber.m3', name: 'tally-core', version: '0.2.1'
+    compile group: 'com.google.guava', name: 'guava', version: '25.1-jre'
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'com.googlecode.junit-toolbox', name: 'junit-toolbox', version: '2.4'
     testCompile group: 'ch.qos.logback', name: 'logback-classic', version: '1.2.3'

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

@@ -78,9 +78,6 @@
 import com.uber.cadence.worker.WorkerOptions;
 import java.lang.reflect.Type;
 import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -98,7 +95,7 @@ public final class TestWorkflowEnvironmentInternal implements TestWorkflowEnviro
 
   private final TestEnvironmentOptions testEnvironmentOptions;
   private final WorkflowServiceWrapper service;
-  private final List<Worker> workers = Collections.synchronizedList(new ArrayList<>());
+  private final Worker.Factory workerFactory;
 
   public TestWorkflowEnvironmentInternal(TestEnvironmentOptions options) {
     if (options == null) {
@@ -108,6 +105,7 @@ public TestWorkflowEnvironmentInternal(TestEnvironmentOptions options) {
     }
     service = new WorkflowServiceWrapper();
     service.lockTimeSkipping("TestWorkflowEnvironmentInternal constructor");
+    workerFactory = new Worker.Factory(service, options.getDomain());
   }
 
   @Override
@@ -127,9 +125,7 @@ public Worker newWorker(
       builder.setDataConverter(testEnvironmentOptions.getDataConverter());
     }
     builder = overrideOptions.apply(builder);
-    Worker result =
-        new Worker(service, testEnvironmentOptions.getDomain(), taskList, builder.build());
-    workers.add(result);
+    Worker result = workerFactory.newWorker(taskList, builder.build());
     return result;
   }
 
@@ -191,16 +187,15 @@ public String getDiagnostics() {
 
   @Override
   public void close() {
-    for (Worker w : workers) {
-      if (w.isStarted()) {
-        w.shutdown(Duration.ofMillis(10));
-      } else {
-        log.warn("Worker was created, but never started for taskList: " + w.getTaskList());
-      }
-    }
+    workerFactory.shutdown(Duration.ofMillis(10));
     service.close();
   }
 
+  @Override
+  public void start() {
+    workerFactory.start();
+  }
+
   private static class WorkflowServiceWrapper implements IWorkflowService {
 
     private final TestWorkflowService impl;

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

@@ -176,4 +176,7 @@ Worker newWorker(
    * service.
    */
   void close();
+
+  /** Starts all the workers created by the test environment instance */
+  void start();
 }

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

@@ -18,6 +18,9 @@
 package com.uber.cadence.worker;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.client.WorkflowClient;
 import com.uber.cadence.converter.DataConverter;
@@ -33,6 +36,8 @@
 import com.uber.m3.util.ImmutableMap;
 import java.lang.reflect.Type;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.TimeUnit;
@@ -49,95 +54,44 @@ public final class Worker {
   private final SyncWorkflowWorker workflowWorker;
   private final SyncActivityWorker activityWorker;
   private final AtomicBoolean started = new AtomicBoolean();
+  private final AtomicBoolean closed = new AtomicBoolean();
 
   /**
-   * Creates worker that connects to the local instance of the Cadence Service that listens on a
-   * default port (7933).
-   *
-   * @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.
-   */
-  public Worker(String domain, String taskList) {
-    this(domain, taskList, null);
-  }
-
-  /**
-   * Creates worker that connects to the local instance of the Cadence Service that listens on a
-   * default port (7933).
+   * Creates worker that connects to an instance of the Cadence Service.
    *
+   * @param service client to the Cadence Service endpoint.
    * @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 DataConverter} override) for configuring worker.
    */
-  public Worker(String domain, String taskList, WorkerOptions options) {
-    this(new WorkflowServiceTChannel(), domain, taskList, options);
-  }
+  private Worker(IWorkflowService service, String domain, String taskList, WorkerOptions options) {
+    Objects.requireNonNull(service, "service should not be null");
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(domain), "domain should not be an empty string");
+    Preconditions.checkArgument(
+        !Strings.isNullOrEmpty(taskList), "taskList should not be an empty string");
 
-  /**
-   * Creates worker that connects to an instance of the Cadence Service.
-   *
-   * @param host of the Cadence Service endpoint
-   * @param port of the Cadence Service endpoint
-   * @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.
-   */
-  public Worker(String host, int port, String domain, String taskList) {
-    this(new WorkflowServiceTChannel(host, port), domain, taskList, null);
-  }
+    this.taskList = taskList;
+    this.options = MoreObjects.firstNonNull(options, new Builder().build());
 
-  /**
-   * Creates worker that connects to an instance of the Cadence Service.
-   *
-   * @param host of the Cadence Service endpoint
-   * @param port of the Cadence Service endpoint
-   * @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 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);
-  }
+    SingleWorkerOptions activityOptions = toActivityOptions(this.options, domain, taskList);
+    activityWorker =
+        this.options.isDisableActivityWorker()
+            ? null
+            : new SyncActivityWorker(service, domain, taskList, activityOptions);
 
-  /**
-   * Creates worker that connects to an instance of the Cadence Service.
-   *
-   * @param service client to the Cadence Service endpoint.
-   * @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 DataConverter} override) for configuring worker.
-   */
-  public Worker(IWorkflowService service, String domain, String taskList, WorkerOptions options) {
-    Objects.requireNonNull(service, "service");
-    Objects.requireNonNull(domain, "domain");
-    this.taskList = Objects.requireNonNull(taskList, "taskList");
-    if (options == null) {
-      options = new Builder().build();
-    }
-    this.options = options;
-    SingleWorkerOptions activityOptions = toActivityOptions(options, domain, taskList);
-    if (!options.isDisableActivityWorker()) {
-      activityWorker = new SyncActivityWorker(service, domain, taskList, activityOptions);
-    } else {
-      activityWorker = null;
-    }
-    SingleWorkerOptions workflowOptions = toWorkflowOptions(options, domain, taskList);
-    if (!options.isDisableWorkflowWorker()) {
-      workflowWorker =
-          new SyncWorkflowWorker(
-              service,
-              domain,
-              taskList,
-              options.getInterceptorFactory(),
-              workflowOptions,
-              options.getMaxWorkflowThreads());
-    } else {
-      workflowWorker = null;
-    }
+    SingleWorkerOptions workflowOptions = toWorkflowOptions(this.options, domain, taskList);
+    workflowWorker =
+        this.options.isDisableWorkflowWorker()
+            ? null
+            : new SyncWorkflowWorker(
+                service,
+                domain,
+                taskList,
+                this.options.getInterceptorFactory(),
+                workflowOptions,
+                this.options.getMaxWorkflowThreads());
   }
 
   private SingleWorkerOptions toActivityOptions(
@@ -190,10 +144,13 @@ private SingleWorkerOptions toWorkflowOptions(
    * workflows are stateful and a new instance is created for each workflow execution.
    */
   public void registerWorkflowImplementationTypes(Class<?>... workflowImplementationClasses) {
-    if (workflowWorker == null) {
-      throw new IllegalStateException("disableWorkflowWorker is set in worker options");
-    }
-    checkNotStarted();
+    Preconditions.checkState(
+        workflowWorker != null,
+        "registerWorkflowImplementationTypes is not allowed when disableWorkflowWorker is set in worker options");
+    Preconditions.checkState(
+        !started.get(),
+        "registerWorkflowImplementationTypes is not allowed after worker has started");
+
     workflowWorker.setWorkflowImplementationTypes(workflowImplementationClasses);
   }
 
@@ -234,20 +191,17 @@ public <R> void addWorkflowImplementationFactory(Class<R> workflowInterface, Fun
    * <p>
    */
   public void registerActivitiesImplementations(Object... activityImplementations) {
-    if (activityWorker == null) {
-      throw new IllegalStateException("disableActivityWorker is set in worker options");
-    }
-    checkNotStarted();
-    activityWorker.setActivitiesImplementation(activityImplementations);
-  }
+    Preconditions.checkState(
+        activityWorker != null,
+        "registerActivitiesImplementations is not allowed when disableWorkflowWorker is set in worker options");
+    Preconditions.checkState(
+        !started.get(),
+        "registerActivitiesImplementations is not allowed after worker has started");
 
-  private void checkNotStarted() {
-    if (started.get()) {
-      throw new IllegalStateException("already started");
-    }
+    activityWorker.setActivitiesImplementation(activityImplementations);
   }
 
-  public void start() {
+  private void start() {
     if (!started.compareAndSet(false, true)) {
       return;
     }
@@ -263,10 +217,14 @@ public boolean isStarted() {
     return started.get();
   }
 
+  public boolean isClosed() {
+    return closed.get();
+  }
+
   /**
    * Shutdown a worker, waiting for activities to complete execution up to the specified timeout.
    */
-  public void shutdown(Duration timeout) {
+  private void shutdown(Duration timeout) {
     try {
       long time = System.currentTimeMillis();
       if (activityWorker != null) {
@@ -276,6 +234,7 @@ public void shutdown(Duration timeout) {
         long left = timeout.toMillis() - (System.currentTimeMillis() - time);
         workflowWorker.shutdownAndAwaitTermination(left, TimeUnit.MILLISECONDS);
       }
+      closed.set(true);
     } catch (InterruptedException e) {
       throw new RuntimeException(e);
     }
@@ -344,4 +303,85 @@ public <R> R queryWorkflowExecution(
   public String getTaskList() {
     return taskList;
   }
+
+  public static final class Factory {
+
+    private final List<Worker> workers = new ArrayList<>();
+    private final IWorkflowService workflowService;
+    private final String domain;
+    private State state = State.Initial;
+
+    private final String statusErrorMessage =
+        "attempted to %s while in %s state. Acceptable States: %s";
+
+    public Factory(String domain) {
+      this(new WorkflowServiceTChannel(), domain);
+    }
+
+    public Factory(String host, int port, String domain) {
+      this(new WorkflowServiceTChannel(host, port), domain);
+    }
+
+    public Factory(IWorkflowService workflowService, String domain) {
+      Objects.requireNonNull(workflowService, "workflowService should not be null");
+      Preconditions.checkArgument(!Strings.isNullOrEmpty(domain), "domain should not be an empty string");
+
+      this.workflowService = workflowService;
+      this.domain = domain;
+    }
+
+    public Worker newWorker(String taskList) {
+      return newWorker(taskList, null);
+    }
+
+    public Worker newWorker(String taskList, WorkerOptions options) {
+      Preconditions.checkArgument(!Strings.isNullOrEmpty(taskList), "taskList should not be an empty string");
+
+      synchronized (this) {
+        Preconditions.checkState(
+            state == State.Initial,
+            String.format(
+                statusErrorMessage, "create new worker", state.name(), State.Initial.name()));
+        Worker worker = new Worker(workflowService, domain, taskList, options);
+        workers.add(worker);
+        return worker;
+      }
+    }
+
+    public void start() {
+      synchronized (this) {
+        Preconditions.checkState(
+            state == State.Initial || state == State.Started,
+            String.format(
+                statusErrorMessage,
+                "start WorkerFactory",
+                state.name(),
+                String.format("%s, %s", State.Initial.name(), State.Initial.name())));
+        if (state == State.Started) {
+          return;
+        }
+        state = State.Started;
+
+        for (Worker worker : workers) {
+          worker.start();
+        }
+      }
+    }
+
+    public void shutdown(Duration timeout) {
+      synchronized (this) {
+        state = State.Shutdown;
+
+        for (Worker worker : workers) {
+          worker.shutdown(timeout);
+        }
+      }
+    }
+
+    enum State {
+      Initial,
+      Started,
+      Shutdown
+    }
+  }
 }