AtomixDocumentTree support for filtering notifications by DocumentPath

Change-Id: I3f4f616bc4f2e488e5433e44f72bcd121b564b0d

@@ -17,10 +17,14 @@
 package org.onosproject.store.service;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Objects;
 
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang.StringUtils;
+
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
@@ -103,6 +107,46 @@ public class DocumentPath implements Comparable<DocumentPath> {
         return ImmutableList.copyOf(pathElements);
     }
 
+    /**
+     * Returns if the specified path belongs to a direct ancestor of the node pointed at by this path.
+     * <p>
+     * Example: {@code root.a} is a direct ancestor of {@code r.a.b.c}; while {@code r.a.x} is not.
+     *
+     * @param other other path
+     * @return {@code true} is yes; {@code false} otherwise.
+     */
+    public boolean isAncestorOf(DocumentPath other) {
+        return !other.equals(this) && other.toString().startsWith(toString());
+    }
+
+    /**
+     * Returns if the specified path is belongs to a subtree rooted this path.
+     * <p>
+     * Example: {@code root.a.b} and {@code root.a.b.c.d.e} are descendants of {@code r.a.b};
+     * while {@code r.a.x.c} is not.
+     *
+     * @param other other path
+     * @return {@code true} is yes; {@code false} otherwise.
+     */
+    public boolean isDescendentOf(DocumentPath other) {
+        return other.equals(this) || other.isAncestorOf(this);
+    }
+
+    /**
+     * Returns the path that points to the least common ancestor of the specified
+     * collection of paths.
+     * @param paths collection of path
+     * @return path to least common ancestor
+     */
+    public static DocumentPath leastCommonAncestor(Collection<DocumentPath> paths) {
+        if (CollectionUtils.isEmpty(paths)) {
+            return null;
+        }
+        return DocumentPath.from(StringUtils.getCommonPrefix(paths.stream()
+                    .map(DocumentPath::toString)
+                    .toArray(String[]::new)));
+    }
+
     @Override
     public int hashCode() {
         return Objects.hash(pathElements);

+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * 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.onosproject.store.service;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link DocumentPath}.
+ */
+public class DocumentPathTest {
+
+    @Test
+    public void testConstruction() {
+        DocumentPath path = DocumentPath.from("root.a.b");
+        assertEquals(path.pathElements(), Arrays.asList("root", "a", "b"));
+        assertEquals(DocumentPath.from("root.a"), path.parent());
+    }
+
+    @Test
+    public void testAncestry() {
+        DocumentPath path1 = DocumentPath.from("root.a.b");
+        DocumentPath path2 = DocumentPath.from("root.a.d");
+        DocumentPath path3 = DocumentPath.from("root.a.b.c");
+        DocumentPath lca = DocumentPath.leastCommonAncestor(Arrays.asList(path1, path2, path3));
+        assertEquals(DocumentPath.from("root.a"), lca);
+        assertTrue(path1.isAncestorOf(path3));
+        assertFalse(path1.isAncestorOf(path2));
+        assertTrue(path3.isDescendentOf(path3));
+        assertTrue(path3.isDescendentOf(path1));
+        assertFalse(path3.isDescendentOf(path2));
+    }
+}

@@ -33,11 +33,11 @@ import java.util.concurrent.Executor;
 
 import org.onlab.util.Match;
 import org.onlab.util.Tools;
-import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Unlisten;
 import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Clear;
 import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Get;
 import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.GetChildren;
 import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Listen;
+import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Unlisten;
 import org.onosproject.store.primitives.resources.impl.AtomixDocumentTreeCommands.Update;
 import org.onosproject.store.service.AsyncDocumentTree;
 import org.onosproject.store.service.DocumentPath;
@@ -56,7 +56,7 @@ import com.google.common.util.concurrent.MoreExecutors;
 public class AtomixDocumentTree extends AbstractResource<AtomixDocumentTree>
     implements AsyncDocumentTree<byte[]> {
 
-    private final Map<DocumentTreeListener<byte[]>, Executor> eventListeners = new HashMap<>();
+    private final Map<DocumentTreeListener<byte[]>, InternalListener> eventListeners = new HashMap<>();
     public static final String CHANGE_SUBJECT = "changeEvents";
 
     protected AtomixDocumentTree(CopycatClient client, Properties options) {
@@ -184,21 +184,21 @@ public class AtomixDocumentTree extends AbstractResource<AtomixDocumentTree>
     public CompletableFuture<Void> addListener(DocumentPath path, DocumentTreeListener<byte[]> listener) {
         checkNotNull(path);
         checkNotNull(listener);
+        InternalListener internalListener = new InternalListener(path, listener, MoreExecutors.directExecutor());
         // TODO: Support API that takes an executor
-        if (isListening()) {
-            eventListeners.putIfAbsent(listener, MoreExecutors.directExecutor());
-            return CompletableFuture.completedFuture(null);
-        } else {
+        if (!eventListeners.containsKey(listener)) {
             return client.submit(new Listen(path))
-                         .thenRun(() -> eventListeners.put(listener, MoreExecutors.directExecutor()));
+                         .thenRun(() -> eventListeners.put(listener, internalListener));
         }
+        return CompletableFuture.completedFuture(null);
     }
 
     @Override
     public CompletableFuture<Void> removeListener(DocumentTreeListener<byte[]> listener) {
         checkNotNull(listener);
-        if (eventListeners.remove(listener) != null && eventListeners.isEmpty()) {
-            return client.submit(new Unlisten()).thenApply(v -> null);
+        InternalListener internalListener = eventListeners.remove(listener);
+        if  (internalListener != null && eventListeners.isEmpty()) {
+            return client.submit(new Unlisten(internalListener.path)).thenApply(v -> null);
         }
         return CompletableFuture.completedFuture(null);
     }
@@ -213,7 +213,26 @@ public class AtomixDocumentTree extends AbstractResource<AtomixDocumentTree>
     }
 
     private void processTreeUpdates(List<DocumentTreeEvent<byte[]>> events) {
-        events.forEach(event ->
-            eventListeners.forEach((listener, executor) -> executor.execute(() -> listener.event(event))));
+        events.forEach(event -> eventListeners.values().forEach(listener -> listener.event(event)));
+    }
+
+    private class InternalListener implements DocumentTreeListener<byte[]> {
+
+        private final DocumentPath path;
+        private final DocumentTreeListener<byte[]> listener;
+        private final Executor executor;
+
+        public InternalListener(DocumentPath path, DocumentTreeListener<byte[]> listener, Executor executor) {
+            this.path = path;
+            this.listener = listener;
+            this.executor = executor;
+        }
+
+        @Override
+        public void event(DocumentTreeEvent<byte[]> event) {
+            if (event.path().isDescendentOf(path)) {
+                executor.execute(() -> listener.event(event));
+            }
+        }
     }
 }

@@ -225,11 +225,10 @@ public class AtomixDocumentTreeCommands {
         }
 
         @Override
-        public void writeObject(BufferOutput<?> buffer, Serializer serializer) {
-        }
-
-        @Override
-        public void readObject(BufferInput<?> buffer, Serializer serializer) {
+        public String toString() {
+            return MoreObjects.toStringHelper(getClass())
+                    .add("path", path())
+                    .toString();
         }
     }
 
@@ -248,11 +247,10 @@ public class AtomixDocumentTreeCommands {
         }
 
         @Override
-        public void writeObject(BufferOutput<?> buffer, Serializer serializer) {
-        }
-
-        @Override
-        public void readObject(BufferInput<?> buffer, Serializer serializer) {
+        public String toString() {
+            return MoreObjects.toStringHelper(getClass())
+                    .add("path", path())
+                    .toString();
         }
     }
 

@@ -26,7 +26,10 @@ import io.atomix.copycat.server.storage.snapshot.SnapshotReader;
 import io.atomix.copycat.server.storage.snapshot.SnapshotWriter;
 import io.atomix.resource.ResourceStateMachine;
 
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
@@ -52,7 +55,7 @@ import org.onosproject.store.service.Versioned;
 import org.slf4j.Logger;
 
 import com.google.common.base.Throwables;
-import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Queues;
 
@@ -64,7 +67,7 @@ public class AtomixDocumentTreeState
     implements SessionListener, Snapshottable {
 
     private final Logger log = getLogger(getClass());
-    private final Map<Long, Commit<? extends Listen>> listeners = new HashMap<>();
+    private final Map<Long, SessionListenCommits> listeners = new HashMap<>();
     private AtomicLong versionCounter = new AtomicLong(0);
     private final DocumentTree<TreeNodeValue> docTree = new DefaultDocumentTree<>(versionCounter::incrementAndGet);
 
@@ -97,25 +100,23 @@ public class AtomixDocumentTreeState
 
     protected void listen(Commit<? extends Listen> commit) {
         Long sessionId = commit.session().id();
-        if (listeners.putIfAbsent(sessionId, commit) != null) {
-            commit.close();
-            return;
-        }
+        listeners.computeIfAbsent(sessionId, k -> new SessionListenCommits()).add(commit);
         commit.session().onStateChange(
                         state -> {
                             if (state == ServerSession.State.CLOSED
                                     || state == ServerSession.State.EXPIRED) {
-                                Commit<? extends Listen> listener = listeners.remove(sessionId);
-                                if (listener != null) {
-                                    listener.close();
-                                }
+                                closeListener(commit.session().id());
                             }
                         });
     }
 
     protected void unlisten(Commit<? extends Unlisten> commit) {
+        Long sessionId = commit.session().id();
         try {
-            closeListener(commit.session().id());
+            SessionListenCommits listenCommits = listeners.get(sessionId);
+            if (listenCommits != null) {
+                listenCommits.remove(commit);
+            }
         } finally {
             commit.close();
         }
@@ -261,10 +262,11 @@ public class AtomixDocumentTreeState
                         result.created() ? Type.CREATED : result.newValue() == null ? Type.DELETED : Type.UPDATED,
                         Optional.ofNullable(result.newValue()),
                         Optional.ofNullable(result.oldValue()));
+
         listeners.values()
-                 .forEach(commit -> commit.session()
-                                          .publish(AtomixDocumentTree.CHANGE_SUBJECT,
-                                                   ImmutableList.of(event)));
+                 .stream()
+                 .filter(l -> event.path().isDescendentOf(l.leastCommonAncestorPath()))
+                 .forEach(listener -> listener.publish(AtomixDocumentTree.CHANGE_SUBJECT, Arrays.asList(event)));
     }
 
     @Override
@@ -287,9 +289,52 @@ public class AtomixDocumentTreeState
     }
 
     private void closeListener(Long sessionId) {
-        Commit<? extends Listen> commit = listeners.remove(sessionId);
-        if (commit != null) {
-            commit.close();
+        SessionListenCommits listenCommits = listeners.remove(sessionId);
+        if (listenCommits != null) {
+            listenCommits.close();
+        }
+    }
+
+    private class SessionListenCommits {
+        private final List<Commit<? extends Listen>> commits = Lists.newArrayList();
+        private DocumentPath leastCommonAncestorPath;
+
+        public void add(Commit<? extends Listen> commit) {
+            commits.add(commit);
+            recomputeLeastCommonAncestor();
+        }
+
+        public void remove(Commit<? extends Unlisten> commit) {
+            // Remove the first listen commit with path matching path in unlisten commit
+            Iterator<Commit<? extends Listen>> iterator = commits.iterator();
+            while (iterator.hasNext()) {
+                Commit<? extends Listen> listenCommit = iterator.next();
+                if (listenCommit.operation().path().equals(commit.operation().path())) {
+                    iterator.remove();
+                    listenCommit.close();
+                }
+            }
+            recomputeLeastCommonAncestor();
+        }
+
+        public DocumentPath leastCommonAncestorPath() {
+            return leastCommonAncestorPath;
+        }
+
+        public <M> void publish(String topic, M message) {
+            commits.stream().findAny().ifPresent(commit -> commit.session().publish(topic, message));
+        }
+
+        public void close() {
+            commits.forEach(Commit::close);
+            commits.clear();
+            leastCommonAncestorPath = null;
+        }
+
+        private void recomputeLeastCommonAncestor() {
+            this.leastCommonAncestorPath = DocumentPath.leastCommonAncestor(commits.stream()
+                    .map(c -> c.operation().path())
+                    .collect(Collectors.toList()));
         }
     }
 }

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import io.atomix.AtomixClient;
 import io.atomix.resource.ResourceType;
 
 import java.util.Map;
@@ -343,12 +344,53 @@ public class AtomixDocumentTreeTest extends AtomixTestBase {
         assertArrayEquals("xy".getBytes(), event.newValue().get().value());
     }
 
+    @Test
+    public void testFilteredNotifications() throws Throwable {
+        AtomixClient client1 = createAtomixClient();
+        AtomixClient client2 = createAtomixClient();
+
+        String treeName = UUID.randomUUID().toString();
+        AtomixDocumentTree tree1 = client1.getResource(treeName, AtomixDocumentTree.class).join();
+        AtomixDocumentTree tree2 = client2.getResource(treeName, AtomixDocumentTree.class).join();
+
+        TestEventListener listener1a = new TestEventListener(3);
+        TestEventListener listener1ab = new TestEventListener(2);
+        TestEventListener listener2abc = new TestEventListener(1);
+
+        tree1.addListener(DocumentPath.from("root.a"), listener1a).join();
+        tree1.addListener(DocumentPath.from("root.a.b"), listener1ab).join();
+        tree2.addListener(DocumentPath.from("root.a.b.c"), listener2abc).join();
+
+        tree1.createRecursive(DocumentPath.from("root.a.b.c"), "abc".getBytes()).join();
+        DocumentTreeEvent<byte[]> event = listener1a.event();
+        assertEquals(DocumentPath.from("root.a"), event.path());
+        event = listener1a.event();
+        assertEquals(DocumentPath.from("root.a.b"), event.path());
+        event = listener1a.event();
+        assertEquals(DocumentPath.from("root.a.b.c"), event.path());
+        event = listener1ab.event();
+        assertEquals(DocumentPath.from("root.a.b"), event.path());
+        event = listener1ab.event();
+        assertEquals(DocumentPath.from("root.a.b.c"), event.path());
+        event = listener2abc.event();
+        assertEquals(DocumentPath.from("root.a.b.c"), event.path());
+    }
+
     private static class TestEventListener implements DocumentTreeListener<byte[]> {
 
-        private final BlockingQueue<DocumentTreeEvent<byte[]>> queue = new ArrayBlockingQueue<>(1);
+        private final BlockingQueue<DocumentTreeEvent<byte[]>> queue;
+
+        public TestEventListener() {
+            this(1);
+        }
+
+        public TestEventListener(int maxEvents) {
+            queue = new ArrayBlockingQueue<>(maxEvents);
+        }
 
         @Override
         public void event(DocumentTreeEvent<byte[]> event) {
+
             try {
                 queue.put(event);
             } catch (InterruptedException e) {