You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@curator.apache.org by ra...@apache.org on 2020/03/29 20:53:21 UTC
[curator] branch
CURATOR-549-zk36-persistent-watcher-tree-cache-iterator updated (9b3c2b7 ->
dace899)
This is an automated email from the ASF dual-hosted git repository.
randgalt pushed a change to branch CURATOR-549-zk36-persistent-watcher-tree-cache-iterator
in repository https://gitbox.apache.org/repos/asf/curator.git.
omit 9b3c2b7 CURATOR-549
new dace899 CURATOR-549
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (9b3c2b7)
\
N -- N -- N refs/heads/CURATOR-549-zk36-persistent-watcher-tree-cache-iterator (dace899)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../org/apache/curator/framework/recipes/cache/TreeCacheIterator.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
[curator] 01/01: CURATOR-549
Posted by ra...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
randgalt pushed a commit to branch CURATOR-549-zk36-persistent-watcher-tree-cache-iterator
in repository https://gitbox.apache.org/repos/asf/curator.git
commit dace8998724b20d29511c8eaba8e75de176ff524
Author: randgalt <ra...@apache.org>
AuthorDate: Sun Mar 29 15:33:13 2020 -0500
CURATOR-549
The next phase of this issue will implement a bridge cache that bridges TreeCache for pre 3.6 SK and CuratorCache for ZK 3.6+. That bridge will need this TreeCache iterator.
---
.../curator/framework/recipes/cache/TreeCache.java | 16 +-
.../framework/recipes/cache/TreeCacheIterator.java | 101 +++++++++++
.../recipes/cache/TestTreeCacheIterator.java | 195 +++++++++++++++++++++
3 files changed, 310 insertions(+), 2 deletions(-)
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
index 3bf804c..fbcd265 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
@@ -47,6 +47,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -217,7 +218,7 @@ public class TreeCache implements Closeable
private static final ChildData DEAD = new ChildData("/", null, null);
- private static boolean isLive(ChildData cd)
+ static boolean isLive(ChildData cd)
{
return cd != null && cd != DEAD;
}
@@ -226,7 +227,7 @@ public class TreeCache implements Closeable
private static final AtomicReferenceFieldUpdater<TreeNode, ConcurrentMap<String, TreeNode>> childrenUpdater = (AtomicReferenceFieldUpdater)AtomicReferenceFieldUpdater.newUpdater(TreeNode.class, ConcurrentMap.class, "children");
- private final class TreeNode implements Watcher, BackgroundCallback
+ final class TreeNode implements Watcher, BackgroundCallback
{
volatile ChildData childData;
final TreeNode parent;
@@ -750,6 +751,17 @@ public class TreeCache implements Closeable
return isLive(result) ? result : null;
}
+ /**
+ * Return an iterator over all nodes in the cache. There are no
+ * guarantees of accuracy; this is merely the most recent view of the data.
+ *
+ * @return a possibly-empty iterator of nodes in the cache
+ */
+ public Iterator<ChildData> iterator()
+ {
+ return new TreeCacheIterator(root);
+ }
+
private void callListeners(final TreeCacheEvent event)
{
listeners.forEach(listener ->
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheIterator.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheIterator.java
new file mode 100644
index 0000000..eed42dc
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheIterator.java
@@ -0,0 +1,101 @@
+/**
+ * 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.curator.framework.recipes.cache;
+
+import com.google.common.collect.Iterators;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.NoSuchElementException;
+
+// depth first iterator over tree cache nodes
+class TreeCacheIterator implements Iterator<ChildData>
+{
+ private final LinkedList<Current> stack = new LinkedList<>();
+ private Current current;
+
+ private static class Current
+ {
+ final Iterator<TreeCache.TreeNode> iterator;
+ TreeCache.TreeNode node;
+
+ Current(Iterator<TreeCache.TreeNode> iterator)
+ {
+ this.iterator = iterator;
+ node = iterator.next();
+ }
+ }
+
+ TreeCacheIterator(TreeCache.TreeNode root)
+ {
+ current = new Current(Iterators.forArray(root));
+ stack.push(current);
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return (current != null) && TreeCache.isLive(current.node.childData);
+ }
+
+ @Override
+ public ChildData next()
+ {
+ if ( current == null )
+ {
+ throw new NoSuchElementException();
+ }
+
+ ChildData result = current.node.childData; // result of next iteration is current node's data
+
+ // set the next node for the next iteration (or note completion)
+
+ do
+ {
+ setNext();
+ } while ( (current != null) && !TreeCache.isLive(current.node.childData) );
+
+ return result;
+ }
+
+ private void setNext()
+ {
+ if ( current.node.children != null )
+ {
+ stack.push(current);
+ current = new Current(current.node.children.values().iterator());
+ }
+ else while ( true )
+ {
+ if ( current.iterator.hasNext() )
+ {
+ current.node = current.iterator.next();
+ break;
+ }
+ else if ( stack.size() > 0 )
+ {
+ current = stack.pop();
+ }
+ else
+ {
+ current = null; // done
+ break;
+ }
+ }
+ }
+}
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCacheIterator.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCacheIterator.java
new file mode 100644
index 0000000..aac967f
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCacheIterator.java
@@ -0,0 +1,195 @@
+/**
+ * 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.curator.framework.recipes.cache;
+
+import com.google.common.collect.Sets;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.compatibility.CuratorTestBase;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ThreadLocalRandom;
+
+public class TestTreeCacheIterator extends CuratorTestBase
+{
+ @Test
+ public void testBasic() throws Exception
+ {
+ final String[] nodes = {
+ "/base/test",
+ "/base/test/3",
+ "/base/test/3/0",
+ "/base/test/3/0/0",
+ "/base/test/3/0/1",
+ "/base/test/3/1",
+ "/base/test/3/1/0",
+ "/base/test/3/1/1",
+ "/base/test/3/2",
+ "/base/test/3/2/0",
+ "/base/test/3/2/1",
+ "/base/test/3/2/3",
+ "/base/test/3/3",
+ "/base/test/3/3/1",
+ "/base/test/3/3/3"
+ };
+
+ try ( CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)) )
+ {
+ client.start();
+
+ String basePath = "/base/test";
+ try (TreeCache treeCache = new TreeCache(client, basePath) )
+ {
+ treeCache.start();
+
+ for ( String node : nodes )
+ {
+ client.create().creatingParentsIfNeeded().forPath(node, node.getBytes());
+ }
+
+ timing.sleepABit(); // let the cache settle
+
+ Iterator<ChildData> iterator = treeCache.iterator();
+ Map<String, byte[]> iteratorValues = new HashMap<>();
+ while ( iterator.hasNext() )
+ {
+ ChildData next = iterator.next();
+ iteratorValues.put(next.getPath(), next.getData());
+ }
+
+ Assert.assertEquals(iteratorValues.size(), nodes.length);
+ for ( String node : nodes )
+ {
+ Assert.assertEquals(iteratorValues.get(node), node.getBytes());
+ }
+ }
+ }
+ }
+
+ @Test
+ public void testIteratorWithRandomGraph() throws Exception
+ {
+ Map<String, String> pathAndData = new HashMap<>();
+ ThreadLocalRandom random = ThreadLocalRandom.current();
+ int nodeQty = random.nextInt(100, 200);
+ int maxPerRow = random.nextInt(1, 10);
+ int maxDepth = random.nextInt(3, 5);
+ try ( CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)) )
+ {
+ client.start();
+
+ String basePath = "/base/test";
+ try (TreeCache treeCache = new TreeCache(client, basePath) )
+ {
+ treeCache.start();
+
+ client.create().creatingParentsIfNeeded().forPath(basePath, "0".getBytes());
+ pathAndData.put(basePath, "0");
+
+ while ( nodeQty-- > 0 )
+ {
+ int thisDepth = random.nextInt(1, maxDepth + 1);
+ StringBuilder path = new StringBuilder(basePath);
+ for ( int i = 0; i < thisDepth; ++i )
+ {
+ path.append("/").append(random.nextInt(maxPerRow));
+ long value = random.nextLong();
+ pathAndData.put(path.toString(), Long.toString(value));
+ client.create().orSetData().forPath(path.toString(), Long.toString(value).getBytes());
+ }
+ }
+
+ timing.sleepABit(); // let the cache settle
+
+ // at this point we have a cached graph of random nodes with random values
+ Iterator<ChildData> iterator = treeCache.iterator();
+ while ( iterator.hasNext() )
+ {
+ ChildData next = iterator.next();
+ Assert.assertTrue(pathAndData.containsKey(next.getPath()));
+ Assert.assertEquals(pathAndData.get(next.getPath()).getBytes(), next.getData());
+ pathAndData.remove(next.getPath());
+ }
+
+ Assert.assertEquals(pathAndData.size(), 0); // above loop should have removed all nodes
+ }
+ }
+ }
+
+ @Test
+ public void testEmptyTree() throws Exception
+ {
+ try (CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)))
+ {
+ client.start();
+
+ try (TreeCache treeCache = new TreeCache(client, "/base/test"))
+ {
+ treeCache.start();
+
+ Iterator<ChildData> iterator = treeCache.iterator();
+ Assert.assertFalse(iterator.hasNext());
+ }
+ }
+ }
+
+ @Test
+ public void testWithDeletedNodes() throws Exception
+ {
+ try (CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1)))
+ {
+ client.start();
+
+ try (TreeCache treeCache = new TreeCache(client, "/foo"))
+ {
+ treeCache.start();
+
+ client.create().forPath("/foo");
+ client.create().forPath("/foo/a1");
+ client.create().forPath("/foo/a2");
+ client.create().forPath("/foo/a2/a2.1");
+ client.create().forPath("/foo/a2/a2.2");
+ client.create().forPath("/foo/a3");
+ client.create().forPath("/foo/a3/a3.1");
+ client.create().forPath("/foo/a3/a3.2");
+
+ client.delete().forPath("/foo/a2/a2.2");
+ client.delete().forPath("/foo/a3/a3.1");
+
+ timing.sleepABit(); // let the cache settle
+
+ Iterator<ChildData> iterator = treeCache.iterator();
+ Set<String> paths = new HashSet<>();
+ while ( iterator.hasNext() )
+ {
+ ChildData next = iterator.next();
+ paths.add(next.getPath());
+ }
+
+ Assert.assertEquals(paths, Sets.newHashSet("/foo", "/foo/a1", "/foo/a2", "/foo/a2/a2.1", "/foo/a3", "/foo/a3/a3.2"));
+ }
+ }
+ }
+}