Adding cross domain signaling (#755)

Adding cross domain signaling.

Author: mindaugasbarcauskas

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

@@ -25,6 +25,7 @@
 import com.uber.cadence.workflow.CompletablePromise;
 import com.uber.cadence.workflow.Promise;
 import com.uber.cadence.workflow.SignalExternalWorkflowException;
+import com.uber.cadence.workflow.SignalOptions;
 import com.uber.cadence.workflow.Workflow;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import com.uber.cadence.workflow.WorkflowInterceptor.WorkflowResult;
@@ -113,4 +114,23 @@ public void signal(String signalName, Object... args) {
       throw e;
     }
   }
+
+  @Override
+  public void signal(SignalOptions signalOptions, Object... args) {
+    Promise<Void> signaled =
+        decisionContext.signalExternalWorkflow(
+            signalOptions.getDomain(), execution.get(), signalOptions.getSignalName(), args);
+    if (AsyncInternal.isAsync()) {
+      AsyncInternal.setAsyncResult(signaled);
+      return;
+    }
+    try {
+      signaled.get();
+    } catch (SignalExternalWorkflowException e) {
+      // Reset stack to the current one. Otherwise it is very confusing to see a stack of
+      // an event handling method.
+      e.setStackTrace(Thread.currentThread().getStackTrace());
+      throw e;
+    }
+  }
 }

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

@@ -22,6 +22,7 @@
 import com.uber.cadence.workflow.ExternalWorkflowStub;
 import com.uber.cadence.workflow.Promise;
 import com.uber.cadence.workflow.SignalExternalWorkflowException;
+import com.uber.cadence.workflow.SignalOptions;
 import com.uber.cadence.workflow.WorkflowInterceptor;
 import java.util.Objects;
 
@@ -59,6 +60,25 @@ public void signal(String signalName, Object... args) {
     }
   }
 
+  @Override
+  public void signal(SignalOptions signalOptions, Object... args) {
+    Promise<Void> signaled =
+        decisionContext.signalExternalWorkflow(
+            signalOptions.getDomain(), execution, signalOptions.getSignalName(), args);
+    if (AsyncInternal.isAsync()) {
+      AsyncInternal.setAsyncResult(signaled);
+      return;
+    }
+    try {
+      signaled.get();
+    } catch (SignalExternalWorkflowException e) {
+      // Reset stack to the current one. Otherwise it is very confusing to see a stack of
+      // an event handling method.
+      e.setStackTrace(Thread.currentThread().getStackTrace());
+      throw e;
+    }
+  }
+
   @Override
   public void cancel() {
     Promise<Void> cancelRequested = decisionContext.cancelWorkflow(execution);

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

@@ -634,10 +634,17 @@ public DecisionContext getContext() {
   @Override
   public Promise<Void> signalExternalWorkflow(
       WorkflowExecution execution, String signalName, Object[] args) {
+    return this.signalExternalWorkflow(null, execution, signalName, args);
+  }
+
+  @Override
+  public Promise<Void> signalExternalWorkflow(
+      String domain, WorkflowExecution execution, String signalName, Object[] args) {
     SignalExternalWorkflowParameters parameters = new SignalExternalWorkflowParameters();
     parameters.setSignalName(signalName);
     parameters.setWorkflowId(execution.getWorkflowId());
     parameters.setRunId(execution.getRunId());
+    parameters.setDomain(domain);
     byte[] input = getDataConverter().toData(args);
     parameters.setInput(input);
     CompletablePromise<Void> result = Workflow.newPromise();

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

@@ -191,6 +191,12 @@ public Random newRandom() {
       throw new UnsupportedOperationException("not implemented");
     }
 
+    @Override
+    public Promise<Void> signalExternalWorkflow(
+        String domain, WorkflowExecution execution, String signalName, Object[] args) {
+      throw new UnsupportedOperationException("not implemented");
+    }
+
     @Override
     public Promise<Void> signalExternalWorkflow(
         WorkflowExecution execution, String signalName, Object[] args) {

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

@@ -121,16 +121,23 @@ public final class TestWorkflowEnvironmentInternal implements TestWorkflowEnviro
   private final WorkflowServiceWrapper service;
   private final WorkerFactory workerFactory;
 
-  public TestWorkflowEnvironmentInternal(TestEnvironmentOptions options) {
+  public TestWorkflowEnvironmentInternal(
+      WorkflowServiceWrapper workflowServiceWrapper, TestEnvironmentOptions options) {
     if (options == null) {
       this.testEnvironmentOptions = new TestEnvironmentOptions.Builder().build();
     } else {
       this.testEnvironmentOptions = options;
     }
-    service = new WorkflowServiceWrapper();
-    service.lockTimeSkipping("TestWorkflowEnvironmentInternal constructor");
+
+    if (workflowServiceWrapper == null) {
+      this.service = new WorkflowServiceWrapper();
+    } else {
+      this.service = workflowServiceWrapper;
+    }
+
+    this.service.lockTimeSkipping("TestWorkflowEnvironmentInternal constructor");
     WorkflowClient client =
-        WorkflowClient.newInstance(service, testEnvironmentOptions.getWorkflowClientOptions());
+        WorkflowClient.newInstance(this.service, testEnvironmentOptions.getWorkflowClientOptions());
     workerFactory =
         WorkerFactory.newInstance(client, testEnvironmentOptions.getWorkerFactoryOptions());
   }
@@ -253,11 +260,11 @@ public WorkerFactory getWorkerFactory() {
     return workerFactory;
   }
 
-  private static class WorkflowServiceWrapper implements IWorkflowService {
+  public static class WorkflowServiceWrapper implements IWorkflowService {
 
     private final TestWorkflowService impl;
 
-    private WorkflowServiceWrapper() {
+    public WorkflowServiceWrapper() {
       impl = new TestWorkflowService();
     }
 

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

@@ -93,11 +93,17 @@
 public interface TestWorkflowEnvironment {
 
   static TestWorkflowEnvironment newInstance() {
-    return new TestWorkflowEnvironmentInternal(new TestEnvironmentOptions.Builder().build());
+    return new TestWorkflowEnvironmentInternal(null, new TestEnvironmentOptions.Builder().build());
   }
 
   static TestWorkflowEnvironment newInstance(TestEnvironmentOptions options) {
-    return new TestWorkflowEnvironmentInternal(options);
+    return new TestWorkflowEnvironmentInternal(null, options);
+  }
+
+  static TestWorkflowEnvironment newInstance(
+      TestWorkflowEnvironmentInternal.WorkflowServiceWrapper workflowService,
+      TestEnvironmentOptions options) {
+    return new TestWorkflowEnvironmentInternal(workflowService, options);
   }
 
   /**

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

@@ -44,4 +44,6 @@ public interface ChildWorkflowStub {
   <R> Promise<R> executeAsync(Class<R> resultClass, Type resultType, Object... args);
 
   void signal(String signalName, Object... args);
+
+  void signal(SignalOptions signalOptions, Object... args);
 }

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

@@ -32,5 +32,7 @@ public interface ExternalWorkflowStub {
 
   void signal(String signalName, Object... args);
 
+  void signal(SignalOptions signalOptions, Object... args);
+
   void cancel();
 }

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

@@ -0,0 +1,68 @@
+/*
+ *  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;
+
+public class SignalOptions {
+
+  private String domain;
+  private String signalName;
+
+  private SignalOptions() {}
+
+  public static SignalOptions.Builder newBuilder() {
+    return new Builder();
+  }
+
+  public static class Builder {
+    private SignalOptions signalOptions = new SignalOptions();
+
+    public SignalOptions.Builder setDomain(String domain) {
+      signalOptions.setDomain(domain);
+      return this;
+    }
+
+    public SignalOptions.Builder setSignalName(String signalName) {
+      signalOptions.setSignalName(signalName);
+      return this;
+    }
+
+    public SignalOptions build() {
+      if (signalOptions.getSignalName() == null) {
+        throw new IllegalArgumentException("Signal name must be provided");
+      }
+
+      return signalOptions;
+    }
+  }
+
+  public SignalOptions setDomain(String domain) {
+    this.domain = domain;
+    return this;
+  }
+
+  public String getDomain() {
+    return this.domain;
+  }
+
+  public SignalOptions setSignalName(String signalName) {
+    this.signalName = signalName;
+    return this;
+  }
+
+  public String getSignalName() {
+    return this.signalName;
+  }
+}

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

@@ -77,6 +77,9 @@ <R> WorkflowResult<R> executeChildWorkflow(
   Promise<Void> signalExternalWorkflow(
       WorkflowExecution execution, String signalName, Object[] args);
 
+  Promise<Void> signalExternalWorkflow(
+      String domain, WorkflowExecution execution, String signalName, Object[] args);
+
   Promise<Void> cancelWorkflow(WorkflowExecution execution);
 
   void sleep(Duration duration);