reduce node lookups in event processing

fcrepo-kernel/src/main/java/org/fcrepo/kernel/observer/DefaultFilter.java

@@ -16,13 +16,15 @@
 package org.fcrepo.kernel.observer;
 
 import static com.google.common.base.Throwables.propagate;
-import static org.fcrepo.kernel.utils.FedoraTypesUtils.isFedoraDatastream;
-import static org.fcrepo.kernel.utils.FedoraTypesUtils.isFedoraObject;
+import static javax.jcr.observation.Event.NODE_ADDED;
+import static javax.jcr.observation.Event.NODE_MOVED;
+import static javax.jcr.observation.Event.NODE_REMOVED;
+import static javax.jcr.observation.Event.PROPERTY_ADDED;
+import static javax.jcr.observation.Event.PROPERTY_CHANGED;
+import static javax.jcr.observation.Event.PROPERTY_REMOVED;
+import static org.fcrepo.kernel.utils.FedoraTypesUtils.isFedoraObjectOrDatastream;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
 import javax.inject.Inject;
-import javax.jcr.Item;
 import javax.jcr.Node;
 import javax.jcr.PathNotFoundException;
 import javax.jcr.RepositoryException;
@@ -46,10 +48,6 @@ public class DefaultFilter implements EventFilter {
     @Inject
     private Repository repository;
 
-    // it's safe to keep the session around, because this code does not mutate
-    // the state of the repository
-    private Session session;
-
     /**
      * Filter observer events to only include events on a FedoraObject or
      * Datastream, or properties of an FedoraObject or Datastream.
@@ -59,34 +57,43 @@ public class DefaultFilter implements EventFilter {
      */
     @Override
     public boolean apply(final Event event) {
+        Session session = null;
         try {
-            final Item item = session.getItem(event.getPath());
-            final Node n = item.isNode() ? (Node)item : item.getParent();
-            return isFedoraObject.apply(n) || isFedoraDatastream.apply(n);
+            String nPath = event.getPath();
+            int nType = event.getType();
+            switch(nType) {
+                case NODE_ADDED:
+                    break;
+                case NODE_REMOVED:
+                    return true;
+                case PROPERTY_ADDED:
+                    nPath = nPath.substring(0, nPath.lastIndexOf('/'));
+                    break;
+                case PROPERTY_REMOVED:
+                    nPath = nPath.substring(0, nPath.lastIndexOf('/'));
+                    break;
+                case PROPERTY_CHANGED:
+                    nPath = nPath.substring(0, nPath.lastIndexOf('/'));
+                    break;
+                case NODE_MOVED:
+                    break;
+                default:
+                    return false;
+            }
+
+            session = repository.login();
+            final Node n = session.getNode(nPath);
+            return isFedoraObjectOrDatastream.apply(n);
         } catch (final PathNotFoundException e) {
             // not a node in the fedora workspace
             return false;
         } catch (final RepositoryException e) {
             throw propagate(e);
+        } finally {
+            if (session != null) {
+                session.logout();
+            }
         }
     }
 
-    /**
-     * Initialize a long-running read-only JCR session
-     * to use for filtering events
-     * @throws RepositoryException
-     */
-    @PostConstruct
-    public void acquireSession() throws RepositoryException {
-        session = repository.login();
-    }
-
-    /**
-     * Log-out of the read-only JCR session before destroying
-     * the filter.
-     */
-    @PreDestroy
-    public void releaseSession() {
-        session.logout();
-    }
 }

fcrepo-kernel/src/main/java/org/fcrepo/kernel/observer/SimpleObserver.java

@@ -31,10 +31,8 @@
 import java.util.Set;
 
 import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import javax.inject.Inject;
-import javax.jcr.Item;
-import javax.jcr.Node;
-import javax.jcr.PathNotFoundException;
 import javax.jcr.RepositoryException;
 import javax.jcr.Session;
 import javax.jcr.observation.Event;
@@ -79,6 +77,7 @@ public class SimpleObserver implements EventListener {
     @Inject
     private EventFilter eventFilter;
 
+    // THIS SESSION SHOULD NOT BE USED TO LOOK UP NODES
     private Session session;
 
     /**
@@ -93,6 +92,16 @@ public void buildListener() throws RepositoryException {
         session.save();
     }
 
+    /**
+     * logout of the session
+     * @throws RepositoryException
+     */
+    @PreDestroy
+    public void stopListening() throws RepositoryException {
+        session.getWorkspace().getObservationManager().removeEventListener(this);
+        session.logout();
+    }
+
     /**
      * Filter JCR events and transform them into our own FedoraEvents.
      *
@@ -101,30 +110,42 @@ public void buildListener() throws RepositoryException {
     @Override
     public void onEvent(final javax.jcr.observation.EventIterator events) {
         // keep track of nodes that trigger events to prevent duplicates
-        final Set<Node> posted = new HashSet<Node>();
+        // size to minimize resizing.
+        final Set<String> posted = new HashSet<String>((int)events.getSize() * 2 / 3);
 
         // post non-duplicate events approved by the filter
         for (final Event e : filter(new EventIterator(events), eventFilter)) {
             try {
-                final Item item = session.getItem(e.getPath());
-                Node n = null;
-                if ( item.isNode() ) {
-                    n = (Node)item;
-                } else {
-                    n = item.getParent();
+                String nPath = e.getPath();
+                int nType = e.getType();
+                // is jump table faster than two bitwise comparisons?
+                switch(nType) {
+                    case NODE_ADDED:
+                        break;
+                    case NODE_REMOVED:
+                        break;
+                    case PROPERTY_ADDED:
+                        nPath = nPath.substring(0, nPath.lastIndexOf('/'));
+                        break;
+                    case PROPERTY_REMOVED:
+                        nPath = nPath.substring(0, nPath.lastIndexOf('/'));
+                        break;
+                    case PROPERTY_CHANGED:
+                        nPath = nPath.substring(0, nPath.lastIndexOf('/'));
+                        break;
+                    case NODE_MOVED:
+                        break;
+                    default:
+                        nPath = null;
                 }
-                if ( n != null && !posted.contains(n) ) {
+                if ( nPath != null && !posted.contains(nPath) ) {
                     EVENT_COUNTER.inc();
-                    LOGGER.debug("Putting event: " + e.toString()
-                        + " on the bus.");
+                    LOGGER.debug("Putting event: {} ({}) on the bus", nPath, nType);
                     eventBus.post(new FedoraEvent(e));
-                    posted.add(n);
+                    posted.add(nPath);
                 } else {
-                    LOGGER.debug("Skipping: " + e);
+                    LOGGER.debug("Skipping event: {} ({}) on the bus", nPath, nType);
                 }
-            } catch (final PathNotFoundException ex) {
-                // we can ignore these
-                LOGGER.trace("Not a node in the Fedora workspace: " + e);
             } catch ( RepositoryException ex ) {
                 throw propagate(ex);
             }

fcrepo-kernel/src/main/java/org/fcrepo/kernel/services/functions/AllTypesPredicate.java

@@ -0,0 +1,41 @@
+/**
+ * Copyright 2013 DuraSpace, Inc.
+ *
+ * Licensed 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.fcrepo.kernel.services.functions;
+
+/**
+ * Predicate to match nodes with all of the given mixin types
+ * @author armintor@gmail.com
+ *
+ */
+public class AllTypesPredicate extends BooleanTypesPredicate {
+
+    private final int test;
+
+    /**
+     * True if all the types specified match.
+     * @param types
+     */
+    public AllTypesPredicate(String...types) {
+        super(types);
+        this.test = types.length;
+    }
+
+    @Override
+    protected boolean test(final int matched) {
+        return matched == test;
+    }
+
+}

fcrepo-kernel/src/main/java/org/fcrepo/kernel/services/functions/AnyTypesPredicate.java

@@ -0,0 +1,38 @@
+/**
+ * Copyright 2013 DuraSpace, Inc.
+ *
+ * Licensed 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.fcrepo.kernel.services.functions;
+
+
+/**
+ * Predicate to match nodes with any of the given mixin types
+ * @author armintor@gmail.com
+ *
+ */
+public class AnyTypesPredicate extends BooleanTypesPredicate {
+    /**
+     * True if any of the types specified match.
+     * @param types
+     */
+    public AnyTypesPredicate(String...types) {
+        super(types);
+    }
+
+    @Override
+    protected boolean test(final int matched) {
+        return matched > 0;
+    }
+
+}

fcrepo-kernel/src/main/java/org/fcrepo/kernel/services/functions/BooleanTypesPredicate.java

@@ -0,0 +1,68 @@
+/**
+ * Copyright 2013 DuraSpace, Inc.
+ *
+ * Licensed 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.fcrepo.kernel.services.functions;
+
+import static com.google.common.base.Throwables.propagate;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import javax.jcr.Node;
+import javax.jcr.RepositoryException;
+import javax.jcr.nodetype.NodeType;
+
+import com.google.common.base.Predicate;
+
+/**
+ * Base class for matching sets of node types
+ * @author armintor@gmail.com
+ *
+ */
+public abstract class BooleanTypesPredicate implements Predicate<Node> {
+
+    protected final Collection<String> nodeTypes;
+
+    /**
+     * Base constructor for function peforming boolean ops on matched node types.
+     * @param types
+     */
+    public BooleanTypesPredicate(String... types) {
+        nodeTypes = Arrays.asList(types);
+    }
+
+    @Override
+    public boolean apply(Node input) {
+        if (input == null) {
+            throw new IllegalArgumentException(
+                    "null node passed to" + getClass().getName()
+            );
+        }
+        int matched = 0;
+        try {
+            for (NodeType nodeType: input.getMixinNodeTypes()) {
+                if (nodeTypes.contains(nodeType.getName())) {
+                    matched++;
+                }
+            }
+        } catch (RepositoryException e) {
+            propagate(e);
+        }
+        return test(matched);
+    }
+
+    protected abstract boolean test(int matched);
+
+}

fcrepo-kernel/src/main/java/org/fcrepo/kernel/utils/FedoraTypesUtils.java

@@ -54,6 +54,7 @@
 import javax.jcr.version.VersionHistory;
 
 import org.fcrepo.jcr.FedoraJcrTypes;
+import org.fcrepo.kernel.services.functions.AnyTypesPredicate;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.joda.time.format.DateTimeFormatter;
@@ -142,37 +143,22 @@ public boolean apply(final Node node) {
     /**
      * Predicate for determining whether this {@link Node} is a Fedora object.
      */
-    public static Predicate<Node> isFedoraObject = new Predicate<Node>() {
-
-        @Override
-        public boolean apply(final Node node) {
-            checkArgument(node != null, "null cannot be a Fedora object!");
-            try {
-                return map(node.getMixinNodeTypes(), nodetype2name).contains(
-                        FEDORA_OBJECT);
-            } catch (final RepositoryException e) {
-                throw propagate(e);
-            }
-        }
-    };
+    public static Predicate<Node> isFedoraObject =
+            new AnyTypesPredicate(FEDORA_OBJECT);
 
     /**
      * Predicate for determining whether this {@link Node} is a Fedora
      * datastream.
      */
-    public static Predicate<Node> isFedoraDatastream = new Predicate<Node>() {
+    public static Predicate<Node> isFedoraDatastream =
+            new AnyTypesPredicate(FEDORA_DATASTREAM);
 
-        @Override
-        public boolean apply(final Node node) {
-            checkArgument(node != null, "null cannot be a Fedora datastream!");
-            try {
-                return map(node.getMixinNodeTypes(), nodetype2name).contains(
-                        FEDORA_DATASTREAM);
-            } catch (final RepositoryException e) {
-                throw propagate(e);
-            }
-        }
-    };
+    /**
+     * Predicate for objects, datastreams, whatever!
+     */
+
+    public static Predicate<Node> isFedoraObjectOrDatastream =
+            new AnyTypesPredicate(FEDORA_OBJECT, FEDORA_DATASTREAM);
 
     /**
      * Translates a {@link NodeType} to its {@link String} name.