class 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()));

        // Just log failed messages, but do not do anything to the queue. Let user control this behavior through SQS queue's visibility timeout setting
        logFailedMessageResults(resultsByStatus.getOrDefault(SQSMessageResult.Status.ERROR, Collections.emptyList()));
    }

    public  MessageDispatcher(final SQSProxy sqsProxy, final MessageProcessorProxy messageProcessorProxy);

}