CollectionResultsRouter extends StepManager implements TaskFinishPoint { @Override public void onMessage(Message message) { if (message instanceof ObjectMessage) { timeoutWatch.update(); try { CollectorResultData collectorResultData = (CollectorResultData) ((ObjectMessage) message) .getObject(); collectorJobScheduler.messageReceived(collectorResultData.getRequestMessageId(), message.getJMSCorrelationID()); String testName = collectorResultData.getTestName(); updateCounters(collectorResultData.getStatus()); LOGGER.info( "Received message {} - url {} in test `{}` collected with status {}. Results received " + "successful {} / failed {} of {} total. CorrelationId: {}.", message.getJMSMessageID(), collectorResultData.getUrl(), testName, collectorResultData.getStatus(), messagesReceivedSuccess.get(), messagesReceivedFailed.get(), getTotalTasksCount(), correlationId); if (collectorResultData.getStatus() == JobStatus.SUCCESS) { onSuccess(collectorResultData, testName); } else { final Url failedUrl = collectorResultData.getUrl(); failedUrl.setErrorMessage(collectorResultData.getProcessingError().getDescription()); updateSuiteUrl(testName, failedUrl); onError(collectorResultData.getProcessingError()); } if (isFinished()) { LOGGER.info("CollectionResultsRouter task finished."); finishTask(); } } catch (JMSException e) { LOGGER.error("Error while collecting results in CollectionResultsRouter", e); onError(ProcessingError.collectingError(e.getMessage())); } } } CollectionResultsRouter(TimeoutWatch timeoutWatch, JmsConnection jmsConnection,
      RunnerConfiguration runnerConfiguration,
      CollectorJobSchedulerService collectorJobScheduler, RunIndexWrapper runIndexWrapper); @Override void onMessage(Message message); void addChangeObserver(ChangeObserver observer); @Override boolean isFinished();  }
@Test public void onMessage_whenSuccess_expectChangeListenersNotifiedAndMessageSent() throws Exception { ObjectMessage message = Mockito.mock(ObjectMessage.class); Step stepA = mockCollectionStep("collectorA", Collections.<Comparator>emptySet()); Comparator mockedComparator = Mockito.mock(Comparator.class); Step stepB = mockCollectionStep("collectorB", Collections.singleton(mockedComparator)); List<Step> steps = Arrays.asList(stepA, stepB); Url url = Mockito.mock(Url.class); when(url.getSteps()).thenReturn(steps); CollectorResultData collectorResultData = CollectorResultData .createSuccessResult(url, "jmsID001", "testNameExample"); when(message.getObject()).thenReturn(collectorResultData); when(message.getJMSCorrelationID()).thenReturn(CORRELATION_ID); when(message.getJMSMessageID()).thenReturn("jmsID0001"); tested.onMessage(message); verify(changeObserver, times(1)).updateAmountToReceive(1); verify(sender, times(1)).send(Matchers.<Message>any()); verify(mockedMessage, times(1)).setJMSCorrelationID(CORRELATION_ID); verify(changeObserver, times(1)).informChangesCompleted(); verify(consumer, times(1)).close(); }
@Test public void onMessage_whenError_expectObserversNotified() throws Exception { ObjectMessage message = Mockito.mock(ObjectMessage.class); Url url = Mockito.mock(Url.class); CollectorResultData collectorResultData = CollectorResultData .createErrorResult(url, ProcessingError .collectingError("error"), "jmsID001", "testNameExample"); when(message.getObject()).thenReturn(collectorResultData); when(message.getJMSCorrelationID()).thenReturn(CORRELATION_ID); when(message.getJMSMessageID()).thenReturn("jmsID0001"); tested.onMessage(message); verify(observer, times(1)).update(Matchers.<Observable>any(), any()); verify(changeObserver, times(1)).informChangesCompleted(); verify(consumer, times(1)).close(); }