Adding SuppressByMixinFilter to suppress events related to nodes with any of a given set of mixins

fcrepo-jms/pom.xml

@@ -106,8 +106,6 @@
     <dependency>
       <groupId>com.jayway.awaitility</groupId>
       <artifactId>awaitility</artifactId>
-      <version>1.6.2</version>
-      <scope>test</scope>
     </dependency>
   </dependencies>
   <build>

fcrepo-kernel-impl/pom.xml

@@ -119,6 +119,10 @@
       <version>${project.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.jayway.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+    </dependency>
   </dependencies>
 
   <build>

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

@@ -15,8 +15,10 @@
  */
 package org.fcrepo.kernel.impl.observer;
 
+import static com.google.common.base.Functions.toStringFunction;
 import static com.google.common.base.Throwables.propagate;
-import static com.google.common.collect.Iterables.transform;
+import static com.google.common.collect.Collections2.transform;
+import static com.google.common.collect.Sets.newHashSet;
 import static org.fcrepo.kernel.FedoraJcrTypes.FEDORA_BINARY;
 import static org.fcrepo.kernel.FedoraJcrTypes.FEDORA_NON_RDF_SOURCE_DESCRIPTION;
 import static org.fcrepo.kernel.FedoraJcrTypes.FEDORA_CONTAINER;
@@ -26,19 +28,17 @@
 import javax.jcr.PathNotFoundException;
 import javax.jcr.RepositoryException;
 import javax.jcr.Session;
-import javax.jcr.nodetype.NodeType;
 import javax.jcr.observation.Event;
 
-import com.google.common.base.Function;
 import com.google.common.base.Predicate;
 
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableSet;
 import org.fcrepo.kernel.observer.EventFilter;
 import org.slf4j.Logger;
 
+import java.util.Arrays;
 import java.util.Collection;
-import java.util.List;
+import java.util.Collections;
+import java.util.HashSet;
 
 /**
  * {@link EventFilter} that passes only events emitted from nodes with a Fedora
@@ -58,12 +58,8 @@ public class DefaultFilter implements EventFilter {
 
     private static final Logger LOGGER = getLogger(DefaultFilter.class);
 
-    private static final Function<NodeType, String> nodetype2string = new Function<NodeType, String>() {
-        @Override
-        public String apply(final NodeType input) {
-            return input.getName();
-        }
-    };
+    private static final HashSet<String> fedoraMixins =
+            newHashSet(FEDORA_BINARY, FEDORA_CONTAINER, FEDORA_NON_RDF_SOURCE_DESCRIPTION, FEDORA_RESOURCE);
 
     /**
      * Default constructor.
@@ -79,14 +75,7 @@ public Predicate<Event> getFilter(final Session session) {
     @Override
     public boolean apply(final Event event) {
         try {
-            final org.modeshape.jcr.api.observation.Event modeEvent = getJcr21Event(event);
-
-            final List<NodeType> nodeTypes = ImmutableList.copyOf(modeEvent.getMixinNodeTypes());
-            final Collection<String> mixinTypes = ImmutableSet.copyOf(transform(nodeTypes, nodetype2string));
-            return mixinTypes.contains(FEDORA_RESOURCE)
-                    || mixinTypes.contains(FEDORA_BINARY)
-                    || mixinTypes.contains(FEDORA_NON_RDF_SOURCE_DESCRIPTION)
-                    || mixinTypes.contains(FEDORA_CONTAINER);
+            return !Collections.disjoint(getMixinTypes(event), fedoraMixins);
         } catch (final PathNotFoundException e) {
             LOGGER.trace("Dropping event from outside our assigned workspace:\n", e);
             return false;
@@ -95,9 +84,12 @@ public boolean apply(final Event event) {
         }
     }
 
-    private static org.modeshape.jcr.api.observation.Event getJcr21Event(final Event event) {
+    protected static Collection<String> getMixinTypes(final Event event)
+            throws PathNotFoundException, RepositoryException {
         try {
-            return (org.modeshape.jcr.api.observation.Event) event;
+            final org.modeshape.jcr.api.observation.Event modeEvent =
+                    (org.modeshape.jcr.api.observation.Event) event;
+            return transform(Arrays.asList(modeEvent.getMixinNodeTypes()), toStringFunction());
         } catch (final ClassCastException e) {
             throw new ClassCastException(event + " is not a Modeshape Event");
         }

fcrepo-kernel-impl/src/main/java/org/fcrepo/kernel/impl/observer/SuppressByMixinFilter.java

@@ -0,0 +1,73 @@
+/**
+ * Copyright 2015 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.impl.observer;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import javax.jcr.PathNotFoundException;
+import javax.jcr.RepositoryException;
+import javax.jcr.Session;
+import javax.jcr.observation.Event;
+
+import com.google.common.base.Predicate;
+
+import org.fcrepo.kernel.exception.RepositoryRuntimeException;
+import org.fcrepo.kernel.observer.EventFilter;
+import org.slf4j.Logger;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * {@link EventFilter} that extends {@link DefaultFilter} to also suppress events
+ * emitted by nodes with a provided set of mixins.
+ *
+ * @author escowles
+ * @since 2015-04-15
+ */
+public class SuppressByMixinFilter extends DefaultFilter implements EventFilter {
+
+    private static final Logger LOGGER = getLogger(SuppressByMixinFilter.class);
+    private Set<String> suppressedMixins;
+
+    /**
+     * Default constructor.
+     */
+    public SuppressByMixinFilter(final Set<String> suppressedMixins) {
+        this.suppressedMixins = suppressedMixins;
+        for (String mixin : suppressedMixins) {
+            LOGGER.info("Suppressing events for nodes with mixin: {}", mixin);
+        }
+    }
+
+    @Override
+    public Predicate<Event> getFilter(final Session session) {
+        return this;
+    }
+
+    @Override
+    public boolean apply(final Event event) {
+        try {
+            return super.apply(event) && Collections.disjoint(getMixinTypes(event), suppressedMixins);
+        } catch (final PathNotFoundException e) {
+            LOGGER.trace("Dropping event from outside our assigned workspace:\n", e);
+            return false;
+        } catch (final RepositoryException e) {
+            throw new RepositoryRuntimeException(e);
+        }
+    }
+
+}

fcrepo-kernel-impl/src/test/java/org/fcrepo/integration/kernel/impl/observer/SuppressByMixinFilterIT.java

@@ -0,0 +1,138 @@
+/**
+ * Copyright 2015 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.integration.kernel.impl.observer;
+
+import static com.jayway.awaitility.Awaitility.await;
+import static com.jayway.awaitility.Duration.ONE_SECOND;
+import static org.fcrepo.kernel.FedoraJcrTypes.FEDORA_CONTAINER;
+import static org.fcrepo.kernel.FedoraJcrTypes.LDP_DIRECT_CONTAINER;
+import static org.fcrepo.kernel.RdfLexicon.REPOSITORY_NAMESPACE;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Callable;
+
+import javax.inject.Inject;
+import javax.jcr.Node;
+import javax.jcr.Repository;
+import javax.jcr.RepositoryException;
+import javax.jcr.Session;
+
+import org.fcrepo.integration.kernel.impl.AbstractIT;
+import org.fcrepo.kernel.observer.FedoraEvent;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.test.context.ContextConfiguration;
+
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
+
+
+/**
+ * <p>SuppressByMixinFilterIT class.</p>
+ *
+ * @author escowles
+ * @since 2015-04-15
+ */
+@ContextConfiguration({"/spring-test/eventing-suppress.xml", "/spring-test/repo.xml"})
+public class SuppressByMixinFilterIT extends AbstractIT {
+
+    /**
+     * Time to wait for a set of test messages, in milliseconds.
+     */
+    private static final long TIMEOUT = 20000;
+
+    @Inject
+    private Repository repository;
+
+    @Inject
+    private EventBus eventBus;
+
+    private Set<String> eventsReceived = new HashSet<>();
+
+    @Test(timeout = TIMEOUT)
+    public void shouldSuppressWithMixin() throws RepositoryException {
+
+        final Session se = repository.login();
+        try {
+            // add node with suppressed mixin ldp:DirectContainer
+            final Node node = se.getRootNode().addNode("/object1");
+            node.addMixin(FEDORA_CONTAINER);
+            node.addMixin(LDP_DIRECT_CONTAINER);
+
+            // add second node without suppressed mixin
+            se.getRootNode().addNode("/object2").addMixin(FEDORA_CONTAINER);
+            se.save();
+
+            // should only see second node
+            waitForEvent("/object2");
+            assertFalse("Event not suppressed!", eventsReceived.contains("/object1"));
+        } catch (final RepositoryException e) {
+            se.logout();
+        }
+    }
+
+    @Test(timeout = TIMEOUT)
+    public void shouldAllowWithoutMixin() throws RepositoryException {
+
+        final Session se = repository.login();
+        try {
+            se.getRootNode().addNode("/object3").addMixin(FEDORA_CONTAINER);
+            se.save();
+
+            // should see one message
+            waitForEvent("/object3");
+        } catch (final RepositoryException e) {
+            se.logout();
+        }
+    }
+
+    @Before
+    public void acquireConnections() {
+        eventBus.register(this);
+    }
+
+    @After
+    public void releaseConnections() {
+        eventBus.unregister(this);
+    }
+
+    @Subscribe
+    public void receiveEvents(final FedoraEvent e) throws RepositoryException {
+        final Set<String> properties = e.getProperties();
+        assertNotNull(properties);
+
+        final String expected = REPOSITORY_NAMESPACE + "mixinTypes";
+        assertTrue("Should contain: " + expected + properties, properties.contains(expected));
+
+        eventsReceived.add(e.getPath());
+    }
+
+    private void waitForEvent(final String id) {
+        await().pollInterval(ONE_SECOND).until(new Callable<Boolean>() {
+
+            @Override
+            public Boolean call() {
+                return eventsReceived.contains(id);
+            }
+        });
+    }
+
+}

fcrepo-kernel-impl/src/test/java/org/fcrepo/kernel/impl/observer/DefaultFilterTest.java

@@ -70,10 +70,10 @@ public class DefaultFilterTest {
     public void setUp() {
         initMocks(this);
         testObj = new DefaultFilter();
-        when(fedoraResource.getName()).thenReturn(FEDORA_RESOURCE);
-        when(fedoraContainer.getName()).thenReturn(FEDORA_CONTAINER);
-        when(fedoraDatastream.getName()).thenReturn(FEDORA_NON_RDF_SOURCE_DESCRIPTION);
-        when(fedoraBinary.getName()).thenReturn(FEDORA_BINARY);
+        when(fedoraResource.toString()).thenReturn(FEDORA_RESOURCE);
+        when(fedoraContainer.toString()).thenReturn(FEDORA_CONTAINER);
+        when(fedoraDatastream.toString()).thenReturn(FEDORA_NON_RDF_SOURCE_DESCRIPTION);
+        when(fedoraBinary.toString()).thenReturn(FEDORA_BINARY);
     }
 
     @Test