MessageDispatcher { public void dispatch(final List<Message> messages) { Preconditions.checkArgument(!messages.isEmpty(), "messages cannot be empty"); Instant start = Instant.now(clock); SQSMessageProcessorResponse response; try { response = messageProcessorProxy.invoke(new SQSMessageProcessorRequest(messages)); stats.record(Duration.between(start, Instant.now(clock)), messages.size()); } catch (MessageProcessorException e) { log.warn("MessageProcessor encountered an error", e); return; } Map<SQSMessageResult.Status, List<SQSMessageResult>> resultsByStatus = response.getMessageResults() .stream() .collect(Collectors.groupingBy(SQSMessageResult::getStatus)); deleteMessages(messages, resultsByStatus.getOrDefault(SQSMessageResult.Status.SUCCESS, Collections.emptyList())); retryMessages(messages, resultsByStatus.getOrDefault(SQSMessageResult.Status.RETRY, Collections.emptyList())); logFailedMessageResults(resultsByStatus.getOrDefault(SQSMessageResult.Status.ERROR, Collections.emptyList())); } MessageDispatcher(final SQSProxy sqsProxy, final MessageProcessorProxy messageProcessorProxy); void dispatch(final List<Message> messages); void reset(); int getEstimatedCapacity(Instant cutoff);  }
@Test public void dispatch_errorResults() throws Exception { Message m1 = mockMessage("1"); Message m2 = mockMessage("2"); List<Message> messages = Lists.newArrayList(m1, m2); mockProcessorResults( messageResult("1", SQSMessageResult.Status.ERROR), messageResult("2", SQSMessageResult.Status.ERROR) ); dispatcher.dispatch(messages); verify(messageProcessorProxy).invoke(new SQSMessageProcessorRequest(messages)); verifyNoMoreInteractions(messageProcessorProxy, sqsProxy); }
@Test public void dispatch_messageProcessorException() throws Exception { List<Message> messages = Lists.newArrayList(mockMessage("1")); doThrow(MessageProcessorException.class).when(messageProcessorProxy).invoke(any()); dispatcher.dispatch(messages); verify(messageProcessorProxy).invoke(new SQSMessageProcessorRequest(messages)); verifyNoMoreInteractions(messageProcessorProxy, sqsProxy); }