Fix and test AnyCommandCompleted (#55)

Author: longquanzheng

README.md

@@ -46,7 +46,7 @@ Run the command `git submodule update --remote --merge` to update IDL to the lat
 
 - [x] Reset workflow API (Cadence only, TODO for Temporal)
 - [ ] Command type(s) for inter-state communications (e.g. internal channel)
-- [ ] AnyCommandCompleted Decider trigger type
+- [x] AnyCommandCompleted Decider trigger type
 - [ ] More workflow start options: IdReusePolicy, initial earch attributes, cron schedule, retry, etc
 - [ ] StateOption: Start/Decide API timeout and retry
 - [ ] Reset workflow by stateId

src/main/java/io/github/cadenceoss/iwf/core/JacksonJsonObjectEncoder.java

@@ -76,6 +76,10 @@ public EncodedObject encode(Object object) {
 
   @Override
   public <T> T decode(EncodedObject encodedObject, Class<T> type) {
+    if (encodedObject == null) {
+      return null;
+    }
+    
     String data = encodedObject.getData();
     if (data == null || data.isEmpty()) {
       return null;

src/main/java/io/github/cadenceoss/iwf/core/command/SignalCommandResult.java

@@ -3,14 +3,16 @@
 import io.github.cadenceoss.iwf.gen.models.SignalResult;
 import org.immutables.value.Value;
 
+import java.util.Optional;
+
 @Value.Immutable
 public abstract class SignalCommandResult {
 
     public abstract String getCommandId();
 
     public abstract String getSignalChannelName();
 
-    public abstract Object getSignalValue();
+    public abstract Optional<Object> getSignalValue();
 
     public abstract SignalResult.SignalRequestStatusEnum getSignalRequestStatusEnum();
 }

src/main/java/io/github/cadenceoss/iwf/core/mapper/SignalResultMapper.java

@@ -5,6 +5,8 @@
 import io.github.cadenceoss.iwf.core.command.SignalCommandResult;
 import io.github.cadenceoss.iwf.gen.models.SignalResult;
 
+import java.util.Optional;
+
 public class SignalResultMapper {
     public static SignalCommandResult fromGenerated(
             SignalResult signalResult,
@@ -14,7 +16,7 @@ public static SignalCommandResult fromGenerated(
                 .commandId(signalResult.getCommandId())
                 .signalRequestStatusEnum(signalResult.getSignalRequestStatus())
                 .signalChannelName(signalResult.getSignalChannelName())
-                .signalValue(objectEncoder.decode(signalResult.getSignalValue(), signalType))
+                .signalValue(Optional.ofNullable(objectEncoder.decode(signalResult.getSignalValue(), signalType)))
                 .build();
     }
 }

src/test/java/io/github/cadenceoss/iwf/integ/SignalTest.java

@@ -28,7 +28,7 @@ public void testBasicSignalWorkflow() throws InterruptedException {
         final String runId = client.StartWorkflow(
                 BasicSignalWorkflow.class, BasicSignalWorkflowState1.STATE_ID, input, wfId, startOptions);
         client.SignalWorkflow(
-                BasicSignalWorkflow.class, wfId, runId, BasicSignalWorkflowState1.SIGNAL_CHANNEL_NAME, Integer.valueOf(2));
+                BasicSignalWorkflow.class, wfId, runId, BasicSignalWorkflowState1.SIGNAL_CHANNEL_NAME_1, Integer.valueOf(2));
         final Integer output = client.GetSimpleWorkflowResultWithWait(Integer.class, wfId);
         Assertions.assertEquals(3, output);
     }

src/test/java/io/github/cadenceoss/iwf/integ/signal/BasicSignalWorkflow.java

@@ -10,7 +10,10 @@
 public class BasicSignalWorkflow implements Workflow {
     @Override
     public List<SignalChannelDef> getSignalChannels() {
-        return Arrays.asList(SignalChannelDef.create(Integer.class, BasicSignalWorkflowState1.SIGNAL_CHANNEL_NAME));
+        return Arrays.asList(
+                SignalChannelDef.create(Integer.class, BasicSignalWorkflowState1.SIGNAL_CHANNEL_NAME_1),
+                SignalChannelDef.create(Integer.class, BasicSignalWorkflowState1.SIGNAL_CHANNEL_NAME_2)
+        );
     }
 
     @Override

src/test/java/io/github/cadenceoss/iwf/integ/signal/BasicSignalWorkflowState1.java

@@ -11,10 +11,13 @@
 import io.github.cadenceoss.iwf.core.command.InterStateChannel;
 import io.github.cadenceoss.iwf.core.command.SignalCommand;
 import io.github.cadenceoss.iwf.core.command.SignalCommandResult;
+import io.github.cadenceoss.iwf.gen.models.SignalResult;
 
 public class BasicSignalWorkflowState1 implements WorkflowState<Integer> {
     public static final String STATE_ID = "signal-s1";
-    public static final String SIGNAL_CHANNEL_NAME = "test-signal";
+    public static final String SIGNAL_CHANNEL_NAME_1 = "test-signal-1";
+
+    public static final String SIGNAL_CHANNEL_NAME_2 = "test-signal-2";
     public static final String COMMAND_ID = "test-signal-id";
 
     @Override
@@ -34,7 +37,10 @@ public CommandRequest start(
             StateLocal stateLocals,
             SearchAttributesRW searchAttributes,
             QueryAttributesRW queryAttributes, final InterStateChannel interStateChannel) {
-        return CommandRequest.forAllCommandCompleted(SignalCommand.create(COMMAND_ID, SIGNAL_CHANNEL_NAME));
+        return CommandRequest.forAnyCommandCompleted(
+                SignalCommand.create(COMMAND_ID, SIGNAL_CHANNEL_NAME_1),
+                SignalCommand.create(COMMAND_ID, SIGNAL_CHANNEL_NAME_2)
+        );
     }
 
     @Override
@@ -46,7 +52,12 @@ public StateDecision decide(
             SearchAttributesRW searchAttributes,
             QueryAttributesRW queryAttributes, final InterStateChannel interStateChannel) {
         SignalCommandResult signalCommandResult = commandResults.getAllSignalCommandResults().get(0);
-        Integer output = input + (Integer) signalCommandResult.getSignalValue();
+        Integer output = input + (Integer) signalCommandResult.getSignalValue().get();
+
+        SignalCommandResult signalCommandResult2 = commandResults.getAllSignalCommandResults().get(1);
+        if (signalCommandResult2.getSignalRequestStatusEnum() != SignalResult.SignalRequestStatusEnum.WAITING) {
+            throw new RuntimeException("the second signal should be waiting");
+        }
         return StateDecision.gracefulCompleteWorkflow(output);
     }
 }