NlpConsumer implements DatashareListener { void findNamedEntities(final String projectName, final String id, final String routing) throws InterruptedException { try { Document doc = indexer.get(projectName, id, routing); if (doc != null) { logger.info("extracting {} entities for document {}", nlpPipeline.getType(), doc.getId()); if (nlpPipeline.initialize(doc.getLanguage())) { List<NamedEntity> namedEntities = nlpPipeline.process(doc); indexer.bulkAdd(projectName, nlpPipeline.getType(), namedEntities, doc); logger.info("added {} named entities to document {}", namedEntities.size(), doc.getId()); nlpPipeline.terminate(doc.getLanguage()); } } else { logger.warn("no document found in index with id " + id); } } catch (IOException e) { logger.error("cannot extract entities of doc " + id, e); } } @Inject  NlpConsumer(Pipeline pipeline, Indexer indexer, BlockingQueue<Message> messageQueue); @Override Integer call();  }
@Test public void test_on_message_does_nothing__when_doc_not_found_in_index() throws Exception { nlpListener.findNamedEntities("projectName","unknownId", "routing"); verify(pipeline, never()).initialize(any(Language.class)); verify(pipeline, never()).process(any()); }
@Test public void test_on_message_do_not_processNLP__when_init_fails() throws Exception { when(pipeline.initialize(any())).thenReturn(false); when(indexer.get(anyString(), anyString(), anyString())).thenReturn(createDoc("content").build()); nlpListener.findNamedEntities("projectName","id", "routing"); verify(pipeline, never()).process(any()); }
@Test public void test_on_message_processNLP__when_doc_found_in_index() throws Exception { when(pipeline.initialize(any())).thenReturn(true); Document doc = createDoc("content").build(); when(pipeline.process(doc)).thenReturn(emptyList()); when(indexer.get("projectName", doc.getId(), "routing")).thenReturn(doc); nlpListener.findNamedEntities("projectName", doc.getId(), "routing"); verify(pipeline).initialize(ENGLISH); verify(pipeline).process(doc); }