SimpleEventProcessor implements EventProcessor { public Map<String, String> getQueues() { Map<String, String> queues = new HashMap<>(); eventToQueueMap.forEach((key, value) -> queues.put(key, value.getName())); return queues; } @Inject  SimpleEventProcessor(ExecutionService executionService,
                                MetadataService metadataService,
                                ActionProcessor actionProcessor,
                                EventQueues eventQueues,
                                JsonUtils jsonUtils,
                                Configuration configuration,
                                ObjectMapper objectMapper); Map<String, String> getQueues(); Map<String, Map<String, Long>> getQueueSizes();  }
@Test public void testEventProcessor() { EventHandler eventHandler = new EventHandler(); eventHandler.setName(UUID.randomUUID().toString()); eventHandler.setActive(true); Map<String, String> taskToDomain = new HashMap<>(); taskToDomain.put("*", "dev"); Action startWorkflowAction = new Action(); startWorkflowAction.setAction(Type.start_workflow); startWorkflowAction.setStart_workflow(new StartWorkflow()); startWorkflowAction.getStart_workflow().setName("workflow_x"); startWorkflowAction.getStart_workflow().setVersion(1); startWorkflowAction.getStart_workflow().setTaskToDomain(taskToDomain); eventHandler.getActions().add(startWorkflowAction); Action completeTaskAction = new Action(); completeTaskAction.setAction(Type.complete_task); completeTaskAction.setComplete_task(new TaskDetails()); completeTaskAction.getComplete_task().setTaskRefName("task_x"); completeTaskAction.getComplete_task().setWorkflowId(UUID.randomUUID().toString()); completeTaskAction.getComplete_task().setOutput(new HashMap<>()); eventHandler.getActions().add(completeTaskAction); eventHandler.setEvent(event); when(metadataService.getAllEventHandlers()).thenReturn(Collections.singletonList(eventHandler)); when(metadataService.getEventHandlersForEvent(event, true)).thenReturn(Collections.singletonList(eventHandler)); when(executionService.addEventExecution(any())).thenReturn(true); when(queue.rePublishIfNoAck()).thenReturn(false); String id = UUID.randomUUID().toString(); AtomicBoolean started = new AtomicBoolean(false); doAnswer((Answer<String>) invocation -> { started.set(true); return id; }).when(workflowExecutor).startWorkflow(eq(startWorkflowAction.getStart_workflow().getName()), eq(startWorkflowAction.getStart_workflow().getVersion()), eq(startWorkflowAction.getStart_workflow().getCorrelationId()), anyMap(), eq(null), eq(event), anyMap()); AtomicBoolean completed = new AtomicBoolean(false); doAnswer((Answer<String>) invocation -> { completed.set(true); return null; }).when(workflowExecutor).updateTask(any()); Task task = new Task(); task.setReferenceTaskName(completeTaskAction.getComplete_task().getTaskRefName()); Workflow workflow = new Workflow(); workflow.setTasks(Collections.singletonList(task)); when(workflowExecutor.getWorkflow(completeTaskAction.getComplete_task().getWorkflowId(), true)).thenReturn(workflow); WorkflowDef workflowDef = new WorkflowDef(); workflowDef.setVersion(startWorkflowAction.getStart_workflow().getVersion()); workflowDef.setName(startWorkflowAction.getStart_workflow().getName()); when(metadataService.getWorkflowDef(any(), any())).thenReturn(workflowDef); SimpleActionProcessor actionProcessor = new SimpleActionProcessor(workflowExecutor, parametersUtils, jsonUtils); SimpleEventProcessor eventProcessor = new SimpleEventProcessor(executionService, metadataService, actionProcessor, eventQueues, jsonUtils, new TestConfiguration(), objectMapper); assertNotNull(eventProcessor.getQueues()); assertEquals(1, eventProcessor.getQueues().size()); String queueEvent = eventProcessor.getQueues().keySet().iterator().next(); assertEquals(eventHandler.getEvent(), queueEvent); String eventProcessorQueue = eventProcessor.getQueues().values().iterator().next(); assertEquals(queueURI, eventProcessorQueue); Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); assertTrue(started.get()); assertTrue(completed.get()); verify(queue, atMost(1)).ack(any()); verify(queue, never()).publish(any()); }
@Test public void testEventHandlerWithCondition() { EventHandler eventHandler = new EventHandler(); eventHandler.setName("cms_intermediate_video_ingest_handler"); eventHandler.setActive(true); eventHandler.setEvent("sqs:dev_cms_asset_ingest_queue"); eventHandler.setCondition("$.Message.testKey1 == 'level1' && $.Message.metadata.testKey2 == 123456"); Map<String, Object> startWorkflowInput = new LinkedHashMap<>(); startWorkflowInput.put("param1", "${Message.metadata.testKey2}"); startWorkflowInput.put("param2", "SQS-${MessageId}"); Action startWorkflowAction = new Action(); startWorkflowAction.setAction(Type.start_workflow); startWorkflowAction.setStart_workflow(new StartWorkflow()); startWorkflowAction.getStart_workflow().setName("cms_artwork_automation"); startWorkflowAction.getStart_workflow().setVersion(1); startWorkflowAction.getStart_workflow().setInput(startWorkflowInput); startWorkflowAction.setExpandInlineJSON(true); eventHandler.getActions().add(startWorkflowAction); eventHandler.setEvent(event); when(metadataService.getAllEventHandlers()).thenReturn(Collections.singletonList(eventHandler)); when(metadataService.getEventHandlersForEvent(event, true)).thenReturn(Collections.singletonList(eventHandler)); when(executionService.addEventExecution(any())).thenReturn(true); when(queue.rePublishIfNoAck()).thenReturn(false); String id = UUID.randomUUID().toString(); AtomicBoolean started = new AtomicBoolean(false); doAnswer((Answer<String>) invocation -> { started.set(true); return id; }).when(workflowExecutor).startWorkflow(eq(startWorkflowAction.getStart_workflow().getName()), eq(startWorkflowAction.getStart_workflow().getVersion()), eq(startWorkflowAction.getStart_workflow().getCorrelationId()), anyMap(), eq(null), eq(event), eq(null)); WorkflowDef workflowDef = new WorkflowDef(); workflowDef.setName(startWorkflowAction.getStart_workflow().getName()); when(metadataService.getWorkflowDef(any(), any())).thenReturn(workflowDef); SimpleActionProcessor actionProcessor = new SimpleActionProcessor(workflowExecutor, parametersUtils, jsonUtils); SimpleEventProcessor eventProcessor = new SimpleEventProcessor(executionService, metadataService, actionProcessor, eventQueues, jsonUtils, new TestConfiguration(), objectMapper); assertNotNull(eventProcessor.getQueues()); assertEquals(1, eventProcessor.getQueues().size()); Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); assertTrue(started.get()); }