ResumeNlpTask implements Callable<Long>, UserTask { @Override public Long call() throws IOException { logger.info("resuming NLP name finding for index {} and {}", projectName, nlpPipelines); Indexer.Searcher searcher = indexer.search(projectName, Document.class).withSource("rootDocument").without(nlpPipelines.toArray(new Pipeline.Type[] {})); List<? extends Entity> docsToProcess = searcher.scroll().collect(toList()); long totalHits = searcher.totalHits(); this.publisher.publish(Channel.NLP, new Message(Message.Type.INIT_MONITORING).add(Message.Field.VALUE, valueOf(totalHits))); do { docsToProcess.forEach(doc -> this.publisher.publish(Channel.NLP, new Message(Message.Type.EXTRACT_NLP) .add(Message.Field.INDEX_NAME, projectName) .add(Message.Field.DOC_ID, doc.getId()) .add(Message.Field.R_ID, ((Document) doc).getRootDocument()))); docsToProcess = searcher.scroll().collect(toList()); } while (docsToProcess.size() != 0); logger.info("sent {} message for {} files without {} pipeline tags", Message.Type.EXTRACT_NLP, totalHits, nlpPipelines); searcher.clearScroll(); this.publisher.publish(Channel.NLP, new ShutdownMessage()); return totalHits; } @Inject  ResumeNlpTask(final Publisher publisher, final Indexer indexer, final PropertiesProvider propertiesProvider,
                         @Assisted final User user, @Assisted final Set<Pipeline.Type> nlpPipelines); @Override Long call(); @Override User getUser();  }
@Test public void test_bug_size_of_search() throws Exception { for (int i = 0; i < 20; i++) { indexer.add(TEST_INDEX, createDoc("doc" + i).with(Pipeline.Type.CORENLP).build()); } Publisher publisher = mock(Publisher.class); ResumeNlpTask resumeNlpTask = new ResumeNlpTask(publisher, indexer, new PropertiesProvider(new HashMap<String, String>() {{ put("defaultProject", "test-datashare"); }}), new User("test"), new HashSet<Pipeline.Type>() {{add(Pipeline.Type.OPENNLP);}}); resumeNlpTask.call(); verify(publisher, times(22)).publish(any(), any()); }