AsyncIndexerService { @Activate public void activate(BundleContext bundleContext, Configuration config) { List<AsyncConfig> asyncIndexerConfig = getAsyncConfig(config.asyncConfigs()); Whiteboard whiteboard = new OsgiWhiteboard(bundleContext); indexRegistration = new IndexMBeanRegistration(whiteboard); indexEditorProvider.start(whiteboard); executor = new WhiteboardExecutor(); executor.start(whiteboard); long leaseTimeOutMin = config.leaseTimeOutMinutes(); if (!(nodeStore instanceof Clusterable)){ leaseTimeOutMin = 0; log.info("Detected non clusterable setup. Lease checking would be disabled for async indexing"); } TrackingCorruptIndexHandler corruptIndexHandler = createCorruptIndexHandler(config); for (AsyncConfig c : asyncIndexerConfig) { AsyncIndexUpdate task = new AsyncIndexUpdate(c.name, nodeStore, indexEditorProvider, statisticsProvider, false); task.setCorruptIndexHandler(corruptIndexHandler); task.setValidatorProviders(Collections.singletonList(validatorProvider)); task.setLeaseTimeOut(TimeUnit.MINUTES.toMillis(leaseTimeOutMin)); indexRegistration.registerAsyncIndexer(task, c.timeIntervalInSecs); closer.register(task); } registerAsyncReindexSupport(whiteboard); log.info("Configured async indexers {} ", asyncIndexerConfig); log.info("Lease time: {} mins and AsyncIndexUpdate configured with {}", leaseTimeOutMin, validatorProvider.getClass().getName()); }  @Activate void activate(BundleContext bundleContext, Configuration config); @Deactivate void deactivate();  }
@Test public void leaseTimeout() throws Exception{ injectDefaultServices(); Map<String,Object> config = ImmutableMap.<String, Object>of( "asyncConfigs", new String[] {"async:5"}, "leaseTimeOutMinutes" , "20" ); MockOsgi.activate(service, context.bundleContext(), config); AsyncIndexUpdate indexUpdate = getIndexUpdate("async"); assertEquals(TimeUnit.MINUTES.toMillis(20), indexUpdate.getLeaseTimeOut()); }
@Test public void changeCollectionEnabled() throws Exception{ injectDefaultServices(); Map<String,Object> config = ImmutableMap.<String, Object>of( "asyncConfigs", new String[] {"async:5"} ); context.registerService(IndexEditorProvider.class, new PropertyIndexEditorProvider()); MockOsgi.activate(service, context.bundleContext(), config); NodeBuilder builder = nodeStore.getRoot().builder(); createIndexDefinition(builder.child(INDEX_DEFINITIONS_NAME), "rootIndex", true, false, ImmutableSet.of("foo"), null) .setProperty(ASYNC_PROPERTY_NAME, "async"); builder.child("testRoot").setProperty("foo", "abc"); nodeStore.merge(builder, EmptyHook.INSTANCE, CommitInfo.EMPTY); CommitInfoCollector infoCollector = new CommitInfoCollector(); nodeStore.addObserver(infoCollector); AsyncIndexUpdate indexUpdate = getIndexUpdate("async"); indexUpdate.run(); CommitContext commitContext = (CommitContext) infoCollector.infos.get(0).getInfo().get(CommitContext.NAME); assertNotNull(commitContext); ChangeSet changeSet = (ChangeSet) commitContext.get(ChangeSet.COMMIT_CONTEXT_OBSERVATION_CHANGESET); assertNotNull(changeSet); }
@Test public void nonClusterableNodeStoreAndLeaseTimeout() throws Exception{ nodeStore = new MemoryNodeStore(); injectDefaultServices(); Map<String,Object> config = ImmutableMap.<String, Object>of( "asyncConfigs", new String[] {"async:5"}, "leaseTimeOutMinutes" , "20" ); MockOsgi.activate(service, context.bundleContext(), config); AsyncIndexUpdate indexUpdate = getIndexUpdate("async"); assertEquals(0, indexUpdate.getLeaseTimeOut()); }
@Test public void corruptIndexTimeout() throws Exception{ injectDefaultServices(); Map<String,Object> config = ImmutableMap.<String, Object>of( "asyncConfigs", new String[] {"async:5"}, "failingIndexTimeoutSeconds" , "43", "errorWarnIntervalSeconds" , "53" ); MockOsgi.activate(service, context.bundleContext(), config); AsyncIndexUpdate indexUpdate = getIndexUpdate("async"); assertEquals(TimeUnit.SECONDS.toMillis(43), indexUpdate.getCorruptIndexHandler().getCorruptIntervalMillis()); assertEquals(TimeUnit.SECONDS.toMillis(53), indexUpdate.getCorruptIndexHandler().getErrorWarnIntervalMillis()); }