@Test public void reprocessingShouldEmptyRepository() throws Exception { MailRepository repository = mailRepositoryStore.select(MailRepositoryUrl.fromPathAndProtocol(PATH, MEMORY_PROTOCOL)); repository.store(mail1); repository.store(mail2); repository.store(mail3); reprocessingService.reprocessAll(PATH, NO_TARGET_PROCESSOR, SPOOL, NOOP_CONSUMER); assertThat(repository.list()).toIterable() .isEmpty(); }
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)))); } }
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)))); } } }
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  ReprocessingService(MailQueueFactory<?> mailQueueFactory,
                               MailRepositoryStoreService mailRepositoryStoreService); }
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  ReprocessingService(MailQueueFactory<?> mailQueueFactory,
                               MailRepositoryStoreService mailRepositoryStoreService); void reprocessAll(MailRepositoryPath path, Optional<String> targetProcessor, MailQueueName targetQueue, Consumer<MailKey> keyListener); void reprocess(MailRepositoryPath path, MailKey key, Optional<String> targetProcessor, MailQueueName targetQueue); }
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  ReprocessingService(MailQueueFactory<?> mailQueueFactory,
                               MailRepositoryStoreService mailRepositoryStoreService); void reprocessAll(MailRepositoryPath path, Optional<String> targetProcessor, MailQueueName targetQueue, Consumer<MailKey> keyListener); void reprocess(MailRepositoryPath path, MailKey key, Optional<String> targetProcessor, MailQueueName targetQueue);  }