wip
diff --git a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/ForceRefreshOperation.java b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/ForceRefreshOperation.java
deleted file mode 100644
index e56dade..0000000
--- a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/ForceRefreshOperation.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Copyright 2012 Netflix, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.netflix.curator.framework.recipes.cache;
-
-class ForceRefreshOperation implements Operation
-{
- private final PathChildrenCache cache;
-
- ForceRefreshOperation(PathChildrenCache cache)
- {
- this.cache = cache;
- }
-
- @Override
- public void invoke() throws Exception
- {
- cache.refresh(true);
- }
-
- @Override
- public int hashCode()
- {
- return ForceRefreshOperation.class.hashCode();
- }
-
- @Override
- public boolean equals(Object obj)
- {
- //noinspection SimplifiableIfStatement
- if ( obj == null )
- {
- return false;
- }
-
- return (this == obj) || (getClass().equals(obj.getClass()));
- }
-
- @Override
- public String toString()
- {
- return "ForceRefreshOperation{}";
- }
-}
diff --git a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCache.java b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCache.java
index 357bcca..0710ab6 100644
--- a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCache.java
+++ b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCache.java
@@ -72,12 +72,14 @@
private final ListenerContainer<PathChildrenCacheListener> listeners = new ListenerContainer<PathChildrenCacheListener>();
private final ConcurrentMap<String, ChildData> currentData = Maps.newConcurrentMap();
+ private volatile Set<String> initialSet = null;
+
private final Watcher childrenWatcher = new Watcher()
{
@Override
public void process(WatchedEvent event)
{
- offerOperation(new RefreshOperation(PathChildrenCache.this));
+ offerOperation(new RefreshOperation(PathChildrenCache.this, RefreshMode.STANDARD));
}
};
@@ -189,7 +191,7 @@
*/
public void start() throws Exception
{
- start(false);
+ start(StartMode.STANDARD);
}
/**
@@ -198,32 +200,73 @@
* @param buildInitial if true, {@link #rebuild()} will be called before this method
* returns in order to get an initial view of the node
* @throws Exception errors
+ * @deprecated use {@link #start(StartMode)}
*/
public void start(boolean buildInitial) throws Exception
{
+ start(buildInitial ? StartMode.BUILD_INITIAL : StartMode.STANDARD);
+ }
+
+ /**
+ * How to load initial data when calling start()
+ */
+ public enum StartMode
+ {
+ /**
+ * Load data in the background - start() will return immediately
+ */
+ STANDARD,
+
+ /**
+ * Load initial data set in the foregroup. start() will block until all initial data is loaded
+ */
+ BUILD_INITIAL,
+
+ /**
+ * Load data in the background and post {@link PathChildrenCacheEvent.Type#INITIALIZED} when done
+ */
+ POST_INITIALIZED_EVENT
+ }
+
+ public void start(StartMode mode) throws Exception
+ {
+ mode = Preconditions.checkNotNull(mode, "mode cannot be null");
+
Preconditions.checkState(!executorService.isShutdown(), "already started");
client.getConnectionStateListenable().addListener(connectionStateListener);
executorService.submit
- (
- new Callable<Void>()
+ (
+ new Callable<Void>()
+ {
+ @Override
+ public Void call() throws Exception
{
- @Override
- public Void call() throws Exception
- {
- mainLoop();
- return null;
- }
+ mainLoop();
+ return null;
}
- );
+ }
+ );
- if ( buildInitial )
+ switch ( mode )
{
- rebuild();
- }
- else
- {
- offerOperation(new RefreshOperation(this));
+ case STANDARD:
+ {
+ offerOperation(new RefreshOperation(this, RefreshMode.STANDARD));
+ break;
+ }
+
+ case BUILD_INITIAL:
+ {
+ rebuild();
+ break;
+ }
+
+ case POST_INITIALIZED_EVENT:
+ {
+ offerOperation(new RefreshOperation(this, RefreshMode.POST_CHILDREN_INITIALIZED_EVENT));
+ break;
+ }
}
}
@@ -252,7 +295,7 @@
}
// this is necessary so that any updates that occurred while rebuilding are taken
- offerOperation(new ForceRefreshOperation(this));
+ offerOperation(new RefreshOperation(this, RefreshMode.FORCE_GET_DATA_AND_STAT));
}
/**
@@ -272,7 +315,7 @@
// this is necessary so that any updates that occurred while rebuilding are taken
// have to rebuild entire tree in case this node got deleted in the interim
- offerOperation(new ForceRefreshOperation(this));
+ offerOperation(new RefreshOperation(this, RefreshMode.FORCE_GET_DATA_AND_STAT));
}
/**
@@ -364,7 +407,7 @@
public void clearAndRefresh() throws Exception
{
currentData.clear();
- offerOperation(new RefreshOperation(this));
+ offerOperation(new RefreshOperation(this, RefreshMode.STANDARD));
}
/**
@@ -376,7 +419,14 @@
currentData.clear();
}
- void refresh(final boolean forceGetDataAndStat) throws Exception
+ enum RefreshMode
+ {
+ STANDARD,
+ FORCE_GET_DATA_AND_STAT,
+ POST_CHILDREN_INITIALIZED_EVENT
+ }
+
+ void refresh(final RefreshMode mode) throws Exception
{
ensurePath.ensure(client.getZookeeperClient());
@@ -385,7 +435,7 @@
@Override
public void processResult(CuratorFramework client, CuratorEvent event) throws Exception
{
- processChildren(event.getChildren(), forceGetDataAndStat);
+ processChildren(event.getChildren(), mode);
}
};
@@ -507,7 +557,7 @@
{
try
{
- offerOperation(new ForceRefreshOperation(this));
+ offerOperation(new RefreshOperation(this, RefreshMode.FORCE_GET_DATA_AND_STAT));
offerOperation(new EventOperation(this, new PathChildrenCacheEvent(PathChildrenCacheEvent.Type.CONNECTION_RECONNECTED, null)));
}
catch ( Exception e )
@@ -519,7 +569,7 @@
}
}
- private void processChildren(List<String> children, boolean forceGetDataAndStat) throws Exception
+ private void processChildren(List<String> children, RefreshMode mode) throws Exception
{
List<String> fullPaths = Lists.transform
(
@@ -541,13 +591,39 @@
remove(fullPath);
}
+ if ( mode == RefreshMode.POST_CHILDREN_INITIALIZED_EVENT )
+ {
+ initialSet = Sets.newHashSet(children);
+ updateInitialSet(null);
+ }
+
for ( String name : children )
{
String fullPath = ZKPaths.makePath(path, name);
- if ( forceGetDataAndStat || !currentData.containsKey(fullPath) )
+ if ( (mode == RefreshMode.FORCE_GET_DATA_AND_STAT) || !currentData.containsKey(fullPath) )
{
getDataAndStat(fullPath);
}
+ else
+ {
+ updateInitialSet(name);
+ }
+ }
+ }
+
+ private void updateInitialSet(String child)
+ {
+ if ( initialSet != null )
+ {
+ if ( child != null )
+ {
+ initialSet.remove(child);
+ }
+ if ( initialSet.size() == 0 )
+ {
+ offerOperation(new EventOperation(this, new PathChildrenCacheEvent(PathChildrenCacheEvent.Type.INITIALIZED, null)));
+ initialSet = null;
+ }
}
}
@@ -574,6 +650,7 @@
{
offerOperation(new EventOperation(this, new PathChildrenCacheEvent(PathChildrenCacheEvent.Type.CHILD_UPDATED, data)));
}
+ updateInitialSet(ZKPaths.getNodeFromPath(fullPath));
}
}
diff --git a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCacheEvent.java b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCacheEvent.java
index 9a37a6c..c601130 100644
--- a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCacheEvent.java
+++ b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/PathChildrenCacheEvent.java
@@ -46,19 +46,25 @@
CHILD_REMOVED,
/**
- * Called when the connection has changed to {@link ConnectionState#SUSPENDED}
+ * send when the connection has changed to {@link ConnectionState#SUSPENDED}
*/
CONNECTION_SUSPENDED,
/**
- * Called when the connection has changed to {@link ConnectionState#RECONNECTED}
+ * set when the connection has changed to {@link ConnectionState#RECONNECTED}
*/
CONNECTION_RECONNECTED,
/**
- * Called when the connection has changed to {@link ConnectionState#LOST}
+ * set when the connection has changed to {@link ConnectionState#LOST}
*/
- CONNECTION_LOST
+ CONNECTION_LOST,
+
+ /**
+ * Sent when start(StartMode.POST_CHILDREN_INITIALIZED_EVENT) has completed loading
+ * the initial data set
+ */
+ INITIALIZED
}
/**
diff --git a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/RefreshOperation.java b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/RefreshOperation.java
index a6988d6..be5bddc 100644
--- a/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/RefreshOperation.java
+++ b/curator-recipes/src/main/java/com/netflix/curator/framework/recipes/cache/RefreshOperation.java
@@ -19,16 +19,18 @@
class RefreshOperation implements Operation
{
private final PathChildrenCache cache;
+ private final PathChildrenCache.RefreshMode mode;
- RefreshOperation(PathChildrenCache cache)
+ RefreshOperation(PathChildrenCache cache, PathChildrenCache.RefreshMode mode)
{
this.cache = cache;
+ this.mode = mode;
}
@Override
public void invoke() throws Exception
{
- cache.refresh(false);
+ cache.refresh(mode);
}
@Override
@@ -52,6 +54,6 @@
@Override
public String toString()
{
- return "RefreshOperation{}";
+ return "RefreshOperation(" + mode + ")";
}
}
diff --git a/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/cache/TestPathChildrenCache.java b/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/cache/TestPathChildrenCache.java
index a40bc18..0ab2bcc 100644
--- a/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/cache/TestPathChildrenCache.java
+++ b/curator-recipes/src/test/java/com/netflix/curator/framework/recipes/cache/TestPathChildrenCache.java
@@ -23,9 +23,12 @@
import com.netflix.curator.framework.recipes.BaseClassForTests;
import com.netflix.curator.retry.RetryOneTime;
import com.netflix.curator.test.KillSession;
+import com.netflix.curator.test.TestingCluster;
import com.netflix.curator.test.Timing;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
import org.testng.Assert;
import org.testng.annotations.Test;
import java.util.List;
@@ -36,6 +39,60 @@
public class TestPathChildrenCache extends BaseClassForTests
{
@Test
+ public void testChildrenInitialized() throws Exception
+ {
+ Timing timing = new Timing();
+ PathChildrenCache cache = null;
+ CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ try
+ {
+ client.start();
+ client.create().forPath("/test");
+
+ cache = new PathChildrenCache(client, "/test", true);
+
+ final CountDownLatch addedLatch = new CountDownLatch(3);
+ final CountDownLatch initLatch = new CountDownLatch(1);
+ cache.getListenable().addListener
+ (
+ new PathChildrenCacheListener()
+ {
+ @Override
+ public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception
+ {
+ if ( event.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED )
+ {
+ addedLatch.countDown();
+ }
+ else if ( event.getType() == PathChildrenCacheEvent.Type.INITIALIZED )
+ {
+ initLatch.countDown();
+ }
+ }
+ }
+ );
+
+ client.create().forPath("/test/1", "1".getBytes());
+ client.create().forPath("/test/2", "2".getBytes());
+ client.create().forPath("/test/3", "3".getBytes());
+
+ cache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
+
+ Assert.assertTrue(timing.awaitLatch(addedLatch));
+ Assert.assertTrue(timing.awaitLatch(initLatch));
+ Assert.assertEquals(cache.getCurrentData().size(), 3);
+ Assert.assertEquals(cache.getCurrentData().get(0).getData(), "1".getBytes());
+ Assert.assertEquals(cache.getCurrentData().get(1).getData(), "2".getBytes());
+ Assert.assertEquals(cache.getCurrentData().get(2).getData(), "3".getBytes());
+ }
+ finally
+ {
+ Closeables.closeQuietly(cache);
+ Closeables.closeQuietly(client);
+ }
+ }
+
+ @Test
public void testUpdateWhenNotCachingData() throws Exception
{
Timing timing = new Timing();
@@ -66,7 +123,7 @@
}
}
);
- cache.start(true);
+ cache.start(PathChildrenCache.StartMode.BUILD_INITIAL);
client.create().forPath("/test/foo", "first".getBytes());
Assert.assertTrue(timing.awaitLatch(addedLatch));
@@ -161,7 +218,7 @@
}
}
);
- cache.start(true);
+ cache.start(PathChildrenCache.StartMode.BUILD_INITIAL);
client.delete().forPath("/test/foo");
Assert.assertTrue(removedLatch.await(10, TimeUnit.SECONDS));
@@ -260,7 +317,7 @@
}
}
);
- cache.start(true);
+ cache.start(PathChildrenCache.StartMode.BUILD_INITIAL);
future.get();
Assert.assertTrue(addedLatch.await(10, TimeUnit.SECONDS));
@@ -362,7 +419,7 @@
}
}
);
- cache.start(true);
+ cache.start(PathChildrenCache.StartMode.BUILD_INITIAL);
client.delete().forPath("/base/a");
Assert.assertTrue(semaphore.tryAcquire(1, 10, TimeUnit.SECONDS));
@@ -489,7 +546,7 @@
latch.countDown();
}
};
- cache.start(true);
+ cache.start(PathChildrenCache.StartMode.BUILD_INITIAL);
latch.await();
