Event extends WorkflowSystemTask { @Override public void start(Workflow workflow, Task task, WorkflowExecutor provider) { Map<String, Object> payload = new HashMap<>(task.getInputData()); payload.put("workflowInstanceId", workflow.getWorkflowId()); payload.put("workflowType", workflow.getWorkflowName()); payload.put("workflowVersion", workflow.getWorkflowVersion()); payload.put("correlationId", workflow.getCorrelationId()); String payloadJson; try { payloadJson = objectMapper.writeValueAsString(payload); } catch (JsonProcessingException e) { String msg = String.format("Error serializing JSON payload for task: %s, workflow: %s", task.getTaskId(), workflow.getWorkflowId()); throw new ApplicationException(INTERNAL_ERROR, msg); } Message message = new Message(task.getTaskId(), payloadJson, task.getTaskId()); ObservableQueue queue = getQueue(workflow, task); if(queue != null) { queue.publish(Collections.singletonList(message)); logger.debug("Published message:{} to queue:{}", message.getId(), queue.getName()); task.getOutputData().putAll(payload); if (isAsyncComplete(task)) { task.setStatus(Status.IN_PROGRESS); } else { task.setStatus(Status.COMPLETED); } } else { task.setReasonForIncompletion("No queue found to publish."); task.setStatus(Status.FAILED); } } @Inject  Event(EventQueues eventQueues, ParametersUtils parametersUtils, ObjectMapper objectMapper); @Override void start(Workflow workflow, Task task, WorkflowExecutor provider); @Override boolean execute(Workflow workflow, Task task, WorkflowExecutor provider); @Override void cancel(Workflow workflow, Task task, WorkflowExecutor provider); @Override boolean isAsync(); static final String NAME; }
@Test public void testAsyncComplete() throws Exception { Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(testWorkflowDefinition); Task task = new Task(); task.getInputData().put("sink", "conductor"); task.getInputData().put("asyncComplete", true); task.setReferenceTaskName("task0"); task.setTaskId("task_id_0"); QueueDAO dao = mock(QueueDAO.class); String[] publishedQueue = new String[1]; List<Message> publishedMessages = new LinkedList<>(); doAnswer((Answer<Void>) invocation -> { String queueName = invocation.getArgument(0, String.class); System.out.println(queueName); publishedQueue[0] = queueName; List<Message> messages = invocation.getArgument(1, List.class); publishedMessages.addAll(messages); return null; }).when(dao).push(any(), any()); doAnswer((Answer<List<String>>) invocation -> { String messageId = invocation.getArgument(1, String.class); if(publishedMessages.get(0).getId().equals(messageId)) { publishedMessages.remove(0); return Collections.singletonList(messageId); } return null; }).when(dao).remove(any(), any()); Map<String, EventQueueProvider> providers = new HashMap<>(); providers.put("conductor", new DynoEventQueueProvider(dao, new TestConfiguration(), Schedulers.from(Executors.newSingleThreadExecutor()))); eventQueues = new EventQueues(providers, parametersUtils); Event event = new Event(eventQueues, parametersUtils, objectMapper); event.start(workflow, task, null); assertEquals(Status.IN_PROGRESS, task.getStatus()); assertNotNull(task.getOutputData()); assertEquals("conductor:" + workflow.getWorkflowName() + ":" + task.getReferenceTaskName(), task.getOutputData().get("event_produced")); assertEquals(task.getOutputData().get("event_produced"), "conductor:" + publishedQueue[0]); assertEquals(1, publishedMessages.size()); assertEquals(task.getTaskId(), publishedMessages.get(0).getId()); assertNotNull(publishedMessages.get(0).getPayload()); }