WorkflowInstanceDao { @SuppressWarnings("null") public void updateWorkflowInstanceAfterExecution(WorkflowInstance instance, WorkflowInstanceAction action, List<WorkflowInstance> childWorkflows, List<WorkflowInstance> workflows, boolean createAction) { Assert.isTrue(action != null, "action can not be null"); Assert.isTrue(childWorkflows != null, "childWorkflows can not be null"); Assert.isTrue(workflows != null, "workflows can not be null"); Map<String, String> changedStateVariables = instance.getChangedStateVariables(); if (!createAction && (!childWorkflows.isEmpty() || !workflows.isEmpty() || !changedStateVariables.isEmpty())) { logger.info("Forcing action creation because new workflow instances are created or state variables are changed."); createAction = true; } if (createAction) { if (sqlVariants.hasUpdateableCTE() && childWorkflows.isEmpty() && workflows.isEmpty()) { updateWorkflowInstanceWithCTE(instance, action, changedStateVariables); } else { updateWorkflowInstanceWithTransaction(instance, action, childWorkflows, workflows, changedStateVariables); } } else { updateWorkflowInstance(instance); } } @Inject  WorkflowInstanceDao(SQLVariants sqlVariants, @NFlow JdbcTemplate nflowJdbcTemplate,
      @NFlow TransactionTemplate transactionTemplate, @NFlow NamedParameterJdbcTemplate nflowNamedParameterJdbcTemplate,
      ExecutorDao executorDao, WorkflowInstanceExecutor workflowInstanceExecutor, WorkflowInstanceFactory workflowInstanceFactory,
      Environment env); long insertWorkflowInstance(WorkflowInstance instance); @SuppressWarnings("null") void updateWorkflowInstanceAfterExecution(WorkflowInstance instance, WorkflowInstanceAction action,
      List<WorkflowInstance> childWorkflows, List<WorkflowInstance> workflows, boolean createAction); int updateWorkflowInstance(WorkflowInstance instance); void recoverWorkflowInstancesFromDeadNodes(); void checkStateVariableValueLength(String name, String value); boolean updateNotRunningWorkflowInstance(WorkflowInstance instance); boolean wakeUpWorkflowExternally(long workflowInstanceId, List<String> expectedStates); boolean wakeupWorkflowInstanceIfNotExecuting(long workflowInstanceId, List<String> expectedStates); WorkflowInstance getWorkflowInstance(long id, Set<WorkflowInstanceInclude> includes, Long maxActions); List<Long> pollNextWorkflowInstanceIds(final int batchSize); List<WorkflowInstance> queryWorkflowInstances(QueryWorkflowInstances query); Stream<WorkflowInstance> queryWorkflowInstancesAsStream(QueryWorkflowInstances query); @Transactional(propagation = MANDATORY) long insertWorkflowInstanceAction(final WorkflowInstance instance, final WorkflowInstanceAction action); long insertWorkflowInstanceAction(final WorkflowInstanceAction action); String getWorkflowInstanceState(long workflowInstanceId); Optional<Integer> getSignal(long workflowInstanceId); @Transactional boolean setSignal(long workflowInstanceId, Optional<Integer> signal, String reason, WorkflowActionType actionType); String getWorkflowInstanceType(long workflowInstanceId);  }
@Test public void updateWorkflowInstanceDoesNotCreateActionWhenCreateActionIsFalse() { WorkflowInstance instance = updateInstanceBuilder().build(); WorkflowInstanceAction action = constructActionBuilder(instance.id).build(); dao.updateWorkflowInstanceAfterExecution(instance, action, noChildWorkflows, emptyWorkflows, false); assertThat(updatedInstance().actions.isEmpty(), is(true)); }
@Test public void updateWorkflowInstanceCreatesActionWhenCreateActionIsTrue() { WorkflowInstance instance = updateInstanceBuilder().build(); WorkflowInstanceAction action = constructActionBuilder(instance.id).build(); dao.updateWorkflowInstanceAfterExecution(instance, action, noChildWorkflows, emptyWorkflows, true); assertThat(updatedInstance().actions.size(), is(1)); }
@Test public void updateWorkflowInstanceCreatesActionWhenChildWorkflowIsCreated() { WorkflowInstance instance = updateInstanceBuilder().build(); WorkflowInstanceAction action = constructActionBuilder(instance.id).build(); WorkflowInstance childWorkflow = constructWorkflowInstanceBuilder().build(); dao.updateWorkflowInstanceAfterExecution(instance, action, asList(childWorkflow), emptyWorkflows, false); assertThat(updatedInstance().actions.size(), is(1)); }
@Test public void updateWorkflowInstanceCreatesActionWhenNewWorkflowIsCreated() { WorkflowInstance instance = updateInstanceBuilder().build(); WorkflowInstanceAction action = constructActionBuilder(instance.id).build(); WorkflowInstance newWorkflow = constructWorkflowInstanceBuilder().build(); dao.updateWorkflowInstanceAfterExecution(instance, action, noChildWorkflows, asList(newWorkflow), false); assertThat(updatedInstance().actions.size(), is(1)); }
@Test public void updateWorkflowInstanceCreatesActionWhenStateVariablesAreModified() { WorkflowInstance instance = updateInstanceBuilder().putStateVariable("foo", "bar").build(); WorkflowInstanceAction action = constructActionBuilder(instance.id).build(); dao.updateWorkflowInstanceAfterExecution(instance, action, noChildWorkflows, emptyWorkflows, false); assertThat(updatedInstance().actions.size(), is(1)); }
@Test public void fakePostgreSQLupdateWorkflowInstance() { JdbcTemplate j = mock(JdbcTemplate.class); WorkflowInstanceDao d = preparePostgreSQLDao(j); ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class); ArgumentCaptor<Object> args = ArgumentCaptor.forClass(Object.class); when(j.queryForObject(sql.capture(), eq(Long.class), args.capture())).thenReturn(42L); DateTime started = now(); WorkflowInstance i2 = new WorkflowInstance.Builder().setState("updateState").setStateText("update text") .setNextActivation(started.plusSeconds(1)).setStatus(executing).setRetries(3).setId(43).putStateVariable("A", "B") .build(); WorkflowInstanceAction a1 = new WorkflowInstanceAction.Builder().setExecutionStart(started).setExecutorId(4) .setExecutionEnd(started.plusMillis(100)).setRetryNo(1).setType(externalChange).setState("test") .setStateText("state text").setWorkflowInstanceId(43).build(); d.updateWorkflowInstanceAfterExecution(i2, a1, noChildWorkflows, emptyWorkflows, false); assertEquals("with wf as (update nflow_workflow set status = ?::workflow_status, state = ?, state_text = ?, " + "next_activation = (case when ?::timestamptz is null then null when external_next_activation is null then " + "?::timestamptz else least(?::timestamptz, external_next_activation) end), external_next_activation = null, " + "executor_id = ?, retries = ?, started = (case when started is null then ? else started end) where id = ? " + "and executor_id = 42 returning id), " + "act as (insert into nflow_workflow_action(workflow_id, executor_id, type, state, state_text, retry_no, " + "execution_start, execution_end) select wf.id, ?, ?::action_type, ?, ?, ?, ?, ? from wf returning id), " + "ins17 as (insert into nflow_workflow_state(workflow_id, action_id, state_key, state_value) " + "select wf.id,act.id,?,? from wf,act) select act.id from act", sql.getValue()); assertThat(args.getAllValues().size(), is(countMatches(sql.getValue(), "?"))); int i = 0; assertThat(args.getAllValues().get(i++), is((Object) i2.status.name())); assertThat(args.getAllValues().get(i++), is((Object) i2.state)); assertThat(args.getAllValues().get(i++), is((Object) i2.stateText)); assertThat(args.getAllValues().get(i++), is((Object) new Timestamp(i2.nextActivation.getMillis()))); assertThat(args.getAllValues().get(i++), is((Object) new Timestamp(i2.nextActivation.getMillis()))); assertThat(args.getAllValues().get(i++), is((Object) new Timestamp(i2.nextActivation.getMillis()))); assertThat(args.getAllValues().get(i++), is((Object) 42)); assertThat(args.getAllValues().get(i++), is((Object) i2.retries)); assertThat(args.getAllValues().get(i++), is((Object) new Timestamp(a1.executionStart.getMillis()))); assertThat(args.getAllValues().get(i++), is((Object) i2.id)); assertThat(args.getAllValues().get(i++), is((Object) 42)); assertThat(args.getAllValues().get(i++), is((Object) a1.type.name())); assertThat(args.getAllValues().get(i++), is((Object) a1.state)); assertThat(args.getAllValues().get(i++), is((Object) a1.stateText)); assertThat(args.getAllValues().get(i++), is((Object) a1.retryNo)); assertThat(args.getAllValues().get(i++), is((Object) new Timestamp(a1.executionStart.getMillis()))); assertThat(args.getAllValues().get(i++), is((Object) new Timestamp(a1.executionEnd.getMillis()))); assertThat(args.getAllValues().get(i++), is((Object) "A")); assertThat(args.getAllValues().get(i++), is((Object) "B")); }