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);  }
@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(); }
@Test public void reprocessingShouldEnqueueAllMails() 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(queueFactory.getQueue(SPOOL).get().browse()) .toIterable() .extracting(item -> item.getMail().getName()) .containsOnly(NAME_1, NAME_2, NAME_3); }
@Test public void reprocessingShouldNotFailOnConcurrentDeletion() throws Exception { MailRepository repository = mailRepositoryStore.select(MailRepositoryUrl.fromPathAndProtocol(PATH, MEMORY_PROTOCOL)); repository.store(mail1); repository.store(mail2); repository.store(mail3); AtomicBoolean shouldPerformRemove = new AtomicBoolean(true); ConsumerChainer<MailKey> concurrentRemoveConsumer = Throwing.consumer(key -> { if (shouldPerformRemove.get()) { shouldPerformRemove.set(false); MailKey toRemove = ImmutableList.of(NAME_1, NAME_2, NAME_3) .stream() .map(MailKey::new) .filter(candidateForRemoval -> !candidateForRemoval.equals(key)) .findFirst() .get(); repository.remove(toRemove); } }); reprocessingService.reprocessAll(PATH, NO_TARGET_PROCESSOR, SPOOL, concurrentRemoveConsumer); assertThat(queueFactory.getQueue(SPOOL).get().browse()) .toIterable() .hasSize(2); }