ScheduledSearchIndexerService extends AbstractService implements Runnable { @Override public void run() { logger.debug("Search Indexer #{} started at {}", counter.incrementAndGet(), Instant.now()); CommandQuery query = new CommandQuery(); query.setTo(MessageRecipient.MANAGEMENT_APIS.name()); query.setTags(Collections.singletonList(CommandTags.DATA_TO_INDEX)); List<CommandEntity> messageEntities = commandService.search(query); messageEntities.forEach(commandEntity -> { if (commandEntity.isExpired()) { commandService.delete(commandEntity.getId()); } else { if (!commandEntity.isProcessedInCurrentNode()) { commandService.ack(commandEntity.getId()); try { searchEngineService.process( mapper.readValue(commandEntity.getContent(), CommandSearchIndexerEntity.class)); } catch (IOException e) { logger.error("Search Indexer has received a bad message.", e); } } } }); logger.debug("Search Indexer #{} ended at {}", counter.get(), Instant.now()); }  @Override void run();  }
@Test public void shouldDoNothing() { when(commandService.search(any())).thenReturn(Collections.emptyList()); service.run(); verify(commandService, never()).ack(anyString()); verify(searchEngineService, never()).process(any()); }
@Test public void shouldInsertAndDelete() { CommandEntity insert = new CommandEntity(); insert.setId("insertid"); insert.setTags(Collections.singletonList(CommandTags.DATA_TO_INDEX)); insert.setContent("{\"id\":\"1\"}"); CommandEntity delete = new CommandEntity(); delete.setId("deleteid"); delete.setTags(Collections.singletonList(CommandTags.DATA_TO_INDEX)); delete.setContent("{\"id\":\"2\"}"); when(commandService.search(any())).thenReturn(Arrays.asList(delete, insert)); service.run(); verify(commandService, times(2)).ack(anyString()); verify(searchEngineService, times(2)).process(any()); }