QueueListenerImpl implements QueueListener { @Override public void onTerminated(List<ExecutionMessage> messages) { ScoreEvent[] scoreEvents = handleTerminatedMessages(messages); if (scoreEvents.length > 0) { try { eventBus.dispatch(scoreEvents); } catch (InterruptedException e) { logger.error("Thread is interrupted. Ignoring... ", e); } } }  @Override void prePersist(List<ExecutionMessage> messages); @Override void onEnqueue(List<ExecutionMessage> messages, int queueSize); @Override void onPoll(List<ExecutionMessage> messages, int queueSize); @Override void onTerminated(List<ExecutionMessage> messages); @Override void onFailed(List<ExecutionMessage> messages); @Override void onPersistMessage(List<ExecutionMessage> messages);  }
@Test public void testOnTerminatedWhenNoMessages() throws InterruptedException { List<ExecutionMessage> messages = new ArrayList<>(); queueListener.onTerminated(messages); verify(eventBus, never()).dispatch(); }
@Test public void testEventsThrownForOnTerminated() throws InterruptedException { List<ExecutionMessage> messages = new ArrayList<>(); messages.add(createExecutionMessage()); messages.add(createExecutionMessage()); ScoreEvent event1 = new ScoreEvent("type1", "event1"); ScoreEvent event2 = new ScoreEvent("type1", "event2"); when(scoreEventFactory.createFinishedEvent(any(Execution.class))).thenReturn(event1, event2); queueListener.onTerminated(messages); verify(eventBus, times(1)).dispatch(event1, event2); }
@Test public void testOnTerminatedNonBranchExecution() { List<ExecutionMessage> messages = new ArrayList<>(); messages.add(createExecutionMessage()); messages.add(createExecutionMessage()); ScoreEvent event1 = new ScoreEvent("type1", "event1"); ScoreEvent event2 = new ScoreEvent("type1", "event2"); when(scoreEventFactory.createFinishedEvent(any(Execution.class))).thenReturn(event1, event2); queueListener.onTerminated(messages); verify(executionStateService, times(1)).deleteExecutionState(Long.valueOf(messages.get(0).getMsgId()), ExecutionSummary.EMPTY_BRANCH); verify(executionStateService, times(1)).deleteExecutionState(Long.valueOf(messages.get(1).getMsgId()), ExecutionSummary.EMPTY_BRANCH); }
@Test public void testOnTerminatedBranchExecution() { List<ExecutionMessage> messages = new ArrayList<>(); Execution execution1 = createBranchExecution(); Execution execution2 = createBranchExecution(); messages.add(createExecutionMessage(execution1)); messages.add(createExecutionMessage(execution2)); queueListener.onTerminated(messages); verify(splitJoinService, times(1)).endBranch((List<Execution>) argThat(hasItem(execution1))); verify(splitJoinService, times(1)).endBranch((List<Execution>) argThat(hasItem(execution2))); }