Index: oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManagerTest.java =================================================================== --- oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManagerTest.java (revision 1466499) +++ oak-core/src/test/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManagerTest.java (working copy) @@ -26,17 +26,20 @@ import static org.junit.Assert.assertTrue; import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import org.apache.jackrabbit.oak.api.PropertyState; import org.apache.jackrabbit.oak.api.Type; import org.apache.jackrabbit.oak.plugins.index.p2.Property2IndexHookProvider; import org.apache.jackrabbit.oak.plugins.index.p2.Property2IndexLookup; -import org.apache.jackrabbit.oak.spi.commit.Editor; +import org.apache.jackrabbit.oak.plugins.memory.MemoryNodeStore; import org.apache.jackrabbit.oak.spi.commit.EditorHook; -import org.apache.jackrabbit.oak.spi.commit.EditorProvider; import org.apache.jackrabbit.oak.spi.query.PropertyValues; import org.apache.jackrabbit.oak.spi.state.NodeBuilder; import org.apache.jackrabbit.oak.spi.state.NodeState; +import org.apache.jackrabbit.oak.spi.state.NodeStore; import org.junit.Test; import com.google.common.collect.ImmutableSet; @@ -265,15 +268,9 @@ Type.NAME); NodeState after = builder.getNodeState(); - EditorProvider provider = new EditorProvider() { - @Override - public Editor getRootEditor(NodeState before, NodeState after, - NodeBuilder builder) { - return new IndexHookManagerDiff( - new Property2IndexHookProvider(), builder); - } - }; - EditorHook hook = new EditorHook(provider); + IndexHookManager im = IndexHookManager + .of(new Property2IndexHookProvider()); + EditorHook hook = new EditorHook(im); NodeState indexed = hook.processCommit(before, after); // check that the index content nodes exist @@ -291,4 +288,112 @@ return c; } + /** + * Async Index Test + * + * + */ + @Test + public void testAsync() throws Exception { + NodeStore store = new MemoryNodeStore(); + NodeState root = store.getRoot(); + + NodeBuilder builder = root.builder(); + + builder.child("oak:index") + .child("rootIndex") + .setProperty("propertyNames", "foo") + .setProperty("type", "p2") + .setProperty("async", "true") + .setProperty(JCR_PRIMARYTYPE, INDEX_DEFINITIONS_NODE_TYPE, + Type.NAME); + builder.child("newchild") + .child("other") + .child("oak:index") + .child("subIndex") + .setProperty("propertyNames", "foo") + .setProperty("type", "p2") + .setProperty("async", "true") + .setProperty(JCR_PRIMARYTYPE, INDEX_DEFINITIONS_NODE_TYPE, + Type.NAME); + + NodeState before = builder.getNodeState(); + // Add nodes + builder.child("testRoot").setProperty("foo", "abc"); + builder.child("newchild").child("other").child("testChild") + .setProperty("foo", "xyz"); + + NodeState after = builder.getNodeState(); + + ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); + + IndexHookManager im = IndexHookManager.of( + new Property2IndexHookProvider()).withObservation(store, + executor); + EditorHook hook = new EditorHook(im); + NodeState indexed = hook.processCommit(before, after); + + executor.shutdown(); + for (int i = 0; i < 3 && !executor.isTerminated(); i++) { + TimeUnit.SECONDS.sleep(1); + } + assertTrue("Waited 3 seconds and indexing is not done", + executor.isTerminated()); + + // first check that the index content nodes exist + checkPathExists(indexed, "oak:index", "rootIndex", ":index"); + checkPathExists(indexed, "newchild", "other", "oak:index", "subIndex", + ":index"); + + Property2IndexLookup lookup = new Property2IndexLookup(indexed); + assertEquals(ImmutableSet.of("testRoot"), find(lookup, "foo", "abc")); + + Property2IndexLookup lookupChild = new Property2IndexLookup(indexed + .getChildNode("newchild").getChildNode("other")); + assertEquals(ImmutableSet.of("testChild"), + find(lookupChild, "foo", "xyz")); + assertEquals(ImmutableSet.of(), find(lookupChild, "foo", "abc")); + + } + + /** + * Async Index Test - no scheduler -> no index + */ + @Test + public void testAsyncNoScheduler() throws Exception { + NodeStore store = new MemoryNodeStore(); + NodeState root = store.getRoot(); + + NodeBuilder builder = root.builder(); + + builder.child("oak:index") + .child("rootIndex") + .setProperty("propertyNames", "foo") + .setProperty("type", "p2") + .setProperty("reindex", "false") + .setProperty("async", "true") + .setProperty(JCR_PRIMARYTYPE, INDEX_DEFINITIONS_NODE_TYPE, + Type.NAME); + + NodeState before = builder.getNodeState(); + // Add nodes + builder.child("testRoot").setProperty("foo", "abc"); + + NodeState after = builder.getNodeState(); + + IndexHookManager im = IndexHookManager + .of(new Property2IndexHookProvider()); + EditorHook hook = new EditorHook(im); + NodeState indexed = hook.processCommit(before, after); + + // check that the index content nodes not exist + assertFalse(checkPathExists(indexed, "oak:index", "rootIndex") + .hasChildNode(":index")); + + } + } Index: oak-core/src/main/java/org/apache/jackrabbit/oak/Oak.java =================================================================== --- oak-core/src/main/java/org/apache/jackrabbit/oak/Oak.java (revision 1466499) +++ oak-core/src/main/java/org/apache/jackrabbit/oak/Oak.java (working copy) @@ -18,6 +18,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ScheduledExecutorService; import javax.annotation.Nonnull; import javax.jcr.NoSuchWorkspaceException; @@ -57,6 +58,7 @@ import static com.google.common.base.Preconditions.checkNotNull; import static com.google.common.collect.Lists.newArrayList; +import static java.util.concurrent.Executors.newScheduledThreadPool; /** * Builder class for constructing {@link ContentRepository} instances with @@ -87,6 +89,8 @@ private SecurityProvider securityProvider; + private ScheduledExecutorService executor = newScheduledThreadPool(0); + private String defaultWorkspaceName = DEFAULT_WORKSPACE_NAME; public Oak(NodeStore store) { @@ -213,6 +217,11 @@ return this; } + @Nonnull + public SecurityProvider getSecurityProvider() { + return this.securityProvider; + } + /** * Associates the given conflict handler with the repository to be created. * @@ -226,6 +235,17 @@ return this; } + @Nonnull + public Oak with(@Nonnull ScheduledExecutorService executorService) { + this.executor = executorService; + return this; + } + + @Nonnull + public ScheduledExecutorService getExecutorService() { + return this.executor; + } + public ContentRepository createContentRepository() { IndexHookProvider indexHooks = CompositeIndexHookProvider.compose(indexHookProviders); OakInitializer.initialize(store, new CompositeInitializer(initializers), indexHooks); @@ -251,7 +271,7 @@ CompositeHook.compose(initHooks)); // add index hooks later to prevent the OakInitializer to do excessive indexing - with(IndexHookManager.of(indexHooks)); + with(IndexHookManager.of(indexHooks).withObservation(store, executor)); withEditorHook(); CommitHook commitHook = CompositeHook.compose(commitHooks); return new ContentRepositoryImpl( Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUtils.java =================================================================== --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUtils.java (revision 1466499) +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexUtils.java (working copy) @@ -152,4 +152,19 @@ return new IndexDefinitionImpl(name, type, concat(path, name)); } + public static boolean isIndexNodeType(NodeState state) { + PropertyState ps = state.getProperty(JCR_PRIMARYTYPE); + return ps != null && !ps.isArray() + && ps.getValue(Type.STRING).equals(INDEX_DEFINITIONS_NODE_TYPE); + } + + public static boolean getBoolean(NodeState state, String property, + boolean def) { + PropertyState ps = state.getProperty(property); + if (ps == null) { + return def; + } + return ps != null && ps.getValue(Type.BOOLEAN); + } + } Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManagerDiff.java =================================================================== --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManagerDiff.java (revision 1466499) +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManagerDiff.java (working copy) @@ -16,12 +16,12 @@ */ package org.apache.jackrabbit.oak.plugins.index; -import static org.apache.jackrabbit.JcrConstants.JCR_PRIMARYTYPE; +import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.ASYNC_PROPERTY_NAME; import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.INDEX_DEFINITIONS_NAME; -import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.INDEX_DEFINITIONS_NODE_TYPE; import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.REINDEX_PROPERTY_NAME; import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_PROPERTY_NAME; -import static org.apache.jackrabbit.oak.plugins.index.IndexConstants.TYPE_UNKNOWN; +import static org.apache.jackrabbit.oak.plugins.index.IndexUtils.getBoolean; +import static org.apache.jackrabbit.oak.plugins.index.IndexUtils.isIndexNodeType; import static org.apache.jackrabbit.oak.plugins.memory.EmptyNodeState.EMPTY_NODE; import java.util.HashSet; @@ -31,6 +31,7 @@ import org.apache.jackrabbit.oak.api.CommitFailedException; import org.apache.jackrabbit.oak.api.PropertyState; import org.apache.jackrabbit.oak.api.Type; +import org.apache.jackrabbit.oak.plugins.index.async.IndexSchedulerImpl; import org.apache.jackrabbit.oak.spi.commit.CompositeEditor; import org.apache.jackrabbit.oak.spi.commit.DefaultEditor; import org.apache.jackrabbit.oak.spi.commit.Editor; @@ -52,13 +53,17 @@ private final IndexHookProvider provider; + private IndexSchedulerImpl indexScheduler; + private final NodeBuilder node; private Editor inner = new DefaultEditor(); - public IndexHookManagerDiff(IndexHookProvider provider, NodeBuilder node) { + public IndexHookManagerDiff(IndexHookProvider provider, NodeBuilder node, + IndexSchedulerImpl indexScheduler) { this.provider = provider; this.node = node; + this.indexScheduler = indexScheduler; } @Override @@ -66,42 +71,53 @@ throws CommitFailedException { NodeState ref = node.getNodeState(); if (ref.hasChildNode(INDEX_DEFINITIONS_NAME)) { - Set existingTypes = new HashSet(); + Set allTypes = new HashSet(); + Set asyncTypes = new HashSet(); Set reindexTypes = new HashSet(); + NodeState index = ref.getChildNode(INDEX_DEFINITIONS_NAME); for (String indexName : index.getChildNodeNames()) { NodeState indexChild = index.getChildNode(indexName); - if (isIndexNodeType(indexChild.getProperty(JCR_PRIMARYTYPE))) { - PropertyState reindexPS = indexChild - .getProperty(REINDEX_PROPERTY_NAME); - boolean reindex = reindexPS == null - || (reindexPS != null && indexChild.getProperty( - REINDEX_PROPERTY_NAME).getValue( - Type.BOOLEAN)); - String type = TYPE_UNKNOWN; + if (isIndexNodeType(indexChild)) { + boolean reindex = getBoolean(indexChild, + REINDEX_PROPERTY_NAME, true); + boolean async = getBoolean(indexChild, ASYNC_PROPERTY_NAME, + false); + String type = null; PropertyState typePS = indexChild .getProperty(TYPE_PROPERTY_NAME); if (typePS != null && !typePS.isArray()) { type = typePS.getValue(Type.STRING); } + if (type == null) { + // skip null types + continue; + } + if (async) { + asyncTypes.add(type); + } if (reindex) { reindexTypes.add(type); } - existingTypes.add(type); + allTypes.add(type); } } - existingTypes.remove(TYPE_UNKNOWN); - reindexTypes.remove(TYPE_UNKNOWN); List hooks = Lists.newArrayList(); List reindex = Lists.newArrayList(); - for (String type : existingTypes) { - List hooksTmp = provider.getIndexHooks( - type, node); - if (reindexTypes.contains(type)) { - reindex.addAll(hooksTmp); + for (String type : allTypes) { + if (asyncTypes.contains(type)) { + // TODO what happens when there is no indexScheduler? + if (indexScheduler != null) { + indexScheduler.addOrUpdate(new TypedEditorProvider( + provider, type)); + } } else { - hooks.addAll(hooksTmp); + if (reindexTypes.contains(type)) { + reindex.addAll(provider.getIndexHooks(type, node)); + } else { + hooks.addAll(provider.getIndexHooks(type, node)); + } } } reindex(reindex, ref); @@ -147,11 +163,6 @@ this.inner.leave(before, after); } - private static boolean isIndexNodeType(PropertyState ps) { - return ps != null && !ps.isArray() - && ps.getValue(Type.STRING).equals(INDEX_DEFINITIONS_NODE_TYPE); - } - @Override public void propertyAdded(PropertyState after) throws CommitFailedException { inner.propertyAdded(after); @@ -187,4 +198,59 @@ return inner.childNodeDeleted(name, before); } -} \ No newline at end of file + /** + * This creates a composite editor from a type-filtered index provider. + * + */ + private static class TypedEditorProvider implements EditorProvider { + + private final IndexHookProvider provider; + + private final String type; + + public TypedEditorProvider(IndexHookProvider provider, String type) { + this.type = type; + this.provider = provider; + } + + /** + * This does not make any effort to filter async definitions. The + * assumption is that given an index type, all of the returned index + * hooks inherit the same async assumption. + * + */ + @Override + public Editor getRootEditor(NodeState before, NodeState after, + NodeBuilder builder) { + return VisibleEditor.wrap(CompositeEditor.compose(provider + .getIndexHooks(type, builder))); + } + + @Override + public int hashCode() { + final int prime = 31; + int result = 1; + result = prime * result + ((type == null) ? 0 : type.hashCode()); + return result; + } + + @Override + public boolean equals(Object obj) { + if (this == obj) + return true; + if (obj == null) + return false; + if (getClass() != obj.getClass()) + return false; + TypedEditorProvider other = (TypedEditorProvider) obj; + if (type == null) { + if (other.type != null) + return false; + } else if (!type.equals(other.type)) + return false; + return true; + } + + } + +} Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/async/IndexSchedulerImpl.java =================================================================== --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/async/IndexSchedulerImpl.java (revision 0) +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/async/IndexSchedulerImpl.java (revision 0) @@ -0,0 +1,126 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.jackrabbit.oak.plugins.index.async; + +import static com.google.common.base.Preconditions.checkNotNull; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; + +import javax.annotation.Nonnull; + +import org.apache.jackrabbit.oak.api.CommitFailedException; +import org.apache.jackrabbit.oak.spi.commit.EditorHook; +import org.apache.jackrabbit.oak.spi.commit.EditorProvider; +import org.apache.jackrabbit.oak.spi.commit.EmptyHook; +import org.apache.jackrabbit.oak.spi.state.NodeState; +import org.apache.jackrabbit.oak.spi.state.NodeStore; +import org.apache.jackrabbit.oak.spi.state.NodeStoreBranch; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class IndexSchedulerImpl { + + // TODO handle deletes as well + + private final NodeStore store; + + private final ScheduledExecutorService executor; + + private final Map ongoing = new ConcurrentHashMap(); + + public IndexSchedulerImpl(@Nonnull ScheduledExecutorService executor, + @Nonnull NodeStore store) { + this.store = store; + this.executor = checkNotNull(executor); + } + + public void addOrUpdate(EditorProvider provider) { + IndexTask task = ongoing.get(provider); + if (task != null) { + // stop existing one + task.stop(); + } + task = new IndexTask(store, provider); + ongoing.put(provider, task); + task.start(executor); + } + + public void removeListener(EditorProvider provider) { + IndexTask task = ongoing.remove(provider); + if (task != null) { + task.stop(); + } + } + + private static class IndexTask implements Runnable { + + private static final Logger log = LoggerFactory + .getLogger(IndexTask.class); + + private final NodeStore store; + + private final EditorProvider provider; + + private ScheduledFuture future; + + private NodeState before; + + public IndexTask(NodeStore store, EditorProvider provider) { + this.store = store; + this.provider = provider; + this.before = store.getRoot(); + } + + public synchronized void start(ScheduledExecutorService executor) { + if (future != null) { + throw new IllegalStateException("IndexTask has already started"); + } + future = executor.scheduleWithFixedDelay(this, 100, 1000, + TimeUnit.MILLISECONDS); + } + + public synchronized void stop() { + if (future == null) { + log.warn("IndexTask has already stopped."); + } + future.cancel(true); + } + + @Override + public void run() { + NodeStoreBranch branch = store.branch(); + NodeState after = branch.getHead(); + try { + EditorHook hook = new EditorHook(provider); + NodeState processed = hook.processCommit(before, after); + + branch.setRoot(processed); + branch.merge(EmptyHook.INSTANCE); + before = after; + } catch (CommitFailedException e) { + log.warn("IndexTask update failed", e); + } + } + } + +} Property changes on: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/async/IndexSchedulerImpl.java ___________________________________________________________________ Added: svn:keywords + Author Date Id Revision Rev URL Added: svn:eol-style + native Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManager.java =================================================================== --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManager.java (revision 1466499) +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexHookManager.java (working copy) @@ -16,11 +16,17 @@ */ package org.apache.jackrabbit.oak.plugins.index; +import java.util.concurrent.ScheduledExecutorService; + +import javax.annotation.Nonnull; + +import org.apache.jackrabbit.oak.plugins.index.async.IndexSchedulerImpl; import org.apache.jackrabbit.oak.spi.commit.Editor; import org.apache.jackrabbit.oak.spi.commit.EditorProvider; import org.apache.jackrabbit.oak.spi.commit.VisibleEditor; import org.apache.jackrabbit.oak.spi.state.NodeBuilder; import org.apache.jackrabbit.oak.spi.state.NodeState; +import org.apache.jackrabbit.oak.spi.state.NodeStore; /** * Keeps existing IndexHooks updated.
@@ -38,6 +44,8 @@ private final IndexHookProvider provider; + private IndexSchedulerImpl indexScheduler; + protected IndexHookManager(IndexHookProvider provider) { this.provider = provider; } @@ -45,6 +53,13 @@ @Override public Editor getRootEditor(NodeState before, NodeState after, NodeBuilder builder) { - return VisibleEditor.wrap(new IndexHookManagerDiff(provider, builder)); + return VisibleEditor.wrap(new IndexHookManagerDiff(provider, builder, + indexScheduler)); } + + public IndexHookManager withObservation(@Nonnull NodeStore store, + @Nonnull ScheduledExecutorService executor) { + this.indexScheduler = new IndexSchedulerImpl(executor, store); + return this; + } } Index: oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexConstants.java =================================================================== --- oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexConstants.java (revision 1466499) +++ oak-core/src/main/java/org/apache/jackrabbit/oak/plugins/index/IndexConstants.java (working copy) @@ -31,6 +31,8 @@ String REINDEX_PROPERTY_NAME = "reindex"; + String ASYNC_PROPERTY_NAME = "async"; + /** * Marks a unique property index. */ Index: oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/Jcr.java =================================================================== --- oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/Jcr.java (revision 1466499) +++ oak-jcr/src/main/java/org/apache/jackrabbit/oak/jcr/Jcr.java (working copy) @@ -16,7 +16,6 @@ */ package org.apache.jackrabbit.oak.jcr; -import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import javax.annotation.Nonnull; @@ -52,10 +51,6 @@ private final Oak oak; - private ScheduledExecutorService executor = Executors.newScheduledThreadPool(0); - - private SecurityProvider securityProvider; - public Jcr(Oak oak) { this.oak = oak; @@ -126,7 +121,6 @@ @Nonnull public final Jcr with(@Nonnull SecurityProvider securityProvider) { oak.with(checkNotNull(securityProvider)); - this.securityProvider = securityProvider; return this; } @@ -138,13 +132,15 @@ @Nonnull public final Jcr with(@Nonnull ScheduledExecutorService executor) { - this.executor = checkNotNull(executor); + oak.with(checkNotNull(executor)); return this; } public Repository createRepository() { return new RepositoryImpl( - oak.createContentRepository(), executor, securityProvider); + oak.createContentRepository(), + oak.getExecutorService(), + oak.getSecurityProvider()); } }