DeduplicateTask extends PipelineTask { @Override public Long call() throws Exception { int duplicates = queue.removeDuplicates(); transferToOutputQueue(); logger.info("removed {} duplicate paths in queue {}", duplicates, queue.getName()); queue.close(); return (long)duplicates; } @Inject  DeduplicateTask(final DocumentCollectionFactory factory, final PropertiesProvider propertiesProvider, @Assisted User user, @Assisted String queueName); @Override Long call();  }
@Test public void test_filter_empty() throws Exception { docCollectionFactory.createQueue(propertyProvider, "test:queue").add(POISON); assertThat(new DeduplicateTask(docCollectionFactory, new PropertiesProvider(), local(), "test:queue").call()).isEqualTo(0); }
@Test public void test_filter_queue_removes_duplicates() throws Exception { docCollectionFactory.createQueue(propertyProvider, "test:queue").put(get("/path/to/doc")); docCollectionFactory.createQueue(propertyProvider, "test:queue").put(get("/path/to/doc")); docCollectionFactory.createQueue(propertyProvider, "test:queue").add(POISON); assertThat(new DeduplicateTask(docCollectionFactory, propertyProvider, local(), "test:queue").call()).isEqualTo(1); assertThat(docCollectionFactory.createQueue(propertyProvider, "test:queue:deduplicate").size()).isEqualTo(2); }