Event extends WorkflowSystemTask { @VisibleForTesting ObservableQueue getQueue(Workflow workflow, Task task) { if (task.getInputData().get("sink") == null) { task.setStatus(Status.FAILED); task.setReasonForIncompletion("No sink specified in task"); return null; } String sinkValueRaw = (String)task.getInputData().get("sink"); Map<String, Object> input = new HashMap<>(); input.put("sink", sinkValueRaw); Map<String, Object> replaced = parametersUtils.getTaskInputV2(input, workflow, task.getTaskId(), null); String sinkValue = (String)replaced.get("sink"); String queueName = sinkValue; if(sinkValue.startsWith("conductor")) { if("conductor".equals(sinkValue)) { queueName = sinkValue + ":" + workflow.getWorkflowName() + ":" + task.getReferenceTaskName(); } else if(sinkValue.startsWith("conductor:")) { queueName = sinkValue.replaceAll("conductor:", ""); queueName = "conductor:" + workflow.getWorkflowName() + ":" + queueName; } else { task.setStatus(Status.FAILED); task.setReasonForIncompletion("Invalid / Unsupported sink specified: " + sinkValue); return null; } } task.getOutputData().put("event_produced", queueName); try { return eventQueues.getQueue(queueName); } catch(IllegalArgumentException e) { logger.error("Error setting up queue: {} for task:{}, workflow:{}", queueName, task.getTaskId(), workflow.getWorkflowId(), e); task.setStatus(Status.FAILED); task.setReasonForIncompletion("Error when trying to access the specified queue/topic: " + sinkValue + ", error: " + e.getMessage()); return null; } } @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 testSinkParam() { String sink = "sqs:queue_name"; WorkflowDef def = new WorkflowDef(); def.setName("wf0"); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(def); Task task1 = new Task(); task1.setReferenceTaskName("t1"); task1.getOutputData().put("q", "t1_queue"); workflow.getTasks().add(task1); Task task2 = new Task(); task2.setReferenceTaskName("t2"); task2.getOutputData().put("q", "task2_queue"); workflow.getTasks().add(task2); Task task = new Task(); task.setReferenceTaskName("event"); task.getInputData().put("sink", sink); task.setTaskType(TaskType.EVENT.name()); workflow.getTasks().add(task); Event event = new Event(eventQueues, parametersUtils, objectMapper); ObservableQueue queue = event.getQueue(workflow, task); assertNotNull(task.getReasonForIncompletion(), queue); assertEquals("queue_name", queue.getName()); assertEquals("sqs", queue.getType()); sink = "sqs:${t1.output.q}"; task.getInputData().put("sink", sink); queue = event.getQueue(workflow, task); assertNotNull(queue); assertEquals("t1_queue", queue.getName()); assertEquals("sqs", queue.getType()); System.out.println(task.getOutputData().get("event_produced")); sink = "sqs:${t2.output.q}"; task.getInputData().put("sink", sink); queue = event.getQueue(workflow, task); assertNotNull(queue); assertEquals("task2_queue", queue.getName()); assertEquals("sqs", queue.getType()); System.out.println(task.getOutputData().get("event_produced")); sink = "conductor"; task.getInputData().put("sink", sink); queue = event.getQueue(workflow, task); assertNotNull(queue); assertEquals(workflow.getWorkflowName() + ":" + task.getReferenceTaskName(), queue.getName()); assertEquals("conductor", queue.getType()); System.out.println(task.getOutputData().get("event_produced")); sink = "sqs:static_value"; task.getInputData().put("sink", sink); queue = event.getQueue(workflow, task); assertNotNull(queue); assertEquals("static_value", queue.getName()); assertEquals("sqs", queue.getType()); assertEquals(sink, task.getOutputData().get("event_produced")); System.out.println(task.getOutputData().get("event_produced")); sink = "bad:queue"; task.getInputData().put("sink", sink); queue = event.getQueue(workflow, task); assertNull(queue); assertEquals(Task.Status.FAILED, task.getStatus()); }
@Test public void testDynamicSinks() { Event event = new Event(eventQueues, parametersUtils, objectMapper); Workflow workflow = new Workflow(); workflow.setWorkflowDefinition(testWorkflowDefinition); Task task = new Task(); task.setReferenceTaskName("task0"); task.setTaskId("task_id_0"); task.setStatus(Status.IN_PROGRESS); task.getInputData().put("sink", "conductor:some_arbitary_queue"); ObservableQueue queue = event.getQueue(workflow, task); assertEquals(Task.Status.IN_PROGRESS, task.getStatus()); assertNotNull(queue); assertEquals("testWorkflow:some_arbitary_queue", queue.getName()); assertEquals("testWorkflow:some_arbitary_queue", queue.getURI()); assertEquals("conductor", queue.getType()); assertEquals("conductor:testWorkflow:some_arbitary_queue", task.getOutputData().get("event_produced")); task.getInputData().put("sink", "conductor"); queue = event.getQueue(workflow, task); assertEquals("not in progress: " + task.getReasonForIncompletion(), Task.Status.IN_PROGRESS, task.getStatus()); assertNotNull(queue); assertEquals("testWorkflow:task0", queue.getName()); task.getInputData().put("sink", "sqs:my_sqs_queue_name"); queue = event.getQueue(workflow, task); assertEquals("not in progress: " + task.getReasonForIncompletion(), Task.Status.IN_PROGRESS, task.getStatus()); assertNotNull(queue); assertEquals("my_sqs_queue_name", queue.getName()); assertEquals("sqs", queue.getType()); task.getInputData().put("sink", "sns:my_sqs_queue_name"); queue = event.getQueue(workflow, task); assertEquals(Task.Status.FAILED, task.getStatus()); }