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 2017/01/01 19:24:11 UTC
[1/2] curator git commit: Reworked filters. Went back to the
CacheSelector multi-method fitler used in TreeCache.
Repository: curator
Updated Branches:
refs/heads/persistent-watch add0d10bb -> 72fe88c3b
Reworked filters. Went back to the CacheSelector multi-method fitler used in TreeCache.
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/2cf7c412
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/2cf7c412
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/2cf7c412
Branch: refs/heads/persistent-watch
Commit: 2cf7c412caf81cb7846a7da5aac3adbc62502d3e
Parents: add0d10
Author: randgalt <ra...@apache.org>
Authored: Sun Jan 1 13:33:13 2017 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Jan 1 13:33:13 2017 -0500
----------------------------------------------------------------------
.../framework/recipes/watch/CacheFilter.java | 24 ---
.../framework/recipes/watch/CacheFilters.java | 118 --------------
.../framework/recipes/watch/CacheSelector.java | 69 +++++++++
.../framework/recipes/watch/CacheSelectors.java | 153 +++++++++++++++++++
.../recipes/watch/CuratorCacheBuilder.java | 58 +------
.../recipes/watch/InternalCuratorCache.java | 34 +++--
.../recipes/watch/InternalNodeCache.java | 8 +-
.../recipes/watch/NoDataCacheFilter.java | 28 ----
.../framework/recipes/watch/RefreshFilter.java | 24 ---
.../framework/recipes/watch/RefreshFilters.java | 71 ---------
.../recipes/watch/StandardCacheFilter.java | 46 ------
.../watch/TestSingleLevelCuratorCache.java | 28 ++--
.../TestSingleLevelCuratorCacheInCluster.java | 2 +-
.../framework/recipes/watch/TestTreeCache.java | 30 ++--
.../recipes/watch/TestTreeCacheRandomTree.java | 74 +++------
15 files changed, 297 insertions(+), 470 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilter.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilter.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilter.java
deleted file mode 100644
index 6cccc88..0000000
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilter.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * 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.watch;
-
-public interface CacheFilter
-{
- CacheAction actionForPath(String mainPath, String checkPath);
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilters.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilters.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilters.java
deleted file mode 100644
index aa06a2d..0000000
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheFilters.java
+++ /dev/null
@@ -1,118 +0,0 @@
-/**
- * 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.watch;
-
-import java.util.Objects;
-
-public class CacheFilters
-{
- private static final CacheFilter statAndData = new StandardCacheFilter(CacheAction.STAT_AND_DATA);
- private static final CacheFilter compressedStatAndData = new StandardCacheFilter(CacheAction.STAT_AND_COMPRESSED_DATA);
- private static final CacheFilter statOnly = new StandardCacheFilter(CacheAction.STAT_ONLY);
- private static final CacheFilter pathOnly = new StandardCacheFilter(CacheAction.PATH_ONLY);
- private static final CacheFilter fullStatAndData = new CacheFilter()
- {
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- return CacheAction.STAT_AND_DATA;
- }
- };
- private static final CacheFilter fullCompressedStatAndData = new CacheFilter()
- {
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- return CacheAction.STAT_AND_COMPRESSED_DATA;
- }
- };
- private static final CacheFilter fullStatOnly = new CacheFilter()
- {
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- return CacheAction.STAT_ONLY;
- }
- };
- private static final CacheFilter fullPathOnly = new CacheFilter()
- {
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- return CacheAction.PATH_ONLY;
- }
- };
-
- public static CacheFilter statAndData()
- {
- return statAndData;
- }
-
- public static CacheFilter compressedData()
- {
- return compressedStatAndData;
- }
-
- public static CacheFilter statOnly()
- {
- return statOnly;
- }
-
- public static CacheFilter pathOnly()
- {
- return pathOnly;
- }
-
- public static CacheFilter fullStatAndData()
- {
- return fullStatAndData;
- }
-
- public static CacheFilter fullCompressedStatAndData()
- {
- return fullCompressedStatAndData;
- }
-
- public static CacheFilter fullStatOnly()
- {
- return fullStatOnly;
- }
-
- public static CacheFilter fullPathOnly()
- {
- return fullPathOnly;
- }
-
- static CacheFilter maxDepth(int maxDepth, final CacheFilter mainFilter)
- {
- final RefreshFilter refreshFilter = RefreshFilters.maxDepth(maxDepth);
- return new CacheFilter()
- {
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- return refreshFilter.descend(mainPath, checkPath) ? mainFilter.actionForPath(mainPath, checkPath) : CacheAction.NOT_STORED;
- }
- };
- }
-
- private CacheFilters()
- {
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelector.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelector.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelector.java
new file mode 100644
index 0000000..6d1f50f
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelector.java
@@ -0,0 +1,69 @@
+/**
+ * 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.watch;
+
+/**
+ * <p>
+ * Controls which nodes a CuratorCache processes. When iterating
+ * over the children of a parent node, a given node's children are
+ * queried only if {@link #traverseChildren(String, String)} returns true.
+ * When caching the list of nodes for a parent node, a given node is
+ * stored only if {@link #actionForPath(String, String)} returns other than
+ * {@link CacheAction#NOT_STORED}.
+ * </p>
+ *
+ * <p>
+ * E.g. Given:
+ * <pre>
+ * root
+ * n1-a
+ * n1-b
+ * n2-a
+ * n2-b
+ * n3-a
+ * n1-c
+ * n1-d
+ * </pre>
+ * You could have a CuratorCache only work with the nodes: n1-a, n1-b, n2-a, n2-b, n1-d
+ * by returning false from traverseChildren() for "/root/n1-b/n2-b" and returning
+ * {@link CacheAction#NOT_STORED} from acceptChild("/root/n1-c").
+ * </p>
+ */
+public interface CacheSelector
+{
+ /**
+ * Return true if children of this path should be cached.
+ * i.e. if false is returned, this node is not queried to
+ * determine if it has children or not
+ *
+ * @param basePath the main path of the cache
+ * @param fullPath full path of the ZNode
+ * @return true/false
+ */
+ boolean traverseChildren(String basePath, String fullPath);
+
+ /**
+ * Return true if this node should be returned from the cache
+ *
+ * @param basePath the main path of the cache
+ * @param fullPath full path of the ZNode
+ * @return true/false
+ */
+ CacheAction actionForPath(String basePath, String fullPath);
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelectors.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelectors.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelectors.java
new file mode 100644
index 0000000..bbe27e9
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheSelectors.java
@@ -0,0 +1,153 @@
+/**
+ * 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.watch;
+
+import org.apache.curator.utils.ZKPaths;
+
+public class CacheSelectors
+{
+ private static final CacheSelector statAndData = new StandardCacheSelector(CacheAction.STAT_AND_DATA);
+ private static final CacheSelector compressedStatAndData = new StandardCacheSelector(CacheAction.STAT_AND_COMPRESSED_DATA);
+ private static final CacheSelector statOnly = new StandardCacheSelector(CacheAction.STAT_ONLY);
+ private static final CacheSelector pathOnly = new StandardCacheSelector(CacheAction.PATH_ONLY);
+
+ private static class StandardCacheSelector implements CacheSelector
+ {
+ private final CacheAction cacheAction;
+
+ private StandardCacheSelector(CacheAction cacheAction)
+ {
+ this.cacheAction = cacheAction;
+ }
+
+ @Override
+ public boolean traverseChildren(String basePath, String fullPath)
+ {
+ return true;
+ }
+
+ @Override
+ public CacheAction actionForPath(String basePath, String fullPath)
+ {
+ return cacheAction;
+ }
+ }
+
+ private static class SingleLevelCacheSelector implements CacheSelector
+ {
+ private final CacheAction cacheAction;
+
+ private SingleLevelCacheSelector(CacheAction cacheAction)
+ {
+ this.cacheAction = cacheAction;
+ }
+
+ @Override
+ public boolean traverseChildren(String basePath, String fullPath)
+ {
+ return basePath.equals(fullPath);
+ }
+
+ @Override
+ public CacheAction actionForPath(String mainPath, String checkPath)
+ {
+ boolean mainPathIsRoot = mainPath.endsWith(ZKPaths.PATH_SEPARATOR);
+ if ( mainPathIsRoot && checkPath.equals(ZKPaths.PATH_SEPARATOR) ) // special case. The parent of "/" is "/"
+ {
+ return CacheAction.NOT_STORED;
+ }
+ else if ( ZKPaths.getPathAndNode(checkPath).getPath().equals(mainPath) )
+ {
+ return cacheAction;
+ }
+ return CacheAction.NOT_STORED;
+ }
+ }
+
+ private static class MaxDepthCacheSelector implements CacheSelector
+ {
+ private final int maxDepth;
+ private final CacheAction cacheAction;
+
+ private MaxDepthCacheSelector(int maxDepth, CacheAction cacheAction)
+ {
+ this.maxDepth = maxDepth;
+ this.cacheAction = cacheAction;
+ }
+
+ @Override
+ public boolean traverseChildren(String basePath, String fullPath)
+ {
+ int mainPathDepth = ZKPaths.split(basePath).size();
+ int checkPathDepth = ZKPaths.split(fullPath).size();
+ int thisDepth = checkPathDepth - mainPathDepth;
+ return (thisDepth <= maxDepth);
+ }
+
+ @Override
+ public CacheAction actionForPath(String basePath, String fullPath)
+ {
+ return traverseChildren(basePath, fullPath) ? cacheAction : CacheAction.NOT_STORED;
+ }
+ }
+
+ public static CacheSelector statAndData()
+ {
+ return statAndData;
+ }
+
+ public static CacheSelector compressedStatAndData()
+ {
+ return compressedStatAndData;
+ }
+
+ public static CacheSelector statOnly()
+ {
+ return statOnly;
+ }
+
+ public static CacheSelector pathOnly()
+ {
+ return pathOnly;
+ }
+
+ public static CacheSelector singleLevel()
+ {
+ return new SingleLevelCacheSelector(CacheAction.STAT_AND_DATA);
+ }
+
+ public static CacheSelector singleLevel(CacheAction cacheAction)
+ {
+ return new SingleLevelCacheSelector(cacheAction);
+ }
+
+ public static CacheSelector maxDepth(int maxDepth)
+ {
+ return maxDepth(maxDepth, CacheAction.STAT_AND_DATA);
+ }
+
+ public static CacheSelector maxDepth(int maxDepth, CacheAction cacheAction)
+ {
+ return new MaxDepthCacheSelector(maxDepth, cacheAction);
+ }
+
+ private CacheSelectors()
+ {
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
index 7af34c1..014584a 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
@@ -18,7 +18,6 @@
*/
package org.apache.curator.framework.recipes.watch;
-import com.google.common.base.Preconditions;
import com.google.common.cache.CacheBuilder;
import org.apache.curator.framework.CuratorFramework;
import java.util.Objects;
@@ -30,20 +29,15 @@ public class CuratorCacheBuilder
private final String path;
private CacheBuilder<Object, Object> cacheBuilder = CacheBuilder.newBuilder();
private boolean singleNode = false;
- private RefreshFilter refreshFilter = null;
private boolean sendRefreshEvents = true;
private boolean refreshOnStart = true;
- private CacheFilter cacheFilter = CacheFilters.statAndData();
private boolean sortChildren = true;
private CachedNodeComparator nodeComparator = CachedNodeComparators.dataAndType();
- private int maxDepth = -1;
+ private CacheSelector cacheSelector = CacheSelectors.statAndData();
public static CuratorCacheBuilder builder(CuratorFramework client, String path)
{
- CuratorCacheBuilder builder = new CuratorCacheBuilder(client, path);
- builder.singleNode = false;
- builder.cacheFilter = CacheFilters.fullStatAndData();
- return builder;
+ return new CuratorCacheBuilder(client, path);
}
public CuratorCache build()
@@ -53,35 +47,12 @@ public class CuratorCacheBuilder
return buildSingleNode();
}
- CacheFilter localCacheFilter = this.cacheFilter;
- RefreshFilter localRefreshFilter = this.refreshFilter;
- if ( maxDepth >= 0 )
- {
- Preconditions.checkState(refreshFilter == null, "You cannot set both maxDepth and a refreshFilter");
- localRefreshFilter = RefreshFilters.maxDepth(maxDepth);
- localCacheFilter = CacheFilters.maxDepth(maxDepth, cacheFilter);
- }
- else if ( localRefreshFilter == null )
- {
- localRefreshFilter = RefreshFilters.tree();
- }
-
- return new InternalCuratorCache(client, path, nodeComparator, localCacheFilter, localRefreshFilter, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart, sortChildren);
+ return new InternalCuratorCache(client, path, cacheSelector, nodeComparator, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart, sortChildren);
}
public CuratorCacheBuilder forSingleNode()
{
singleNode = true;
- refreshFilter = null;
- cacheFilter = CacheFilters.statAndData();
- return this;
- }
-
- public CuratorCacheBuilder forSingleLevel()
- {
- singleNode = false;
- refreshFilter = RefreshFilters.singleLevel();
- cacheFilter = CacheFilters.statAndData();
return this;
}
@@ -109,18 +80,6 @@ public class CuratorCacheBuilder
return this;
}
- public CuratorCacheBuilder withCacheFilter(CacheFilter cacheFilter)
- {
- this.cacheFilter = Objects.requireNonNull(cacheFilter, "cacheFilter cannot be null");
- return this;
- }
-
- public CuratorCacheBuilder withRefreshFilter(RefreshFilter refreshFilter)
- {
- this.refreshFilter = Objects.requireNonNull(refreshFilter, "primingFilter cannot be null");
- return this;
- }
-
public CuratorCacheBuilder sendingRefreshEvents(boolean sendRefreshEvents)
{
this.sendRefreshEvents = sendRefreshEvents;
@@ -145,18 +104,15 @@ public class CuratorCacheBuilder
return this;
}
- public CuratorCacheBuilder withMaxDepth(int maxDepth)
+ public CuratorCacheBuilder withCacheSelector(CacheSelector cacheSelector)
{
- Preconditions.checkArgument(maxDepth >= 0, "maxDepth must be >= 0");
- this.maxDepth = maxDepth;
+ this.cacheSelector = Objects.requireNonNull(cacheSelector, "cacheSelector cannot be null");
return this;
}
- private CuratorCache buildSingleNode()
+ private InternalNodeCache buildSingleNode()
{
- Preconditions.checkState(refreshFilter == null, "Single node caches do not use RefreshFilters");
- Preconditions.checkState(maxDepth < 0, "Single node caches do not support maxDepth");
- return new InternalNodeCache(client, path, nodeComparator, cacheFilter, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart);
+ return null;// TODO new InternalNodeCache(client, path, cacheSelector, nodeComparator, cacheFilter, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart);
}
private CuratorCacheBuilder(CuratorFramework client, String path)
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
index bff8b2a..101ca6d 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
@@ -41,31 +41,35 @@ import java.util.concurrent.atomic.AtomicInteger;
class InternalCuratorCache extends CuratorCacheBase implements Watcher
{
+ private static final CachedNode nullNode = new CachedNode();
private final PersistentWatcher watcher;
private final CuratorFramework client;
private final String basePath;
+ private final CacheSelector cacheSelector;
private final CachedNodeComparator nodeComparator;
- private final CacheFilter cacheFilter;
- private final RefreshFilter refreshFilter;
private final boolean sortChildren;
- private static final CachedNode nullNode = new CachedNode();
- private static final RefreshFilter nopRefreshFilter = new RefreshFilter()
+ private final CacheSelector singleNodeCacheSelector = new CacheSelector()
{
@Override
- public boolean descend(String mainPath, String checkPath)
+ public boolean traverseChildren(String basePath, String fullPath)
{
return false;
}
+
+ @Override
+ public CacheAction actionForPath(String basePath, String fullPath)
+ {
+ return cacheSelector.actionForPath(basePath, fullPath);
+ }
};
- InternalCuratorCache(CuratorFramework client, String path, CachedNodeComparator nodeComparator, CacheFilter cacheFilter, final RefreshFilter refreshFilter, Cache<String, CachedNode> cache, boolean sendRefreshEvents, final boolean refreshOnStart, boolean sortChildren)
+ InternalCuratorCache(CuratorFramework client, String path, final CacheSelector cacheSelector, CachedNodeComparator nodeComparator, Cache<String, CachedNode> cache, boolean sendRefreshEvents, final boolean refreshOnStart, boolean sortChildren)
{
super(cache, sendRefreshEvents);
this.client = Objects.requireNonNull(client, "client cannot be null");
this.basePath = Objects.requireNonNull(path, "path cannot be null");
+ this.cacheSelector = Objects.requireNonNull(cacheSelector, "cacheSelector cannot be null");
this.nodeComparator = Objects.requireNonNull(nodeComparator, "nodeComparator cannot be null");
- this.cacheFilter = Objects.requireNonNull(cacheFilter, "cacheFilter cannot be null");
- this.refreshFilter = Objects.requireNonNull(refreshFilter, "primingFilter cannot be null");
this.sortChildren = sortChildren;
watcher = new PersistentWatcher(client, path)
{
@@ -75,7 +79,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
long count = refreshCount();
if ( (refreshOnStart && (count == 0)) || (count > 0) )
{
- internalRefresh(basePath, new NotifyingRefresher(InternalCuratorCache.this, basePath), refreshFilter);
+ internalRefresh(basePath, new NotifyingRefresher(InternalCuratorCache.this, basePath), cacheSelector);
}
}
};
@@ -114,7 +118,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
case NodeCreated:
case NodeDataChanged:
{
- internalRefresh(event.getPath(), new Refresher(InternalCuratorCache.this), nopRefreshFilter);
+ internalRefresh(event.getPath(), new Refresher(InternalCuratorCache.this), singleNodeCacheSelector);
break;
}
}
@@ -133,7 +137,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
{
CountDownLatch latch = new CountDownLatch(1);
Refresher refresher = new NotifyingRefresher(this, path, latch);
- internalRefresh(path, refresher, refreshFilter);
+ internalRefresh(path, refresher, cacheSelector);
return latch;
}
return new CountDownLatch(0);
@@ -142,7 +146,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
@VisibleForTesting
volatile Exchanger<Object> debugRebuildTestExchanger;
- private void internalRefresh(final String path, final Refresher refresher, final RefreshFilter refreshFilter)
+ private void internalRefresh(final String path, final Refresher refresher, final CacheSelector cacheSelector)
{
if ( !isStarted() )
{
@@ -180,7 +184,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
}
for ( String child : children )
{
- internalRefresh(ZKPaths.makePath(path, child), refresher, refreshFilter);
+ internalRefresh(ZKPaths.makePath(path, child), refresher, cacheSelector);
}
}
}
@@ -200,7 +204,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
}
};
- CacheAction cacheAction = cacheFilter.actionForPath(basePath, path);
+ CacheAction cacheAction = cacheSelector.actionForPath(basePath, path);
switch ( cacheAction )
{
case NOT_STORED:
@@ -241,7 +245,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
}
}
- if ( refreshFilter.descend(basePath, path) )
+ if ( cacheSelector.traverseChildren(basePath, path) )
{
refresher.increment();
try
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
index 0968e7a..d3f327a 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
@@ -34,7 +34,6 @@ import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Objects;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Exchanger;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -45,9 +44,9 @@ class InternalNodeCache extends CuratorCacheBase
private final Logger log = LoggerFactory.getLogger(getClass());
private final WatcherRemoveCuratorFramework client;
private final String path;
- private final CacheFilter cacheFilter;
private final AtomicReference<CachedNode> data = new AtomicReference<>(null);
private final AtomicBoolean isConnected = new AtomicBoolean(true);
+ private final CacheSelector cacheSelector = CacheSelectors.statAndData();
private static final CachedNode nullNode = new CachedNode();
private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
{
@@ -94,12 +93,11 @@ class InternalNodeCache extends CuratorCacheBase
}
};
- InternalNodeCache(CuratorFramework client, String path, CachedNodeComparator nodeComparator, CacheFilter cacheFilter, Cache<String, CachedNode> cache, boolean sendRefreshEvents, boolean refreshOnStart)
+ InternalNodeCache(CuratorFramework client, String path, CachedNodeComparator nodeComparator, Cache<String, CachedNode> cache, boolean sendRefreshEvents, boolean refreshOnStart)
{
super(cache, sendRefreshEvents);
this.client = client.newWatcherRemoveCuratorFramework();
this.path = PathUtils.validatePath(path);
- this.cacheFilter = Objects.requireNonNull(cacheFilter, "cacheFilter cannot be null");
}
@Override
@@ -179,7 +177,7 @@ class InternalNodeCache extends CuratorCacheBase
}
else if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
{
- switch ( cacheFilter.actionForPath(path, path) )
+ switch ( cacheSelector.actionForPath(path, path) )
{
default:
case NOT_STORED:
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/NoDataCacheFilter.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/NoDataCacheFilter.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/NoDataCacheFilter.java
deleted file mode 100644
index 97e86da..0000000
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/NoDataCacheFilter.java
+++ /dev/null
@@ -1,28 +0,0 @@
-/**
- * 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.watch;
-
-public class NoDataCacheFilter implements CacheFilter
-{
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- return CacheAction.STAT_ONLY;
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilter.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilter.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilter.java
deleted file mode 100644
index 8a29826..0000000
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilter.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * 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.watch;
-
-public interface RefreshFilter
-{
- boolean descend(String mainPath, String checkPath);
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilters.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilters.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilters.java
deleted file mode 100644
index a9f6b13..0000000
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/RefreshFilters.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/**
- * 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.watch;
-
-import org.apache.curator.utils.ZKPaths;
-
-public class RefreshFilters
-{
- private static final RefreshFilter singleLevel = new RefreshFilter()
- {
- @Override
- public boolean descend(String mainPath, String checkPath)
- {
- return mainPath.equals(checkPath);
- }
- };
-
- private static final RefreshFilter tree = new RefreshFilter()
- {
- @Override
- public boolean descend(String mainPath, String checkPath)
- {
- return true;
- }
- };
-
- public static RefreshFilter singleLevel()
- {
- return singleLevel;
- }
-
- public static RefreshFilter tree()
- {
- return tree;
- }
-
- static RefreshFilter maxDepth(final int maxDepth)
- {
- return new RefreshFilter()
- {
- @Override
- public boolean descend(String mainPath, String checkPath)
- {
- int mainPathDepth = ZKPaths.split(mainPath).size();
- int checkPathDepth = ZKPaths.split(checkPath).size();
- int thisDepth = checkPathDepth - mainPathDepth;
- return (thisDepth <= maxDepth);
- }
- };
- }
-
- private RefreshFilters()
- {
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/StandardCacheFilter.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/StandardCacheFilter.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/StandardCacheFilter.java
deleted file mode 100644
index e27355d..0000000
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/StandardCacheFilter.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/**
- * 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.watch;
-
-import org.apache.curator.utils.ZKPaths;
-
-class StandardCacheFilter implements CacheFilter
-{
- private final CacheAction cacheAction;
-
- StandardCacheFilter(CacheAction cacheAction)
- {
- this.cacheAction = cacheAction;
- }
-
- @Override
- public CacheAction actionForPath(String mainPath, String checkPath)
- {
- boolean mainPathIsRoot = mainPath.endsWith(ZKPaths.PATH_SEPARATOR);
- if ( mainPathIsRoot && checkPath.equals(ZKPaths.PATH_SEPARATOR) ) // special case. The parent of "/" is "/"
- {
- return CacheAction.NOT_STORED;
- }
- else if ( ZKPaths.getPathAndNode(checkPath).getPath().equals(mainPath) )
- {
- return cacheAction;
- }
- return CacheAction.NOT_STORED;
- }
-}
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCache.java
index 0305e55..2a3591b 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCache.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCache.java
@@ -56,7 +56,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
{
client.start();
- final CuratorCache cache = CuratorCacheBuilder.builder(client, "/").forSingleLevel().build();
+ final CuratorCache cache = CuratorCacheBuilder.builder(client, "/").withCacheSelector(CacheSelectors.singleLevel()).build();
final CountDownLatch addedLatch = new CountDownLatch(1);
CacheListener listener = new CacheListener()
{
@@ -114,7 +114,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.start();
final CountDownLatch latch = new CountDownLatch(1);
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build();
cache.getListenable().addListener(new CacheListener()
{
@Override
@@ -149,7 +149,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.create().forPath("/test/one", "hey there".getBytes());
final BlockingQueue<CacheEvent> events = new LinkedBlockingQueue<>();
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().sendingRefreshEvents(false).build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).sendingRefreshEvents(false).build();
cache.getListenable().addListener(new CacheListener()
{
@Override
@@ -183,7 +183,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.start();
client.create().forPath("/test");
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build();
final CountDownLatch addedLatch = new CountDownLatch(3);
final CountDownLatch initLatch = new CountDownLatch(1);
@@ -233,7 +233,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.start();
client.create().forPath("/test");
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().sendingRefreshEvents(false).build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).sendingRefreshEvents(false).build();
final CountDownLatch addedLatch = new CountDownLatch(3);
cache.getListenable().addListener(new CacheListener()
@@ -279,7 +279,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
final CountDownLatch updatedLatch = new CountDownLatch(1);
final CountDownLatch addedLatch = new CountDownLatch(1);
client.create().creatingParentsIfNeeded().forPath("/test");
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().withCacheFilter(CacheFilters.statOnly()).build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel(CacheAction.STAT_ONLY)).build();
cache.getListenable().addListener(new CacheListener()
{
@Override
@@ -337,7 +337,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
final CountDownLatch removedLatch = new CountDownLatch(1);
final CountDownLatch postRemovedLatch = new CountDownLatch(1);
final CountDownLatch dataLatch = new CountDownLatch(1);
- try ( CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build() )
+ try ( CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build() )
{
cache.getListenable().addListener(new CacheListener()
{
@@ -405,7 +405,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.create().forPath("/test/snafu", "original".getBytes());
final CountDownLatch addedLatch = new CountDownLatch(2);
- try ( final CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build() )
+ try ( final CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build() )
{
cache.getListenable().addListener(new CacheListener()
{
@@ -499,7 +499,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
final List<CacheEvent> events = Lists.newArrayList();
final Semaphore semaphore = new Semaphore(0);
- cache = CuratorCacheBuilder.builder(client, "/base").forSingleLevel().sendingRefreshEvents(false).build();
+ cache = CuratorCacheBuilder.builder(client, "/base").withCacheSelector(CacheSelectors.singleLevel()).sendingRefreshEvents(false).build();
cache.getListenable().addListener(new CacheListener()
{
@Override
@@ -553,7 +553,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
final List<CacheEvent> events = Lists.newArrayList();
final Semaphore semaphore = new Semaphore(0);
- try ( final CuratorCache cache = CuratorCacheBuilder.builder(client, "/base").forSingleLevel().sendingRefreshEvents(false).build() )
+ try ( final CuratorCache cache = CuratorCacheBuilder.builder(client, "/base").withCacheSelector(CacheSelectors.singleLevel()).sendingRefreshEvents(false).build() )
{
cache.getListenable().addListener(new CacheListener()
{
@@ -596,7 +596,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.start();
client.create().forPath("/test");
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build();
cache.start();
final CountDownLatch childAddedLatch = new CountDownLatch(1);
@@ -674,8 +674,8 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
private void internalTestMode(CuratorFramework client, boolean cacheData) throws Exception
{
- CacheFilter cacheFilter = cacheData ? CacheFilters.statAndData() : CacheFilters.statOnly();
- try (CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().withCacheFilter(cacheFilter).build() )
+ CacheSelector cacheSelector = CacheSelectors.singleLevel(cacheData ? CacheAction.STAT_AND_DATA : CacheAction.STAT_ONLY);
+ try (CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(cacheSelector).build() )
{
final CountDownLatch latch = new CountDownLatch(2);
cache.getListenable().addListener(new CacheListener()
@@ -721,7 +721,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.create().forPath("/test");
final BlockingQueue<CacheEvent> events = new LinkedBlockingQueue<>();
- try ( CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build() )
+ try ( CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build() )
{
cache.getListenable().addListener
(
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCacheInCluster.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCacheInCluster.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCacheInCluster.java
index 3153764..87fea35 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCacheInCluster.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestSingleLevelCuratorCacheInCluster.java
@@ -50,7 +50,7 @@ public class TestSingleLevelCuratorCacheInCluster
client.start();
client.create().creatingParentsIfNeeded().forPath("/test");
- cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build();
+ cache = CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.singleLevel()).build();
final CountDownLatch resetLatch = new CountDownLatch(1);
final CountDownLatch reconnectLatch = new CountDownLatch(1);
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCache.java
index 0074c11..7491a89 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCache.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCache.java
@@ -39,27 +39,21 @@ public class TestTreeCache extends BaseTestTreeCache
client.create().forPath("/root/n1-c");
client.create().forPath("/root/n1-d");
- CacheFilter cacheFilter = new CacheFilter()
+ CacheSelector selector = new CacheSelector()
{
@Override
- public CacheAction actionForPath(String mainPath, String checkPath)
+ public boolean traverseChildren(String basePath, String fullPath)
{
- if ( checkPath.equals("/root/n1-c") )
- {
- return CacheAction.NOT_STORED;
- }
- return CacheAction.STAT_AND_DATA;
+ return !fullPath.equals("/root/n1-b/n2-b");
}
- };
- RefreshFilter refreshFilter = new RefreshFilter()
- {
+
@Override
- public boolean descend(String mainPath, String checkPath)
+ public CacheAction actionForPath(String basePath, String fullPath)
{
- return !checkPath.equals("/root/n1-b/n2-b");
+ return fullPath.equals("/root/n1-c") ? CacheAction.NOT_STORED : CacheAction.STAT_AND_DATA;
}
};
- cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/root").withCacheFilter(cacheFilter).withRefreshFilter(refreshFilter));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/root").withCacheSelector(selector));
cache.start();
assertEvent(CacheEvent.NODE_CREATED, "/root");
@@ -132,7 +126,7 @@ public class TestTreeCache extends BaseTestTreeCache
client.create().forPath("/test/3", "three".getBytes());
client.create().forPath("/test/2/sub", "two-sub".getBytes());
- cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test").withMaxDepth(0));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.maxDepth(0)));
cache.start();
assertEvent(CacheEvent.NODE_CREATED, "/test");
assertEvent(CacheEvent.CACHE_REFRESHED);
@@ -153,7 +147,7 @@ public class TestTreeCache extends BaseTestTreeCache
client.create().forPath("/test/3", "three".getBytes());
client.create().forPath("/test/2/sub", "two-sub".getBytes());
- cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test").withMaxDepth(1));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.maxDepth(1)));
cache.start();
assertEvent(CacheEvent.NODE_CREATED, "/test");
assertEvent(CacheEvent.NODE_CREATED, "/test/1", "one".getBytes());
@@ -181,7 +175,7 @@ public class TestTreeCache extends BaseTestTreeCache
client.create().forPath("/test/foo/bar/3", "three".getBytes());
client.create().forPath("/test/foo/bar/2/sub", "two-sub".getBytes());
- cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test/foo/bar").withMaxDepth(1));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test/foo/bar").withCacheSelector(CacheSelectors.maxDepth((1))));
cache.start();
assertEvent(CacheEvent.NODE_CREATED, "/test/foo/bar");
assertEvent(CacheEvent.NODE_CREATED, "/test/foo/bar/1", "one".getBytes());
@@ -231,7 +225,7 @@ public class TestTreeCache extends BaseTestTreeCache
client.create().forPath("/test");
client.create().forPath("/test/one", "hey there".getBytes());
- cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/").withMaxDepth(1));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/").withCacheSelector(CacheSelectors.maxDepth((1))));
cache.start();
assertEvent(CacheEvent.NODE_CREATED, "/");
assertEvent(CacheEvent.NODE_CREATED, "/test");
@@ -324,7 +318,7 @@ public class TestTreeCache extends BaseTestTreeCache
{
client.create().forPath("/test");
- cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test").withCacheFilter(CacheFilters.fullStatOnly()));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(client, "/test").withCacheSelector(CacheSelectors.statOnly()));
cache.start();
assertEvent(CacheEvent.NODE_CREATED, "/test");
assertEvent(CacheEvent.CACHE_REFRESHED);
http://git-wip-us.apache.org/repos/asf/curator/blob/2cf7c412/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCacheRandomTree.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCacheRandomTree.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCacheRandomTree.java
index a9f6007..9f6625c 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCacheRandomTree.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestTreeCacheRandomTree.java
@@ -1,3 +1,21 @@
+/**
+ * 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.watch;
import com.google.common.collect.Iterables;
@@ -47,60 +65,6 @@ public class TestTreeCacheRandomTree extends BaseTestTreeCache
doTestGiantRandomDeepTree();
}
- @Test
- public void temp() throws Exception
- {
- String[] strs = {"C/",
- "D/8050ee1037ad1124",
- "C/",
- "U/58add61a4aec0f95",
- "U/58add61a4aec0f95",
- "D/58add61a4aec0f95",
- "C/",
- "D/57d822ba79849e8b",
- "U/",
- "C/",
- "C/86db5aeba8b33121",
- "C/86db5aeba8b33121/fa45819eff99bf8f",
- "C/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581",
- "U/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50",
- "C/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50",
- "U/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50/be79114eaa5c78a0",
- "U/",
- "C/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50/be79114eaa5c78a0",
- "U/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50/be79114eaa5c78a0/f1672fc970f3d64e",
- "U/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50/be79114eaa5c78a0/f1672fc970f3d64e",
- "D/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50/be79114eaa5c78a0/f1672fc970f3d64e",
- "U/86db5aeba8b33121/fa45819eff99bf8f/1a6025e155810581/695419ff443d5e50/be79114eaa5c78a0"
- };
-
- client.create().forPath("/tree", null);
- CuratorFramework cl = client.usingNamespace("tree");
- for ( String s : strs )
- {
- String opcode = s.substring(0, 1);
- String path = s.substring(1);
- if ( opcode.equals("C") )
- {
- cl.create().forPath(path);
- }
- else if ( opcode.equals("U") )
- {
- byte[] newData = new byte[10];
- random.nextBytes(newData);
- cl.setData().forPath(path, newData);
- }
- else if ( opcode.equals("D") )
- {
- cl.delete().forPath(path);
- }
- else
- {
- throw new RuntimeException("bad");
- }
- }
- }
-
/**
* Randomly construct a large tree of test data in memory, mirror it into ZK, and then use
* a TreeCache to follow the changes. At each step, assert that TreeCache matches our
@@ -112,7 +76,7 @@ public class TestTreeCacheRandomTree extends BaseTestTreeCache
CuratorFramework cl = client.usingNamespace("tree");
if ( withDepth )
{
- cache = buildWithListeners(CuratorCacheBuilder.builder(cl, "/").withMaxDepth(TEST_DEPTH));
+ cache = buildWithListeners(CuratorCacheBuilder.builder(cl, "/").withCacheSelector(CacheSelectors.maxDepth(TEST_DEPTH)));
}
else
{
[2/2] curator git commit: Ported/finished NodeCache and tests
Posted by ra...@apache.org.
Ported/finished NodeCache and tests
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/72fe88c3
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/72fe88c3
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/72fe88c3
Branch: refs/heads/persistent-watch
Commit: 72fe88c3b99e43e905cce40c178a08cb2c409b78
Parents: 2cf7c41
Author: randgalt <ra...@apache.org>
Authored: Sun Jan 1 14:24:04 2017 -0500
Committer: randgalt <ra...@apache.org>
Committed: Sun Jan 1 14:24:04 2017 -0500
----------------------------------------------------------------------
.../recipes/watch/CompositeCuratorCache.java | 14 +
.../framework/recipes/watch/CuratorCache.java | 7 +
.../recipes/watch/CuratorCacheBase.java | 10 +-
.../recipes/watch/CuratorCacheBuilder.java | 21 +-
.../recipes/watch/InternalCuratorCache.java | 2 +-
.../recipes/watch/InternalNodeCache.java | 40 ++-
.../recipes/watch/PersistentWatcher.java | 2 +
.../framework/recipes/watch/TestNodeCache.java | 263 +++++++++++++++++++
8 files changed, 327 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CompositeCuratorCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CompositeCuratorCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CompositeCuratorCache.java
index cd1978a..d81c074 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CompositeCuratorCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CompositeCuratorCache.java
@@ -210,6 +210,20 @@ public class CompositeCuratorCache implements CuratorCache
}
@Override
+ public CachedNode getMain()
+ {
+ for ( CuratorCache cache : caches.values() )
+ {
+ CachedNode node = cache.getMain();
+ if ( node != null )
+ {
+ return node;
+ }
+ }
+ return null;
+ }
+
+ @Override
public CachedNode get(String path)
{
for ( CuratorCache cache : caches.values() )
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCache.java
index fa09376..bbc161e 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCache.java
@@ -109,6 +109,13 @@ public interface CuratorCache extends Closeable
Map<String, CachedNode> childrenAtPath(String path);
/**
+ * Return the node data stored for the main path (the path passed to the builder) in the cache or null
+ *
+ * @return node data or null
+ */
+ CachedNode getMain();
+
+ /**
* Return the node data stored for the path in the cache or null
*
* @param path node full path
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBase.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBase.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBase.java
index e97119c..2e8e5be 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBase.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBase.java
@@ -38,6 +38,7 @@ import java.util.concurrent.atomic.AtomicReference;
abstract class CuratorCacheBase implements CuratorCache
{
protected final Cache<String, CachedNode> cache;
+ private final String mainPath;
private final ListenerContainer<CacheListener> listeners = new ListenerContainer<>();
private final AtomicReference<State> state = new AtomicReference<>(State.LATENT);
private final AtomicReference<CountDownLatch> initialRefreshLatch = new AtomicReference<>(new CountDownLatch(1));
@@ -56,8 +57,9 @@ abstract class CuratorCacheBase implements CuratorCache
CLOSED
}
- protected CuratorCacheBase(Cache<String, CachedNode> cache, boolean sendRefreshEvents)
+ protected CuratorCacheBase(String mainPath, Cache<String, CachedNode> cache, boolean sendRefreshEvents)
{
+ this.mainPath = Objects.requireNonNull(mainPath, "mainPath cannot be null");
this.cache = Objects.requireNonNull(cache, "cache cannot be null");
this.sendRefreshEvents = sendRefreshEvents;
}
@@ -141,6 +143,12 @@ abstract class CuratorCacheBase implements CuratorCache
}
@Override
+ public CachedNode getMain()
+ {
+ return cache.asMap().get(mainPath);
+ }
+
+ @Override
public CachedNode get(String path)
{
return cache.asMap().get(path);
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
index 014584a..effa4e0 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CuratorCacheBuilder.java
@@ -18,6 +18,7 @@
*/
package org.apache.curator.framework.recipes.watch;
+import com.google.common.base.Preconditions;
import com.google.common.cache.CacheBuilder;
import org.apache.curator.framework.CuratorFramework;
import java.util.Objects;
@@ -28,7 +29,7 @@ public class CuratorCacheBuilder
private final CuratorFramework client;
private final String path;
private CacheBuilder<Object, Object> cacheBuilder = CacheBuilder.newBuilder();
- private boolean singleNode = false;
+ private CacheAction singleNodeCacheAction = null;
private boolean sendRefreshEvents = true;
private boolean refreshOnStart = true;
private boolean sortChildren = true;
@@ -42,9 +43,10 @@ public class CuratorCacheBuilder
public CuratorCache build()
{
- if ( singleNode )
+ if ( singleNodeCacheAction != null )
{
- return buildSingleNode();
+ Preconditions.checkState(cacheSelector == null, "Single node mode does not support CacheSelectors");
+ return new InternalNodeCache(client, path, singleNodeCacheAction, nodeComparator, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart);
}
return new InternalCuratorCache(client, path, cacheSelector, nodeComparator, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart, sortChildren);
@@ -52,7 +54,13 @@ public class CuratorCacheBuilder
public CuratorCacheBuilder forSingleNode()
{
- singleNode = true;
+ return forSingleNode(CacheAction.STAT_AND_DATA);
+ }
+
+ public CuratorCacheBuilder forSingleNode(CacheAction cacheAction)
+ {
+ cacheSelector = null;
+ singleNodeCacheAction = Objects.requireNonNull(cacheAction, "cacheAction cannot be null");
return this;
}
@@ -110,11 +118,6 @@ public class CuratorCacheBuilder
return this;
}
- private InternalNodeCache buildSingleNode()
- {
- return null;// TODO new InternalNodeCache(client, path, cacheSelector, nodeComparator, cacheFilter, cacheBuilder.<String, CachedNode>build(), sendRefreshEvents, refreshOnStart);
- }
-
private CuratorCacheBuilder(CuratorFramework client, String path)
{
this.client = Objects.requireNonNull(client, "client cannot be null");
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
index 101ca6d..d26d6df 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalCuratorCache.java
@@ -65,7 +65,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
InternalCuratorCache(CuratorFramework client, String path, final CacheSelector cacheSelector, CachedNodeComparator nodeComparator, Cache<String, CachedNode> cache, boolean sendRefreshEvents, final boolean refreshOnStart, boolean sortChildren)
{
- super(cache, sendRefreshEvents);
+ super(path, cache, sendRefreshEvents);
this.client = Objects.requireNonNull(client, "client cannot be null");
this.basePath = Objects.requireNonNull(path, "path cannot be null");
this.cacheSelector = Objects.requireNonNull(cacheSelector, "cacheSelector cannot be null");
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
index d3f327a..0dd6789 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/InternalNodeCache.java
@@ -37,16 +37,15 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Exchanger;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicReference;
class InternalNodeCache extends CuratorCacheBase
{
private final Logger log = LoggerFactory.getLogger(getClass());
private final WatcherRemoveCuratorFramework client;
private final String path;
- private final AtomicReference<CachedNode> data = new AtomicReference<>(null);
+ private final CacheAction cacheAction;
+ private final CachedNodeComparator nodeComparator;
private final AtomicBoolean isConnected = new AtomicBoolean(true);
- private final CacheSelector cacheSelector = CacheSelectors.statAndData();
private static final CachedNode nullNode = new CachedNode();
private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
{
@@ -93,11 +92,13 @@ class InternalNodeCache extends CuratorCacheBase
}
};
- InternalNodeCache(CuratorFramework client, String path, CachedNodeComparator nodeComparator, Cache<String, CachedNode> cache, boolean sendRefreshEvents, boolean refreshOnStart)
+ InternalNodeCache(CuratorFramework client, String path, CacheAction cacheAction, CachedNodeComparator nodeComparator, Cache<String, CachedNode> cache, boolean sendRefreshEvents, boolean refreshOnStart)
{
- super(cache, sendRefreshEvents);
+ super(path, cache, sendRefreshEvents);
this.client = client.newWatcherRemoveCuratorFramework();
this.path = PathUtils.validatePath(path);
+ this.cacheAction = cacheAction;
+ this.nodeComparator = nodeComparator;
}
@Override
@@ -117,32 +118,26 @@ class InternalNodeCache extends CuratorCacheBase
@Override
public CountDownLatch refreshAll()
{
- return null; // TODO
+ return refresh(path);
}
@Override
public CountDownLatch refresh(String path)
{
- return null; // TODO
- }
+ Preconditions.checkArgument(this.path.equals(path), "Bad path: " + path);
- public void refreXshAll()
- {
+ CountDownLatch latch = new CountDownLatch(1);
+ NotifyingRefresher refresher = new NotifyingRefresher(this, path, latch);
try
{
- reset(null);
+ reset(refresher);
}
catch ( Exception e )
{
ThreadUtils.checkInterrupted(e);
// TODO
}
- }
-
- public void reXfresh(String path)
- {
- Preconditions.checkArgument(this.path.equals(path), "Bad path: " + path);
- refreshAll();
+ return latch;
}
private void reset(Refresher refresher) throws Exception
@@ -177,7 +172,7 @@ class InternalNodeCache extends CuratorCacheBase
}
else if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
{
- switch ( cacheSelector.actionForPath(path, path) )
+ switch ( cacheAction )
{
default:
case NOT_STORED:
@@ -217,16 +212,19 @@ class InternalNodeCache extends CuratorCacheBase
volatile Exchanger<Object> rebuildTestExchanger;
private void setNewData(CachedNode newData) throws InterruptedException
{
- CachedNode previousData = data.getAndSet(newData);
+ CachedNode previousData = (newData != null) ? cache.asMap().put(path, newData) : cache.asMap().remove(path);
if ( newData == null )
{
- notifyListeners(CacheEvent.NODE_DELETED, path, newData);
+ if ( previousData != null )
+ {
+ notifyListeners(CacheEvent.NODE_DELETED, path, newData);
+ }
}
else if ( previousData == null )
{
notifyListeners(CacheEvent.NODE_CREATED, path, newData);
}
- else if ( !previousData.equals(newData) )
+ else if ( !nodeComparator.isSame(previousData, newData) )
{
notifyListeners(CacheEvent.NODE_CHANGED, path, newData);
}
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
index ff7dc82..b25983f 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/PersistentWatcher.java
@@ -28,6 +28,7 @@ import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.listen.ListenerContainer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.utils.ThreadUtils;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
@@ -129,6 +130,7 @@ public class PersistentWatcher implements Closeable
}
catch ( Exception e )
{
+ ThreadUtils.checkInterrupted(e);
// TODO
}
listeners.clear();
http://git-wip-us.apache.org/repos/asf/curator/blob/72fe88c3/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestNodeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestNodeCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestNodeCache.java
new file mode 100644
index 0000000..7874a72
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/watch/TestNodeCache.java
@@ -0,0 +1,263 @@
+/**
+ * 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.watch;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.UnhandledErrorListener;
+import org.apache.curator.framework.imps.TestCleanState;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
+import org.apache.curator.test.KillSession;
+import org.apache.curator.test.Timing;
+import org.apache.curator.utils.CloseableUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicReference;
+
+public class TestNodeCache extends BaseClassForTests
+{
+ private final Timing timing = new Timing();
+
+ @Test
+ public void testDeleteThenCreate() throws Exception
+ {
+ CuratorCache cache = null;
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+ client.start();
+ try
+ {
+ client.create().creatingParentsIfNeeded().forPath("/test/foo", "one".getBytes());
+
+ final AtomicReference<Throwable> error = new AtomicReference<Throwable>();
+ client.getUnhandledErrorListenable().addListener(new UnhandledErrorListener()
+ {
+ @Override
+ public void unhandledError(String message, Throwable e)
+ {
+ error.set(e);
+ }
+ });
+
+ final Semaphore semaphore = new Semaphore(0);
+ cache = CuratorCacheBuilder.builder(client, "/test/foo").forSingleNode().sendingRefreshEvents(false).build();
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ semaphore.release();
+ }
+ });
+ Assert.assertTrue(timing.awaitLatch(cache.start()));
+
+ Assert.assertEquals(cache.getMain().getData(), "one".getBytes());
+
+ client.delete().forPath("/test/foo");
+ Assert.assertTrue(semaphore.tryAcquire(1, 10, TimeUnit.SECONDS));
+ client.create().forPath("/test/foo", "two".getBytes());
+ Assert.assertTrue(semaphore.tryAcquire(1, 10, TimeUnit.SECONDS));
+
+ Throwable t = error.get();
+ if ( t != null )
+ {
+ Assert.fail("Assert", t);
+ }
+
+ Assert.assertEquals(cache.getMain().getData(), "two".getBytes());
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache);
+ TestCleanState.closeAndTestClean(client);
+ }
+ }
+
+ @Test
+ public void testRebuildAgainstOtherProcesses() throws Exception
+ {
+ CuratorCache cache = null;
+ final CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), new RetryOneTime(1));
+ client.start();
+ try
+ {
+ client.create().forPath("/test");
+ client.create().forPath("/test/snafu", "original".getBytes());
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ cache = CuratorCacheBuilder.builder(client, "/test/snafu").forSingleNode().sendingRefreshEvents(false).build();
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ latch.countDown();
+ }
+ });
+ ((InternalNodeCache)cache).rebuildTestExchanger = new Exchanger<Object>();
+
+ ExecutorService service = Executors.newSingleThreadExecutor();
+ final CuratorCache finalCache = cache;
+ Future<Object> future = service.submit(new Callable<Object>()
+ {
+ @Override
+ public Object call() throws Exception
+ {
+ ((InternalNodeCache)finalCache).rebuildTestExchanger.exchange(new Object(), 10, TimeUnit.SECONDS);
+
+ // simulate another process updating the node while we're rebuilding
+ client.setData().forPath("/test/snafu", "other".getBytes());
+
+ CachedNode currentData = finalCache.getMain();
+ Assert.assertNotNull(currentData);
+
+ ((InternalNodeCache)finalCache).rebuildTestExchanger.exchange(new Object(), 10, TimeUnit.SECONDS);
+
+ return null;
+ }
+ });
+ cache.start();
+ future.get();
+
+ Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
+ Assert.assertNotNull(cache.getMain());
+ Assert.assertEquals(cache.getMain().getData(), "other".getBytes());
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache);
+ TestCleanState.closeAndTestClean(client);
+ }
+ }
+
+ @Test
+ public void testKilledSession() throws Exception
+ {
+ CuratorCache cache = null;
+ CuratorFramework client = null;
+ try
+ {
+ client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ client.start();
+ client.create().creatingParentsIfNeeded().forPath("/test/node", "start".getBytes());
+
+ cache = CuratorCacheBuilder.builder(client, "/test/node").forSingleNode().sendingRefreshEvents(false).build();
+ Assert.assertTrue(timing.awaitLatch(cache.start()));
+
+ final CountDownLatch latch = new CountDownLatch(1);
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ latch.countDown();
+ }
+ });
+
+ KillSession.kill(client.getZookeeperClient().getZooKeeper());
+ Thread.sleep(timing.multiple(1.5).session());
+
+ Assert.assertEquals(cache.getMain().getData(), "start".getBytes());
+
+ client.setData().forPath("/test/node", "new data".getBytes());
+ Assert.assertTrue(timing.awaitLatch(latch));
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache);
+ TestCleanState.closeAndTestClean(client);
+ }
+ }
+
+ @Test
+ public void testBasics() throws Exception
+ {
+ CuratorCache cache = null;
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ client.start();
+ try
+ {
+ client.create().forPath("/test");
+
+ cache = CuratorCacheBuilder.builder(client, "/test/node").forSingleNode().sendingRefreshEvents(false).build();
+ Assert.assertTrue(timing.awaitLatch(cache.start()));
+
+ final Semaphore semaphore = new Semaphore(0);
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ semaphore.release();
+ }
+ });
+
+ Assert.assertNull(cache.getMain());
+
+ client.create().forPath("/test/node", "a".getBytes());
+ Assert.assertTrue(timing.acquireSemaphore(semaphore));
+ Assert.assertEquals(cache.getMain().getData(), "a".getBytes());
+
+ client.setData().forPath("/test/node", "b".getBytes());
+ Assert.assertTrue(timing.acquireSemaphore(semaphore));
+ Assert.assertEquals(cache.getMain().getData(), "b".getBytes());
+
+ client.delete().forPath("/test/node");
+ Assert.assertTrue(timing.acquireSemaphore(semaphore));
+ Assert.assertNull(cache.getMain());
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache);
+ TestCleanState.closeAndTestClean(client);
+ }
+ }
+
+ @Test
+ public void testRefreshEvent() throws Exception
+ {
+ CuratorCache cache = null;
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ client.start();
+ try
+ {
+ client.create().forPath("/test");
+
+ cache = CuratorCacheBuilder.builder(client, "/test/node").forSingleNode().build();
+ cache.start();
+
+ final BlockingQueue<CacheEvent> events = new LinkedBlockingQueue<>();
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ events.add(event);
+ }
+ });
+ Assert.assertEquals(events.poll(timing.milliseconds(), TimeUnit.MILLISECONDS), CacheEvent.CACHE_REFRESHED);
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache);
+ TestCleanState.closeAndTestClean(client);
+ }
+ }
+}