class ReprocessingService {

    public void reprocessAll(MailRepositoryPath path, Optional<String> targetProcessor, MailQueueName targetQueue, Consumer<MailKey> keyListener) throws MailRepositoryStore.MailRepositoryStoreException, MessagingException {
        try (Reprocessor reprocessor = new Reprocessor(getMailQueue(targetQueue), targetProcessor)) {
            mailRepositoryStoreService
                .getRepositories(path)
                .forEach(Throwing.consumer((MailRepository repository) ->
                    Iterators.toStream(repository.list())
                        .peek(keyListener)
                        .map(Throwing.function(key -> Optional.ofNullable(repository.retrieve(key))))
                        .flatMap(Optional::stream)
                        .forEach(mail -> reprocessor.reprocess(repository, mail))));
        }
    }

    @Inject public  ReprocessingService(MailQueueFactory<?> mailQueueFactory,
                               MailRepositoryStoreService mailRepositoryStoreService);

}