WorkflowInstanceDao { public List<Long> pollNextWorkflowInstanceIds(final int batchSize) { if (sqlVariants.hasUpdateReturning()) { return pollNextWorkflowInstanceIdsWithUpdateReturning(batchSize); } return pollNextWorkflowInstanceIdsWithTransaction(batchSize); } @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 pollNextWorkflowInstancesReturnInstancesInCorrectOrder() { long olderLowPrio = createInstance(2, (short) 1); long newerLowPrio = createInstance(1, (short) 1); long newerHighPrio = createInstance(1, (short) 2); List<Long> ids = dao.pollNextWorkflowInstanceIds(1); assertThat(ids, is(asList(newerHighPrio))); ids = dao.pollNextWorkflowInstanceIds(1); assertThat(ids, is(asList(olderLowPrio))); ids = dao.pollNextWorkflowInstanceIds(1); assertThat(ids, is(asList(newerLowPrio))); }
@Test public void fakePostgreSQLpollNextWorkflowInstances() { JdbcTemplate j = mock(JdbcTemplate.class); WorkflowInstanceDao d = preparePostgreSQLDao(j); ArgumentCaptor<String> sql = ArgumentCaptor.forClass(String.class); when(j.queryForList(sql.capture(), eq(Long.class))).thenReturn(asList(1L, 2L, 3L)); assertThat(d.pollNextWorkflowInstanceIds(5), is(asList(1L, 2L, 3L))); assertEquals( "update nflow_workflow set executor_id = 42, status = 'executing'::workflow_status, external_next_activation = null where id in (select id from nflow_workflow where executor_id is null and status in ('created'::workflow_status, 'inProgress'::workflow_status) and next_activation <= current_timestamp and group matches order by priority desc, next_activation asc limit 5 for update skip locked) and executor_id is null returning id", sql.getValue()); }