WorkFlowExecutionController { public Set<State> postEvent(EventData eventData, String stateMachineInstanceId) { Event event = persistEvent(stateMachineInstanceId, eventData); return processEvent(event, stateMachineInstanceId); } @Inject  WorkFlowExecutionController(EventsDAO eventsDAO, StateMachinesDAO stateMachinesDAO,
                                       StatesDAO statesDAO, AuditDAO auditDAO, ExecutionNodeTaskDispatcher executionNodeTaskDispatcher,
                                       RedriverRegistry redriverRegistry, MetricsClient metricsClient,
                                       ClientElbPersistenceService clientElbPersistenceService); Set<State> initAndStart(StateMachine stateMachine); void updateTaskStatusAndPostEvent(StateMachine stateMachine, EventAndExecutionData eventAndExecutionData); void updateTaskStatusAndHandlePathCancellation(String stateMachineId, EventAndExecutionData eventAndExecutionData); void handlePathCancellation(String stateMachineId, EventData eventData); Set<State> postEvent(EventData eventData, String stateMachineInstanceId); @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) Event persistEvent(String stateMachineInstanceId, EventData eventData); Set<State> processEvent(Event event, String stateMachineInstanceId); @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) void updateTaskStatus(String machineId, Long stateId, ExecutionUpdateData executionUpdateData); void updateExecutionStatus(String stateMachineId, Long taskId, Status status, long retryCount, long currentRetryCount, String errorMessage, boolean deleteFromRedriver); @Transactional @SelectDataSource(type = DataSourceType.READ_WRITE, storage = Storage.SHARDED) void updateEventData(String machineId, EventData eventData); void unsidelineState(String stateMachineId, Long stateId); void incrementExecutionRetries(String stateMachineId, Long taskId); void redriveTask(String machineId, Long taskId); void cancelWorkflow(StateMachine stateMachine); static String getRouterName(String taskName);  }
@Test public void testEventPost_shouldNotFetchEventDataFromDBIfStateIsDependantOnSingleEvent() throws Exception { final EventData testEventData = new EventData("event1", "foo", "someStringData", "runtime"); when(eventsDAO.findBySMIdAndName("standard-machine", "event1")).thenReturn(new Event("event1", "foo", Event.EventStatus.pending, "1", null, null)); when(eventsDAO.findTriggeredOrCancelledEventsNamesBySMId("standard-machine")).thenReturn(Collections.singletonList("event1")); workFlowExecutionController.postEvent(testEventData, "standard-machine"); verify(eventsDAO, times(0)).findByEventNamesAndSMId("standard-machine", Collections.singletonList("event1")); }
@Test public void testEventPost_taskRedriveDelay() throws Exception { final EventData testEventData1 = new EventData("event1", "java.lang.String", "42", "runtime"); when(eventsDAO.findBySMIdAndName("standard-machine", "event1")).thenReturn(new Event("event1", "java.lang.String", Event.EventStatus.pending, "1", null, null)); EventData[] expectedEvents1 = new EventData[]{new EventData("event1", "java.lang.String", "42", "runtime")}; when(eventsDAO.findByEventNamesAndSMId("standard-machine", Collections.singletonList("event1"))).thenReturn(Arrays.asList(expectedEvents1)); when(eventsDAO.findTriggeredOrCancelledEventsNamesBySMId("standard-machine")).thenReturn(Collections.singletonList("event1")); workFlowExecutionController.postEvent(testEventData1, "standard-machine"); final EventData testEventData0 = new EventData("event0", "java.lang.String", "42", "runtime"); when(eventsDAO.findBySMIdAndName("standard-machine", "event0")).thenReturn(new Event("event0", "java.lang.String", Event.EventStatus.pending, "1", null, null)); EventData[] expectedEvents0 = new EventData[]{new EventData("event0", "java.lang.String", "42", "runtime")}; when(eventsDAO.findByEventNamesAndSMId("standard-machine", Collections.singletonList("event0"))).thenReturn(Arrays.asList(expectedEvents0)); when(eventsDAO.findTriggeredOrCancelledEventsNamesBySMId("standard-machine")).thenReturn(Collections.singletonList("event0")); workFlowExecutionController.postEvent(testEventData0, "standard-machine"); Thread.sleep(2000); verify(redriverRegistry).registerTask(2L, "standard-machine", 32800); verify(redriverRegistry).registerTask(4L, "standard-machine", 8400); }
@Test public void testEventPost_shouldNotSendExecuteTaskIfItIsAlreadyCompleted() throws Exception { final EventData testEventData = new EventData("event0", "java.lang.String", "42", "runtime"); when(eventsDAO.findBySMIdAndName("standard-machine", "event0")).thenReturn(new Event("event0", "java.lang.String", Event.EventStatus.pending, "1", null, null)); EventData[] expectedEvents = new EventData[]{new EventData("event0", "java.lang.String", "42", "runtime")}; when(eventsDAO.findTriggeredOrCancelledEventsNamesBySMId("standard-machine")).thenReturn(Collections.singletonList("event0")); when(executionNodeTaskDispatcher.forwardExecutionMessage(anyString(), anyObject())).thenReturn(Response.Status.ACCEPTED.getStatusCode()); workFlowExecutionController.postEvent(testEventData, "standard-machine"); State state = stateMachinesDAO.findById("standard-machine").getStates().stream().filter((s) -> s.getId() == 4L).findFirst().orElse(null); state.setStatus(Status.completed); workFlowExecutionController.postEvent(testEventData, "standard-machine"); verify(executionNodeTaskDispatcher, times(1)).forwardExecutionMessage(anyString(), anyObject()); verifyNoMoreInteractions(executionNodeTaskDispatcher); }
@Test public void testEventPost_shouldSendExecuteTaskIfItIsNotCompleted() throws Exception { final EventData testEventData = new EventData("event0", "java.lang.String", "42", "runtime"); when(eventsDAO.findBySMIdAndName("standard-machine", "event0")).thenReturn(new Event("event0", "java.lang.String", Event.EventStatus.pending, "1", null, null)); EventData[] expectedEvents = new EventData[]{new EventData("event0", "java.lang.String", "42", "runtime")}; when(eventsDAO.findTriggeredOrCancelledEventsNamesBySMId("standard-machine")).thenReturn(Collections.singletonList("event0")); when(executionNodeTaskDispatcher.forwardExecutionMessage(anyString(), anyObject())).thenReturn(Response.Status.ACCEPTED.getStatusCode()); workFlowExecutionController.postEvent(testEventData, "standard-machine"); StateMachine stateMachine = stateMachinesDAO.findById("standard-machine"); State state = stateMachinesDAO.findById("standard-machine").getStates().stream().filter((s) -> s.getId() == 4L).findFirst().orElse(null); state.setStatus(Status.errored); workFlowExecutionController.postEvent(testEventData, "standard-machine"); verify(executionNodeTaskDispatcher, times(2)).forwardExecutionMessage(anyString(), anyObject()); verifyNoMoreInteractions(executionNodeTaskDispatcher); }
@Test public void testEventPost_shouldNotSendExecuteTaskIfItIsCancelled() throws Exception { final EventData testEventData = new EventData("event0", "java.lang.String", "42", "runtime"); when(eventsDAO.findBySMIdAndName("standard-machine", "event0")).thenReturn(new Event("event0", "java.lang.String", Event.EventStatus.pending, "1", null, null)); EventData[] expectedEvents = new EventData[]{testEventData}; when(eventsDAO.findTriggeredOrCancelledEventsNamesBySMId("standard-machine")).thenReturn(Collections.singletonList("event0")); when(executionNodeTaskDispatcher.forwardExecutionMessage(anyString(), anyObject())).thenReturn(Response.Status.ACCEPTED.getStatusCode()); workFlowExecutionController.postEvent(testEventData, "standard-machine"); StateMachine stateMachine = stateMachinesDAO.findById("standard-machine"); State state = stateMachine.getStates().stream().filter((s) -> s.getId() == 4L).findFirst().orElse(null); state.setStatus(Status.cancelled); workFlowExecutionController.postEvent(testEventData, "standard-machine"); verify(executionNodeTaskDispatcher, times(1)).forwardExecutionMessage(anyString(), anyObject()); }