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 2016/12/30 18:23:25 UTC
curator git commit: CacheListener needs to get the affected node.
Also, PATH_ONLY still needs to store the stat
Repository: curator
Updated Branches:
refs/heads/persistent-watch bf73f0d39 -> 076583d14
CacheListener needs to get the affected node. Also, PATH_ONLY still needs to store the stat
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/076583d1
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/076583d1
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/076583d1
Branch: refs/heads/persistent-watch
Commit: 076583d14506e3e761ca061cd51a358a97c08eb6
Parents: bf73f0d
Author: randgalt <ra...@apache.org>
Authored: Fri Dec 30 13:23:14 2016 -0500
Committer: randgalt <ra...@apache.org>
Committed: Fri Dec 30 13:23:14 2016 -0500
----------------------------------------------------------------------
.../framework/recipes/watch/CacheListener.java | 2 +-
.../recipes/watch/CuratorCacheBase.java | 4 +-
.../recipes/watch/CuratorCacheBuilder.java | 2 +-
.../recipes/watch/InternalCuratorCache.java | 33 ++---
.../recipes/watch/InternalNodeCache.java | 6 +-
.../framework/recipes/watch/Refresher.java | 2 +-
.../watch/TestSingleLevelCuratorCache.java | 124 ++++++++++---------
7 files changed, 83 insertions(+), 90 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheListener.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheListener.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheListener.java
index 06f3151..7a2a6a5 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheListener.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/CacheListener.java
@@ -20,5 +20,5 @@ package org.apache.curator.framework.recipes.watch;
public interface CacheListener
{
- void process(CacheEvent event, String path);
+ void process(CacheEvent event, String path, CachedNode affectedNode);
}
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/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 fcab38c..4803e69 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
@@ -193,7 +193,7 @@ abstract class CuratorCacheBase implements CuratorCache
}
}
- void notifyListeners(final CacheEvent eventType, final String path)
+ void notifyListeners(final CacheEvent eventType, final String path, final CachedNode affectedNode)
{
if ( state.get() != State.STARTED )
{
@@ -210,7 +210,7 @@ abstract class CuratorCacheBase implements CuratorCache
@Override
public Void apply(CacheListener listener)
{
- listener.process(eventType, path);
+ listener.process(eventType, path, affectedNode);
return null;
}
};
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/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 5bb73ba..e22aa1c 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
@@ -108,7 +108,7 @@ public class CuratorCacheBuilder
return this;
}
- public CuratorCacheBuilder withPrimingFilter(RefreshFilter refreshFilter)
+ public CuratorCacheBuilder withRefreshFilter(RefreshFilter refreshFilter)
{
this.refreshFilter = Objects.requireNonNull(refreshFilter, "primingFilter cannot be null");
return this;
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/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 ca83e9c..aa47363 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
@@ -98,9 +98,10 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
case NodeDeleted:
{
- if ( cache.asMap().remove(event.getPath()) != null )
+ CachedNode removed = cache.asMap().remove(event.getPath());
+ if ( removed != null )
{
- notifyListeners(CacheEvent.NODE_DELETED, event.getPath());
+ notifyListeners(CacheEvent.NODE_DELETED, event.getPath(), removed);
}
break;
}
@@ -154,15 +155,16 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
{
if ( event.getType() == CuratorEventType.GET_DATA )
{
+ CacheAction cacheAction = (CacheAction)event.getContext();
CachedNode newNode = new CachedNode(event.getStat(), event.getData());
- CachedNode oldNode = cache.asMap().put(path, newNode);
+ CachedNode oldNode = cache.asMap().put(path, (cacheAction == CacheAction.PATH_ONLY) ? new CachedNode(event.getStat()) : newNode);
if ( oldNode == null )
{
- notifyListeners(CacheEvent.NODE_CREATED, path);
+ notifyListeners(CacheEvent.NODE_CREATED, path, newNode);
}
else if ( !newNode.equals(oldNode) )
{
- notifyListeners(CacheEvent.NODE_CHANGED, path);
+ notifyListeners(CacheEvent.NODE_CHANGED, path, newNode);
}
}
else if ( event.getType() == CuratorEventType.CHILDREN )
@@ -185,7 +187,8 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
}
};
- switch ( cacheFilter.actionForPath(path) )
+ CacheAction cacheAction = cacheFilter.actionForPath(path);
+ switch ( cacheAction )
{
case NOT_STORED:
{
@@ -194,24 +197,12 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
}
case PATH_ONLY:
- {
- if ( cache.asMap().put(path, nullNode) == null )
- {
- notifyListeners(CacheEvent.NODE_CREATED, path);
- }
- else
- {
- notifyListeners(CacheEvent.NODE_CHANGED, path);
- }
- break;
- }
-
case PATH_AND_DATA:
{
try
{
refresher.increment();
- client.getData().inBackground(callback).forPath(path);
+ client.getData().inBackground(callback, cacheAction).forPath(path);
}
catch ( Exception e )
{
@@ -226,7 +217,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
try
{
refresher.increment();
- client.getData().decompressed().inBackground(callback).forPath(path);
+ client.getData().decompressed().inBackground(callback, cacheAction).forPath(path);
}
catch ( Exception e )
{
@@ -242,7 +233,7 @@ class InternalCuratorCache extends CuratorCacheBase implements Watcher
refresher.increment();
try
{
- client.getChildren().inBackground(callback).forPath(path);
+ client.getChildren().inBackground(callback, cacheAction).forPath(path);
}
catch ( Exception e )
{
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/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 3860679..b78480a 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
@@ -222,15 +222,15 @@ class InternalNodeCache extends CuratorCacheBase
CachedNode previousData = data.getAndSet(newData);
if ( newData == null )
{
- notifyListeners(CacheEvent.NODE_DELETED, path);
+ notifyListeners(CacheEvent.NODE_DELETED, path, newData);
}
else if ( previousData == null )
{
- notifyListeners(CacheEvent.NODE_CREATED, path);
+ notifyListeners(CacheEvent.NODE_CREATED, path, newData);
}
else if ( !previousData.equals(newData) )
{
- notifyListeners(CacheEvent.NODE_CHANGED, path);
+ notifyListeners(CacheEvent.NODE_CHANGED, path, newData);
}
if ( rebuildTestExchanger != null )
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/Refresher.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/Refresher.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/Refresher.java
index e81b475..fa2ae43 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/Refresher.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/watch/Refresher.java
@@ -50,7 +50,7 @@ class Refresher
{
if ( count.decrementAndGet() <= 0 )
{
- cacheBase.notifyListeners(CacheEvent.CACHE_REFRESHED, refreshPath);
+ cacheBase.notifyListeners(CacheEvent.CACHE_REFRESHED, refreshPath, cacheBase.get(refreshPath));
if ( latch != null )
{
latch.countDown();
http://git-wip-us.apache.org/repos/asf/curator/blob/076583d1/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 4bc423a..4ccf3b7 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
@@ -24,7 +24,6 @@ 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.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
@@ -67,7 +66,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
CacheListener listener = new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( (event == CacheEvent.NODE_CREATED) && path.equals("/baz"))
{
@@ -124,7 +123,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( event == CacheEvent.CACHE_REFRESHED )
{
@@ -159,7 +158,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
events.offer(event);
}
@@ -196,7 +195,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( event == CacheEvent.NODE_CREATED )
{
@@ -245,7 +244,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
Assert.assertNotEquals(event, CacheEvent.CACHE_REFRESHED);
if ( event == CacheEvent.NODE_CREATED )
@@ -290,7 +289,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( event == CacheEvent.NODE_CHANGED )
{
@@ -349,7 +348,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( event == CacheEvent.CACHE_REFRESHED )
{
@@ -366,7 +365,8 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
{
try
{
- Assert.assertEquals(cache.get(path).getData(), "two".getBytes());
+ Assert.assertNotNull(affectedNode);
+ Assert.assertEquals(affectedNode.getData(), "two".getBytes());
}
finally
{
@@ -416,7 +416,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( event == CacheEvent.NODE_CREATED )
{
@@ -509,7 +509,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
events.add(event);
semaphore.release();
@@ -564,7 +564,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
cache.getListenable().addListener(new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( cache.refreshCount() > 0 )
{
@@ -591,10 +591,10 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
}
}
- //@Test
+ @Test
public void testKilledSession() throws Exception
{
- PathChildrenCache cache = null;
+ CuratorCache cache = null;
CuratorFramework client = null;
try
{
@@ -602,39 +602,43 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
client.start();
client.create().forPath("/test");
- cache = new PathChildrenCache(client, "/test", true);
+ cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().build();
cache.start();
final CountDownLatch childAddedLatch = new CountDownLatch(1);
final CountDownLatch lostLatch = new CountDownLatch(1);
final CountDownLatch reconnectedLatch = new CountDownLatch(1);
final CountDownLatch removedLatch = new CountDownLatch(1);
- cache.getListenable().addListener
- (
- new PathChildrenCacheListener()
+ client.getConnectionStateListenable().addListener(new ConnectionStateListener()
+ {
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState)
+ {
+ if ( newState == ConnectionState.LOST )
{
- @Override
- public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
- {
- if ( event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED )
- {
- childAddedLatch.countDown();
- }
- else if ( event.getType() == PathChildrenCacheEvent.Type.CONNECTION_LOST )
- {
- lostLatch.countDown();
- }
- else if ( event.getType() == PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED )
- {
- reconnectedLatch.countDown();
- }
- else if ( event.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED )
- {
- removedLatch.countDown();
- }
- }
+ lostLatch.countDown();
}
- );
+ else if ( newState.isConnected() )
+ {
+ reconnectedLatch.countDown();
+ }
+ }
+ });
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ if ( event == CacheEvent.NODE_CREATED )
+ {
+ childAddedLatch.countDown();
+ }
+ else if ( event == CacheEvent.NODE_DELETED )
+ {
+ removedLatch.countDown();
+ }
+ }
+ });
client.create().withMode(CreateMode.EPHEMERAL).forPath("/test/me", "data".getBytes());
Assert.assertTrue(timing.awaitLatch(childAddedLatch));
@@ -651,7 +655,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
}
}
- //@Test
+ @Test
public void testModes() throws Exception
{
CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
@@ -660,9 +664,9 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
{
client.create().forPath("/test");
- for ( boolean cacheData : new boolean[]{false, true} )
+ for ( int i = 0; i < 2; ++i )
{
- internalTestMode(client, cacheData);
+ internalTestMode(client, i > 0);
client.delete().forPath("/test/one");
client.delete().forPath("/test/two");
@@ -720,40 +724,38 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
private void internalTestMode(CuratorFramework client, boolean cacheData) throws Exception
{
- try ( PathChildrenCache cache = new PathChildrenCache(client, "/test", cacheData) )
+ CacheFilter cacheFilter = new SingleLevelCacheFilter("/test", cacheData ? CacheAction.PATH_AND_DATA : CacheAction.PATH_ONLY);
+ try (CuratorCache cache = CuratorCacheBuilder.builder(client, "/test").forSingleLevel().withCacheFilter(cacheFilter).build() )
{
final CountDownLatch latch = new CountDownLatch(2);
- cache.getListenable().addListener
- (
- new PathChildrenCacheListener()
+ cache.getListenable().addListener(new CacheListener()
+ {
+ @Override
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
+ {
+ if ( event == CacheEvent.NODE_CREATED )
{
- @Override
- public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
- {
- if ( event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED )
- {
- latch.countDown();
- }
- }
+ latch.countDown();
}
- );
+ }
+ });
cache.start();
client.create().forPath("/test/one", "one".getBytes());
client.create().forPath("/test/two", "two".getBytes());
Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
- for ( ChildData data : cache.getCurrentData() )
+ for ( CachedNode data : cache.getAll() )
{
if ( cacheData )
{
- Assert.assertNotNull(data.getData());
- Assert.assertNotNull(data.getStat());
+ Assert.assertTrue(data.getData().length > 0);
+ Assert.assertTrue(data.getStat().getDataLength() > 0);
}
else
{
- Assert.assertNull(data.getData());
- Assert.assertNotNull(data.getStat());
+ Assert.assertTrue(data.getData().length == 0);
+ Assert.assertTrue(data.getStat().getDataLength() > 0);
}
}
}
@@ -776,7 +778,7 @@ public class TestSingleLevelCuratorCache extends BaseClassForTests
new CacheListener()
{
@Override
- public void process(CacheEvent event, String path)
+ public void process(CacheEvent event, String path, CachedNode affectedNode)
{
if ( path.equals("/test/one") )
{