Refactored WorkflowExternalInvocationHandler to initialize UntypedWorkflowStubImpl once (#128)

mfateev · web-flow · commit 2569b4d44f48 · 2018-04-02T15:17:21.000-07:00
* Refactored WorkflowExternalInvocationHandler to initialize UntypedWorkflowStubImpl once

* Changed WorkflowClient to use Optional for runId
diff --git a/src/main/java/com/uber/cadence/client/DuplicateWorkflowException.java b/src/main/java/com/uber/cadence/client/DuplicateWorkflowException.java
@@ -18,6 +18,7 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 /**
  * This exception is thrown in the following cases:
@@ -51,6 +52,6 @@ public final class DuplicateWorkflowException extends WorkflowException {
 
   public DuplicateWorkflowException(
       WorkflowExecution execution, String workflowType, String message) {
-    super(message, execution, workflowType, null);
+    super(message, execution, Optional.of(workflowType), null);
   }
 }
diff --git a/src/main/java/com/uber/cadence/client/UntypedWorkflowStub.java b/src/main/java/com/uber/cadence/client/UntypedWorkflowStub.java
@@ -18,6 +18,7 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -28,7 +29,7 @@ public interface UntypedWorkflowStub {
 
   WorkflowExecution start(Object... args);
 
-  String getWorkflowType();
+  Optional<String> getWorkflowType();
 
   WorkflowExecution getExecution();
 
@@ -63,4 +64,6 @@ public interface UntypedWorkflowStub {
 
   /** Request cancellation. */
   void cancel();
+
+  Optional<WorkflowOptions> getOptions();
 }
diff --git a/src/main/java/com/uber/cadence/client/WorkflowClient.java b/src/main/java/com/uber/cadence/client/WorkflowClient.java
@@ -31,6 +31,7 @@
 import com.uber.cadence.workflow.Functions.Proc4;
 import com.uber.cadence.workflow.Functions.Proc5;
 import com.uber.cadence.workflow.Functions.Proc6;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -174,11 +175,22 @@ static WorkflowClient newInstance(
    * Creates workflow client stub for a known execution. Use it to send signals or queries to a
    * running workflow. Do not call methods annotated with @WorkflowMethod.
    *
-   * @param workflowInterface interface that given workflow implements
-   * @param execution workflow id and optional run id for execution
+   * @param workflowInterface interface that given workflow implements.
+   * @param workflowId Workflow id.
+   * @return Stub that implements workflowInterface and can be used to signal or query it.
+   */
+  <T> T newWorkflowStub(Class<T> workflowInterface, String workflowId);
+
+  /**
+   * Creates workflow client stub for a known execution. Use it to send signals or queries to a
+   * running workflow. Do not call methods annotated with @WorkflowMethod.
+   *
+   * @param workflowInterface interface that given workflow implements.
+   * @param workflowId Workflow id.
+   * @param runId Run id of the workflow execution.
    * @return Stub that implements workflowInterface and can be used to signal or query it.
    */
-  <T> T newWorkflowStub(Class<T> workflowInterface, WorkflowExecution execution);
+  <T> T newWorkflowStub(Class<T> workflowInterface, String workflowId, Optional<String> runId);
 
   /**
    * Creates workflow untyped client stub that can be used to start a single workflow execution.
@@ -191,14 +203,29 @@ static WorkflowClient newInstance(
    */
   UntypedWorkflowStub newUntypedWorkflowStub(String workflowType, WorkflowOptions options);
 
+  /**
+   * Creates workflow untyped client stub for a known execution. Use it to send signals or queries
+   * to a running workflow. Do not call methods annotated with @WorkflowMethod.
+   *
+   * @param workflowId workflow id and optional run id for execution
+   * @param runId runId of the workflow execution. If not provided the last workflow with the given
+   *     workflowId is assumed.
+   * @param workflowType type of the workflow. Optional as it is used for error reporting only.
+   * @return Stub that can be used to start workflow and later to signal or query it.
+   */
+  UntypedWorkflowStub newUntypedWorkflowStub(
+      String workflowId, Optional<String> runId, Optional<String> workflowType);
+
   /**
    * Creates workflow untyped client stub for a known execution. Use it to send signals or queries
    * to a running workflow. Do not call methods annotated with @WorkflowMethod.
    *
    * @param execution workflow id and optional run id for execution
+   * @param workflowType type of the workflow. Optional as it is used for error reporting only.
    * @return Stub that can be used to start workflow and later to signal or query it.
    */
-  UntypedWorkflowStub newUntypedWorkflowStub(WorkflowExecution execution);
+  UntypedWorkflowStub newUntypedWorkflowStub(
+      WorkflowExecution execution, Optional<String> workflowType);
 
   /**
    * Creates new {@link ActivityCompletionClient} that can be used to complete activities
diff --git a/src/main/java/com/uber/cadence/client/WorkflowException.java b/src/main/java/com/uber/cadence/client/WorkflowException.java
@@ -19,6 +19,7 @@
 
 import com.uber.cadence.WorkflowExecution;
 import com.uber.cadence.workflow.ChildWorkflowException;
+import java.util.Optional;
 
 /**
  * Base exception for all workflow failures returned by an external client. Note that inside a
@@ -27,22 +28,22 @@
 public class WorkflowException extends RuntimeException {
 
   private final WorkflowExecution execution;
-  private final String workflowType;
+  private final Optional<String> workflowType;
 
   protected WorkflowException(
-      String message, WorkflowExecution execution, String workflowType, Throwable cause) {
+      String message, WorkflowExecution execution, Optional<String> workflowType, Throwable cause) {
     super(getMessage(message, execution, workflowType), cause);
     this.execution = execution;
     this.workflowType = workflowType;
   }
 
   private static String getMessage(
-      String message, WorkflowExecution execution, String workflowType) {
+      String message, WorkflowExecution execution, Optional<String> workflowType) {
     StringBuilder result = new StringBuilder();
     result.append(message);
-    if (workflowType != null) {
+    if (workflowType.isPresent()) {
       result.append(", WorkflowType=\"");
-      result.append(workflowType);
+      result.append(workflowType.get());
       result.append("\"");
     }
     result.append(", WorkflowExecution=\"");
@@ -54,7 +55,7 @@ public WorkflowExecution getExecution() {
     return execution;
   }
 
-  public String getWorkflowType() {
+  public Optional<String> getWorkflowType() {
     return workflowType;
   }
 }
diff --git a/src/main/java/com/uber/cadence/client/WorkflowFailureException.java b/src/main/java/com/uber/cadence/client/WorkflowFailureException.java
@@ -18,6 +18,7 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 /**
  * Indicates that a workflow failed. An original cause of the workflow failure can be retrieved
@@ -29,7 +30,7 @@ public final class WorkflowFailureException extends WorkflowException {
 
   public WorkflowFailureException(
       WorkflowExecution execution,
-      String workflowType,
+      Optional<String> workflowType,
       long decisionTaskCompletedEventId,
       Throwable failure) {
     super(
diff --git a/src/main/java/com/uber/cadence/client/WorkflowNotFoundException.java b/src/main/java/com/uber/cadence/client/WorkflowNotFoundException.java
@@ -18,6 +18,7 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 /**
  * Thrown when workflow with the given id is not known to the cadence service. It could be because
@@ -26,7 +27,7 @@
 public final class WorkflowNotFoundException extends WorkflowException {
 
   public WorkflowNotFoundException(
-      WorkflowExecution execution, String workflowType, String message) {
+      WorkflowExecution execution, Optional<String> workflowType, String message) {
     super(message, execution, workflowType, null);
   }
 }
diff --git a/src/main/java/com/uber/cadence/client/WorkflowQueryException.java b/src/main/java/com/uber/cadence/client/WorkflowQueryException.java
@@ -18,10 +18,11 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 public final class WorkflowQueryException extends WorkflowException {
 
   public WorkflowQueryException(WorkflowExecution execution, String message) {
-    super(message, execution, null, null);
+    super(message, execution, Optional.empty(), null);
   }
 }
diff --git a/src/main/java/com/uber/cadence/client/WorkflowServiceException.java b/src/main/java/com/uber/cadence/client/WorkflowServiceException.java
@@ -18,10 +18,11 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 public final class WorkflowServiceException extends WorkflowException {
 
   public WorkflowServiceException(WorkflowExecution execution, String message) {
-    super(message, execution, null, null);
+    super(message, execution, Optional.empty(), null);
   }
 }
diff --git a/src/main/java/com/uber/cadence/client/WorkflowTerminatedException.java b/src/main/java/com/uber/cadence/client/WorkflowTerminatedException.java
@@ -18,6 +18,7 @@
 package com.uber.cadence.client;
 
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 /**
  * Indicates that a workflow was forcefully terminated by an external command to Cadence service.
@@ -29,7 +30,7 @@ public final class WorkflowTerminatedException extends WorkflowException {
 
   public WorkflowTerminatedException(
       WorkflowExecution execution,
-      String workflowType,
+      Optional<String> workflowType,
       String reason,
       String identity,
       byte[] details) {
diff --git a/src/main/java/com/uber/cadence/client/WorkflowTimedOutException.java b/src/main/java/com/uber/cadence/client/WorkflowTimedOutException.java
@@ -19,6 +19,7 @@
 
 import com.uber.cadence.TimeoutType;
 import com.uber.cadence.WorkflowExecution;
+import java.util.Optional;
 
 /**
  * Indicates that a workflow exceeded its execution timeout and was forcefully terminated by the
@@ -29,7 +30,7 @@ public final class WorkflowTimedOutException extends WorkflowException {
   private final TimeoutType timeoutType;
 
   public WorkflowTimedOutException(
-      WorkflowExecution execution, String workflowType, TimeoutType timeoutType) {
+      WorkflowExecution execution, Optional<String> workflowType, TimeoutType timeoutType) {
     super(String.valueOf(timeoutType) + " timeout type", execution, workflowType, null);
     this.timeoutType = timeoutType;
   }
diff --git a/src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java b/src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java
@@ -54,6 +54,7 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
@@ -93,7 +94,7 @@ public static byte[] getWorkflowExecutionResult(
       IWorkflowService service,
       String domain,
       WorkflowExecution workflowExecution,
-      String workflowType,
+      Optional<String> workflowType,
       long timeout,
       TimeUnit unit)
       throws TimeoutException, CancellationException, WorkflowExecutionFailedException,
@@ -108,7 +109,7 @@ public static CompletableFuture<byte[]> getWorkflowExecutionResultAsync(
       IWorkflowService service,
       String domain,
       WorkflowExecution workflowExecution,
-      String workflowType,
+      Optional<String> workflowType,
       long timeout,
       TimeUnit unit) {
     return getInstanceCloseEventAsync(service, domain, workflowExecution, timeout, unit)
@@ -117,7 +118,7 @@ public static CompletableFuture<byte[]> getWorkflowExecutionResultAsync(
   }
 
   private static byte[] getResultFromCloseEvent(
-      WorkflowExecution workflowExecution, String workflowType, HistoryEvent closeEvent) {
+      WorkflowExecution workflowExecution, Optional<String> workflowType, HistoryEvent closeEvent) {
     if (closeEvent == null) {
       throw new IllegalStateException("Workflow is still running");
     }
diff --git a/src/main/java/com/uber/cadence/internal/sync/UntypedWorkflowStubImpl.java b/src/main/java/com/uber/cadence/internal/sync/UntypedWorkflowStubImpl.java
@@ -39,6 +39,7 @@
 import com.uber.cadence.internal.external.GenericWorkflowClientExternal;
 import com.uber.cadence.internal.replay.QueryWorkflowParameters;
 import com.uber.cadence.internal.replay.SignalExternalWorkflowParameters;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
@@ -51,19 +52,20 @@ class UntypedWorkflowStubImpl implements UntypedWorkflowStub {
 
   private final GenericWorkflowClientExternal genericClient;
   private final DataConverter dataConverter;
-  private final String workflowType;
+  private final Optional<String> workflowType;
   private AtomicReference<WorkflowExecution> execution = new AtomicReference<>();
-  private final WorkflowOptions options;
+  private final Optional<WorkflowOptions> options;
 
   UntypedWorkflowStubImpl(
       GenericWorkflowClientExternal genericClient,
       DataConverter dataConverter,
+      Optional<String> workflowType,
       WorkflowExecution execution) {
     this.genericClient = genericClient;
     this.dataConverter = dataConverter;
-    this.workflowType = null;
+    this.workflowType = workflowType;
     this.execution.set(execution);
-    this.options = null;
+    this.options = Optional.empty();
   }
 
   UntypedWorkflowStubImpl(
@@ -73,8 +75,8 @@ class UntypedWorkflowStubImpl implements UntypedWorkflowStub {
       WorkflowOptions options) {
     this.genericClient = genericClient;
     this.dataConverter = dataConverter;
-    this.workflowType = workflowType;
-    this.options = options;
+    this.workflowType = Optional.of(workflowType);
+    this.options = Optional.of(options);
   }
 
   @Override
@@ -90,14 +92,13 @@ public void signal(String signalName, Object... input) {
     genericClient.signalWorkflowExecution(p);
   }
 
-  WorkflowExecution startWithOptions(WorkflowOptions o, Object... args) {
-    if (o == null) {
-      throw new IllegalStateException(
-          "UntypedWorkflowStub wasn't created through "
-              + "WorkflowClient::newUntypedWorkflowStub(String workflowType, WorkflowOptions options)");
-    }
+  private WorkflowExecution startWithOptions(WorkflowOptions o, Object... args) {
     if (execution.get() != null) {
-      throw new IllegalStateException("already started for execution=" + execution.get());
+      throw new DuplicateWorkflowException(
+          execution.get(),
+          workflowType.get(),
+          "Cannot reuse a stub instance to start more than one workflow execution. The stub "
+              + "points to already started execution.");
     }
     StartWorkflowExecutionParameters p = new StartWorkflowExecutionParameters();
     p.setTaskStartToCloseTimeoutSeconds(o.getTaskStartToCloseTimeout().getSeconds());
@@ -108,7 +109,7 @@ WorkflowExecution startWithOptions(WorkflowOptions o, Object... args) {
     }
     p.setExecutionStartToCloseTimeoutSeconds(o.getExecutionStartToCloseTimeout().getSeconds());
     p.setInput(dataConverter.toData(args));
-    p.setWorkflowType(new WorkflowType().setName(workflowType));
+    p.setWorkflowType(new WorkflowType().setName(workflowType.get()));
     p.setTaskList(o.getTaskList());
     p.setChildPolicy(o.getChildPolicy());
     try {
@@ -118,18 +119,21 @@ WorkflowExecution startWithOptions(WorkflowOptions o, Object... args) {
           new WorkflowExecution().setWorkflowId(p.getWorkflowId()).setRunId(e.getRunId()));
       WorkflowExecution execution =
           new WorkflowExecution().setWorkflowId(p.getWorkflowId()).setRunId(e.getRunId());
-      throw new DuplicateWorkflowException(execution, workflowType, e.getMessage());
+      throw new DuplicateWorkflowException(execution, workflowType.get(), e.getMessage());
     }
     return execution.get();
   }
 
   @Override
   public WorkflowExecution start(Object... args) {
-    return startWithOptions(WorkflowOptions.merge(null, options), args);
+    if (!options.isPresent()) {
+      throw new IllegalStateException("Required parameter WorkflowOptions is missing");
+    }
+    return startWithOptions(WorkflowOptions.merge(null, options.get()), args);
   }
 
   @Override
-  public String getWorkflowType() {
+  public Optional<String> getWorkflowType() {
     return workflowType;
   }
 
@@ -266,6 +270,11 @@ public void cancel() {
     genericClient.requestCancelWorkflowExecution(execution.get());
   }
 
+  @Override
+  public Optional<WorkflowOptions> getOptions() {
+    return options;
+  }
+
   private void checkStarted() {
     if (execution.get() == null || execution.get().getWorkflowId() == null) {
       throw new IllegalStateException("Null workflowId. Was workflow started?");
diff --git a/src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java b/src/main/java/com/uber/cadence/internal/sync/WorkflowClientInternal.java
diff --git a/src/main/java/com/uber/cadence/internal/sync/WorkflowExternalInvocationHandler.java b/src/main/java/com/uber/cadence/internal/sync/WorkflowExternalInvocationHandler.java
diff --git a/src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java b/src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java
diff --git a/src/test/java/com/uber/cadence/workflow/WorkflowTest.java b/src/test/java/com/uber/cadence/workflow/WorkflowTest.java

Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@`
`18`	`18`	`package com.uber.cadence.client;`
`19`	`19`
`20`	`20`	`import com.uber.cadence.WorkflowExecution;`
	`21`	`+import java.util.Optional;`
`21`	`22`
`22`	`23`	`/**`
`23`	`24`	`* This exception is thrown in the following cases:`
`@@ -51,6 +52,6 @@ public final class DuplicateWorkflowException extends WorkflowException {`
`51`	`52`
`52`	`53`	`public DuplicateWorkflowException(`
`53`	`54`	`WorkflowExecution execution, String workflowType, String message) {`
`54`		`- super(message, execution, workflowType, null);`
	`55`	`+ super(message, execution, Optional.of(workflowType), null);`
`55`	`56`	`}`
`56`	`57`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,10 +18,11 @@`
`18`	`18`	`package com.uber.cadence.client;`
`19`	`19`
`20`	`20`	`import com.uber.cadence.WorkflowExecution;`
	`21`	`+import java.util.Optional;`
`21`	`22`
`22`	`23`	`public final class WorkflowQueryException extends WorkflowException {`
`23`	`24`
`24`	`25`	`public WorkflowQueryException(WorkflowExecution execution, String message) {`
`25`		`- super(message, execution, null, null);`
	`26`	`+ super(message, execution, Optional.empty(), null);`
`26`	`27`	`}`
`27`	`28`	`}`