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 2014/07/31 17:03:22 UTC
[1/5] git commit: CURATOR-33 recursive TreeCache recipe
Repository: curator
Updated Branches:
refs/heads/CURATOR-33 [created] fd5c8aa1d
CURATOR-33 recursive TreeCache recipe
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/547a7e23
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/547a7e23
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/547a7e23
Branch: refs/heads/CURATOR-33
Commit: 547a7e23eb48fece49420bbcfdc84e763bf74461
Parents: c358bbc
Author: Scott Blum <sc...@squareup.com>
Authored: Mon Jul 14 20:38:07 2014 -0400
Committer: Scott Blum <sc...@squareup.com>
Committed: Fri Jul 18 03:01:33 2014 -0400
----------------------------------------------------------------------
.../java/org/apache/curator/utils/ZKPaths.java | 29 +
.../org/apache/curator/utils/TestZKPaths.java | 38 ++
.../framework/recipes/cache/TreeCache.java | 600 +++++++++++++++++++
.../framework/recipes/cache/TreeCacheEvent.java | 126 ++++
.../recipes/cache/TreeCacheListener.java | 37 ++
.../framework/recipes/cache/TestTreeCache.java | 421 +++++++++++++
6 files changed, 1251 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/547a7e23/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java b/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
index 04b2141..1ffd71b 100644
--- a/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
+++ b/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
@@ -24,6 +24,7 @@ import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -112,6 +113,34 @@ public class ZKPaths
}
/**
+ * Given a full path, return the the individual parts, without slashes.
+ * The root path will return an empty list.
+ *
+ * @param path the path
+ * @return an array of parts
+ */
+ public static List<String> split(String path)
+ {
+ PathUtils.validatePath(path);
+ if (path.length() == 1) {
+ return Collections.emptyList();
+ }
+ int lastPos = 1;
+ List<String> parts = new ArrayList<String>();
+ while (true)
+ {
+ int nextPos = path.indexOf('/', lastPos);
+ if (nextPos < 0) {
+ parts.add(path.substring(lastPos));
+ break;
+ }
+ parts.add(path.substring(lastPos, nextPos));
+ lastPos = nextPos + 1;
+ }
+ return parts;
+ }
+
+ /**
* Make sure all the nodes in the path are created. NOTE: Unlike File.mkdirs(), Zookeeper doesn't distinguish
* between directories and files. So, every node in the path is created. The data for each node is an empty blob
*
http://git-wip-us.apache.org/repos/asf/curator/blob/547a7e23/curator-client/src/test/java/org/apache/curator/utils/TestZKPaths.java
----------------------------------------------------------------------
diff --git a/curator-client/src/test/java/org/apache/curator/utils/TestZKPaths.java b/curator-client/src/test/java/org/apache/curator/utils/TestZKPaths.java
new file mode 100644
index 0000000..fd03929
--- /dev/null
+++ b/curator-client/src/test/java/org/apache/curator/utils/TestZKPaths.java
@@ -0,0 +1,38 @@
+/**
+ * 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.utils;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+public class TestZKPaths
+{
+
+ @Test
+ public void testSplit()
+ {
+ Assert.assertEquals(ZKPaths.split("/"), new ArrayList<String>());
+ Assert.assertEquals(ZKPaths.split("/test"), Arrays.asList("test"));
+ Assert.assertEquals(ZKPaths.split("/test/one"), Arrays.asList("test", "one"));
+ Assert.assertEquals(ZKPaths.split("/test/one/two"), Arrays.asList("test", "one", "two"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/547a7e23/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
new file mode 100644
index 0000000..31bff1e
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
@@ -0,0 +1,600 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.curator.framework.recipes.cache;
+
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableSortedSet;
+import com.google.common.collect.Maps;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.api.BackgroundCallback;
+import org.apache.curator.framework.api.CuratorEvent;
+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.CloseableExecutorService;
+import org.apache.curator.utils.ThreadUtils;
+import org.apache.curator.utils.ZKPaths;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.SortedSet;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * <p>A utility that attempts to keep all data from all children of a ZK path locally cached. This class
+ * will watch the ZK path, respond to update/create/delete events, pull down the data, etc. You can
+ * register a listener that will get notified when changes occur.</p>
+ * <p></p>
+ * <p><b>IMPORTANT</b> - it's not possible to stay transactionally in sync. Users of this class must
+ * be prepared for false-positives and false-negatives. Additionally, always use the version number
+ * when updating data to avoid overwriting another process' change.</p>
+ */
+public class TreeCache implements Closeable
+{
+ private static final Logger LOG = LoggerFactory.getLogger(TreeCache.class);
+
+ private enum NodeState
+ {
+ PENDING, LIVE, DEAD
+ }
+
+ final class TreeNode implements Watcher, BackgroundCallback
+ {
+ private final AtomicReference<NodeState> nodeState = new AtomicReference<NodeState>(NodeState.PENDING);
+ private final String path;
+ private final TreeNode parent;
+ private final AtomicReference<Stat> stat = new AtomicReference<Stat>();
+ private final AtomicReference<byte[]> data = new AtomicReference<byte[]>();
+ private final AtomicReference<ConcurrentMap<String, TreeNode>> children = new AtomicReference<ConcurrentMap<String, TreeNode>>();
+
+ TreeNode(String path, TreeNode parent)
+ {
+ this.path = path;
+ this.parent = parent;
+ }
+
+ private void refreshChildren() throws Exception
+ {
+ outstandingOps.incrementAndGet();
+ client.getChildren().usingWatcher(this).inBackground(this).forPath(path);
+ }
+
+ private void refreshData() throws Exception
+ {
+ outstandingOps.incrementAndGet();
+ if ( dataIsCompressed )
+ {
+ client.getData().decompressed().usingWatcher(this).inBackground(this).forPath(path);
+ }
+ else
+ {
+ client.getData().usingWatcher(this).inBackground(this).forPath(path);
+ }
+ }
+
+ private void wasReconnected() throws Exception
+ {
+ refreshData();
+ refreshChildren();
+ ConcurrentMap<String, TreeNode> childMap = children.get();
+ if ( childMap != null )
+ {
+ for ( TreeNode child : childMap.values() )
+ {
+ child.wasReconnected();
+ }
+ }
+ }
+
+ private void wasCreated() throws Exception
+ {
+ refreshData();
+ refreshChildren();
+ }
+
+ private void wasDeleted() throws Exception
+ {
+ stat.set(null);
+ data.set(null);
+ client.clearWatcherReferences(this);
+ ConcurrentMap<String, TreeNode> childMap = children.getAndSet(null);
+ if ( childMap != null )
+ {
+ ArrayList<TreeNode> childCopy = new ArrayList<TreeNode>(childMap.values());
+ childMap.clear();
+ for ( TreeNode child : childCopy )
+ {
+ child.wasDeleted();
+ }
+ }
+
+ if ( treeState.get() == TreeState.CLOSED )
+ {
+ return;
+ }
+
+ if ( nodeState.compareAndSet(NodeState.LIVE, NodeState.DEAD) )
+ {
+ publishEvent(TreeCacheEvent.Type.NODE_REMOVED, path);
+ }
+
+ if ( parent == null )
+ {
+ // Root node; use an exist query to watch for existence.
+ client.checkExists().usingWatcher(this).inBackground().forPath(path);
+ }
+ else
+ {
+ // Remove from parent if we're currently a child
+ ConcurrentMap<String, TreeNode> parentChildMap = parent.children.get();
+ if ( parentChildMap != null )
+ {
+ parentChildMap.remove(ZKPaths.getNodeFromPath(path), this);
+ }
+ }
+ }
+
+ @Override
+ public void process(WatchedEvent event)
+ {
+ System.out.println(event);
+ try
+ {
+ switch ( event.getType() )
+ {
+ case NodeCreated:
+ assert parent == null;
+ wasCreated();
+ break;
+ case NodeChildrenChanged:
+ refreshChildren();
+ break;
+ case NodeDataChanged:
+ refreshData();
+ break;
+ case NodeDeleted:
+ wasDeleted();
+ break;
+ }
+ }
+ catch ( Exception e )
+ {
+ handleException(e);
+ }
+ }
+
+ @Override
+ public void processResult(CuratorFramework client, CuratorEvent event) throws Exception
+ {
+ System.out.println(event);
+ switch ( event.getType() )
+ {
+ case EXISTS:
+ // TODO: should only happen for root node
+ if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
+ {
+ nodeState.compareAndSet(NodeState.DEAD, NodeState.PENDING);
+ wasCreated();
+ }
+ else if ( event.getResultCode() == KeeperException.Code.NONODE.intValue() )
+ {
+ wasDeleted();
+ }
+ break;
+ case CHILDREN:
+ if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
+ {
+ stat.set(event.getStat());
+
+ if ( event.getChildren().isEmpty() )
+ {
+ break;
+ }
+
+ ConcurrentMap<String, TreeNode> childMap = children.get();
+ if ( childMap == null )
+ {
+ childMap = Maps.newConcurrentMap();
+ if ( !children.compareAndSet(null, childMap) )
+ {
+ childMap = children.get();
+ }
+ }
+
+ for ( String child : event.getChildren() )
+ {
+ String fullPath = ZKPaths.makePath(path, child);
+ if ( !childMap.containsKey(child) )
+ {
+ TreeNode node = new TreeNode(fullPath, this);
+ if ( childMap.putIfAbsent(child, node) == null )
+ {
+ node.wasCreated();
+ }
+ }
+ }
+ }
+ else if ( event.getResultCode() == KeeperException.Code.NONODE.intValue() )
+ {
+ wasDeleted();
+ }
+ break;
+ case GET_DATA:
+ if ( event.getResultCode() == KeeperException.Code.OK.intValue() )
+ {
+ Stat oldStat = stat.getAndSet(event.getStat());
+ if ( cacheData )
+ {
+ data.set(event.getData());
+ }
+
+ if ( nodeState.compareAndSet(NodeState.PENDING, NodeState.LIVE) )
+ {
+ publishEvent(TreeCacheEvent.Type.NODE_ADDED, new ChildData(event.getPath(), event.getStat(), event.getData()));
+ }
+ else if ( oldStat.getMzxid() != event.getStat().getMzxid() )
+ {
+ publishEvent(TreeCacheEvent.Type.NODE_UPDATED, new ChildData(event.getPath(), event.getStat(), event.getData()));
+ }
+ }
+ else if ( event.getResultCode() == KeeperException.Code.NONODE.intValue() )
+ {
+ wasDeleted();
+ }
+ break;
+ default:
+ handleException(new Exception(String.format("Unknown event %s", event)));
+ }
+
+ if ( outstandingOps.decrementAndGet() == 0 )
+ {
+ if ( treeState.compareAndSet(TreeState.LATENT, TreeState.STARTED) )
+ {
+ publishEvent(TreeCacheEvent.Type.INITIALIZED);
+ }
+ }
+ }
+ }
+
+ private enum TreeState
+ {
+ LATENT,
+ STARTED,
+ CLOSED
+ }
+
+ /**
+ * Detemines when to publish the initialized event.
+ */
+ private final AtomicLong outstandingOps = new AtomicLong(0);
+
+ private final TreeNode root;
+ private final CuratorFramework client;
+ private final CloseableExecutorService executorService;
+ private final boolean cacheData;
+ private final boolean dataIsCompressed;
+ private final ListenerContainer<TreeCacheListener> listeners = new ListenerContainer<TreeCacheListener>();
+ private final AtomicReference<TreeState> treeState = new AtomicReference<TreeState>(TreeState.LATENT);
+
+ private final ConnectionStateListener connectionStateListener = new ConnectionStateListener()
+ {
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState)
+ {
+ handleStateChange(newState);
+ }
+ };
+
+ private static final ThreadFactory defaultThreadFactory = ThreadUtils.newThreadFactory("TreeCache");
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param cacheData if true, node contents are cached in addition to the stat
+ */
+ public TreeCache(CuratorFramework client, String path, boolean cacheData)
+ {
+ this(client, path, cacheData, false, new CloseableExecutorService(Executors.newSingleThreadExecutor(defaultThreadFactory), true));
+ }
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param cacheData if true, node contents are cached in addition to the stat
+ * @param threadFactory factory to use when creating internal threads
+ */
+ public TreeCache(CuratorFramework client, String path, boolean cacheData, ThreadFactory threadFactory)
+ {
+ this(client, path, cacheData, false, new CloseableExecutorService(Executors.newSingleThreadExecutor(threadFactory), true));
+ }
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param cacheData if true, node contents are cached in addition to the stat
+ * @param dataIsCompressed if true, data in the path is compressed
+ * @param threadFactory factory to use when creating internal threads
+ */
+ public TreeCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, ThreadFactory threadFactory)
+ {
+ this(client, path, cacheData, dataIsCompressed, new CloseableExecutorService(Executors.newSingleThreadExecutor(threadFactory), true));
+ }
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param cacheData if true, node contents are cached in addition to the stat
+ * @param dataIsCompressed if true, data in the path is compressed
+ * @param executorService ExecutorService to use for the TreeCache's background thread
+ */
+ public TreeCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, final ExecutorService executorService)
+ {
+ this(client, path, cacheData, dataIsCompressed, new CloseableExecutorService(executorService));
+ }
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param cacheData if true, node contents are cached in addition to the stat
+ * @param dataIsCompressed if true, data in the path is compressed
+ * @param executorService Closeable ExecutorService to use for the TreeCache's background thread
+ */
+ public TreeCache(CuratorFramework client, String path, boolean cacheData, boolean dataIsCompressed, final CloseableExecutorService executorService)
+ {
+ this.root = new TreeNode(path, null);
+ this.client = client;
+ this.cacheData = cacheData;
+ this.dataIsCompressed = dataIsCompressed;
+ this.executorService = executorService;
+ }
+
+ /**
+ * Start the cache. The cache is not started automatically. You must call this method.
+ *
+ * @throws Exception errors
+ */
+ public void start() throws Exception
+ {
+ client.getConnectionStateListenable().addListener(connectionStateListener);
+ root.wasCreated();
+ }
+
+ /**
+ * Close/end the cache
+ *
+ * @throws java.io.IOException errors
+ */
+ @Override
+ public void close() throws IOException
+ {
+ if ( treeState.compareAndSet(TreeState.STARTED, TreeState.CLOSED) )
+ {
+ client.getConnectionStateListenable().removeListener(connectionStateListener);
+ listeners.clear();
+ executorService.close();
+ try
+ {
+ root.wasDeleted();
+ }
+ catch ( Exception e )
+ {
+ handleException(e);
+ }
+ }
+ }
+
+ /**
+ * Return the cache listenable
+ *
+ * @return listenable
+ */
+ public ListenerContainer<TreeCacheListener> getListenable()
+ {
+ return listeners;
+ }
+
+ private TreeNode find(String fullPath)
+ {
+ if ( !fullPath.startsWith(root.path) )
+ {
+ return null;
+ }
+
+ TreeNode current = root;
+ if ( fullPath.length() > root.path.length() )
+ {
+ List<String> split = ZKPaths.split(fullPath.substring(root.path.length()));
+ for ( String part : split )
+ {
+ ConcurrentMap<String, TreeNode> map = current.children.get();
+ if ( map == null )
+ {
+ return null;
+ }
+ current = map.get(part);
+ if ( current == null )
+ {
+ return null;
+ }
+ }
+ }
+ return current;
+ }
+
+ /**
+ * Return the current set of children. There are no guarantees of accuracy. This is
+ * merely the most recent view of the data. The data is returned in sorted order. If there is
+ * no child with that path, <code>null</code> is returned.
+ *
+ * @param fullPath full path to the node to check
+ * @return a possibly-empty list of children if the node is alive, or null
+ */
+ public SortedSet<String> getCurrentChildren(String fullPath)
+ {
+ TreeNode node = find(fullPath);
+ if ( node == null || node.nodeState.get() != NodeState.LIVE )
+ {
+ return null;
+ }
+ ConcurrentMap<String, TreeNode> map = node.children.get();
+ SortedSet<String> result;
+ if ( map == null )
+ {
+ result = Collections.emptySortedSet();
+ }
+ else
+ {
+ result = ImmutableSortedSet.copyOf(map.keySet());
+ }
+
+ // Double-check liveness after retreiving children.
+ return node.nodeState.get() == NodeState.LIVE ? result : null;
+ }
+
+ /**
+ * Return the current data for the given path. There are no guarantees of accuracy. This is
+ * merely the most recent view of the data. If there is no child with that path,
+ * <code>null</code> is returned.
+ *
+ * @param fullPath full path to the node to check
+ * @return data if the node is alive, or null
+ */
+ public ChildData getCurrentData(String fullPath)
+ {
+ TreeNode node = find(fullPath);
+ if ( node == null || node.nodeState.get() != NodeState.LIVE )
+ {
+ return null;
+ }
+ ChildData result = new ChildData(node.path, node.stat.get(), node.data.get());
+ // Double-check liveness after retreiving stat / data.
+ return node.nodeState.get() == NodeState.LIVE ? result : null;
+ }
+
+ void callListeners(final TreeCacheEvent event)
+ {
+ listeners.forEach(new Function<TreeCacheListener, Void>()
+ {
+ @Override
+ public Void apply(TreeCacheListener listener)
+ {
+ try
+ {
+ listener.childEvent(client, event);
+ }
+ catch ( Exception e )
+ {
+ handleException(e);
+ }
+ return null;
+ }
+ }
+ );
+ }
+
+ /**
+ * Default behavior is just to log the exception
+ *
+ * @param e the exception
+ */
+ protected void handleException(Throwable e)
+ {
+ LOG.error("", e);
+ }
+
+ private void handleStateChange(ConnectionState newState)
+ {
+ switch ( newState )
+ {
+ case SUSPENDED:
+ publishEvent(TreeCacheEvent.Type.CONNECTION_SUSPENDED);
+ break;
+
+ case LOST:
+ publishEvent(TreeCacheEvent.Type.CONNECTION_LOST);
+ break;
+
+ case RECONNECTED:
+ try
+ {
+ root.wasReconnected();
+ publishEvent(TreeCacheEvent.Type.CONNECTION_RECONNECTED);
+ }
+ catch ( Exception e )
+ {
+ handleException(e);
+ }
+ break;
+ }
+ }
+
+ private void publishEvent(TreeCacheEvent.Type type)
+ {
+ publishEvent(new TreeCacheEvent(type, null));
+ }
+
+ private void publishEvent(TreeCacheEvent.Type type, String path)
+ {
+ publishEvent(new TreeCacheEvent(type, new ChildData(path, null, null)));
+ }
+
+ private void publishEvent(TreeCacheEvent.Type type, ChildData data)
+ {
+ publishEvent(new TreeCacheEvent(type, data));
+ }
+
+ private void publishEvent(final TreeCacheEvent event)
+ {
+ System.out.println(event);
+ if ( treeState.get() != TreeState.CLOSED )
+ {
+ executorService.submit(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ {
+ try
+ {
+ callListeners(event);
+ }
+ catch ( Exception e )
+ {
+ handleException(e);
+ }
+ }
+ }
+ });
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/547a7e23/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheEvent.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheEvent.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheEvent.java
new file mode 100644
index 0000000..2080d26
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheEvent.java
@@ -0,0 +1,126 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.curator.framework.recipes.cache;
+
+/**
+ * POJO that abstracts a change to a path
+ */
+public class TreeCacheEvent
+{
+ private final Type type;
+ private final ChildData data;
+
+ /**
+ * Type of change
+ */
+ public enum Type
+ {
+ /**
+ * A node was added.
+ */
+ NODE_ADDED,
+
+ /**
+ * A node's data was changed
+ */
+ NODE_UPDATED,
+
+ /**
+ * A node was removed from the tree
+ */
+ NODE_REMOVED,
+
+ /**
+ * Called when the connection has changed to {@link org.apache.curator.framework.state.ConnectionState#SUSPENDED}
+ * <p/>
+ * This is exposed so that users of the class can be notified of issues that *might* affect normal operation.
+ * The TreeCache is written such that listeners are not expected to do anything special on this
+ * event, except for those people who want to cause some application-specific logic to fire when this occurs.
+ * While the connection is down, the TreeCache will continue to have its state from before it lost
+ * the connection and after the connection is restored, the TreeCache will emit normal child events
+ * for all of the adds, deletes and updates that happened during the time that it was disconnected.
+ */
+ CONNECTION_SUSPENDED,
+
+ /**
+ * Called when the connection has changed to {@link org.apache.curator.framework.state.ConnectionState#RECONNECTED}
+ * <p/>
+ * This is exposed so that users of the class can be notified of issues that *might* affect normal operation.
+ * The TreeCache is written such that listeners are not expected to do anything special on this
+ * event, except for those people who want to cause some application-specific logic to fire when this occurs.
+ * While the connection is down, the TreeCache will continue to have its state from before it lost
+ * the connection and after the connection is restored, the TreeCache will emit normal child events
+ * for all of the adds, deletes and updates that happened during the time that it was disconnected.
+ */
+ CONNECTION_RECONNECTED,
+
+ /**
+ * Called when the connection has changed to {@link org.apache.curator.framework.state.ConnectionState#LOST}
+ * <p/>
+ * This is exposed so that users of the class can be notified of issues that *might* affect normal operation.
+ * The TreeCache is written such that listeners are not expected to do anything special on this
+ * event, except for those people who want to cause some application-specific logic to fire when this occurs.
+ * While the connection is down, the TreeCache will continue to have its state from before it lost
+ * the connection and after the connection is restored, the TreeCache will emit normal child events
+ * for all of the adds, deletes and updates that happened during the time that it was disconnected.
+ */
+ CONNECTION_LOST,
+
+ /**
+ * Posted when the initial cache has been populated.
+ */
+ INITIALIZED
+ }
+
+ /**
+ * @param type event type
+ * @param data event data or null
+ */
+ public TreeCacheEvent(Type type, ChildData data)
+ {
+ this.type = type;
+ this.data = data;
+ }
+
+ /**
+ * @return change type
+ */
+ public Type getType()
+ {
+ return type;
+ }
+
+ /**
+ * @return the node's data
+ */
+ public ChildData getData()
+ {
+ return data;
+ }
+
+ @Override
+ public String toString()
+ {
+ return TreeCacheEvent.class.getSimpleName() + "{" +
+ "type=" + type +
+ ", data=" + data +
+ '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/547a7e23/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheListener.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheListener.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheListener.java
new file mode 100644
index 0000000..6ea342d
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCacheListener.java
@@ -0,0 +1,37 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.curator.framework.recipes.cache;
+
+import org.apache.curator.framework.CuratorFramework;
+
+/**
+ * Listener for {@link TreeCache} changes
+ */
+public interface TreeCacheListener
+{
+ /**
+ * Called when a change has occurred
+ *
+ * @param client the client
+ * @param event describes the change
+ * @throws Exception errors
+ */
+ public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/curator/blob/547a7e23/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
new file mode 100644
index 0000000..b8a009b
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
@@ -0,0 +1,421 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.curator.framework.recipes.cache;
+
+import com.google.common.collect.ImmutableSortedSet;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.UnhandledErrorListener;
+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.apache.zookeeper.CreateMode;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+
+public class TestTreeCache extends BaseClassForTests
+{
+ private final Timing timing = new Timing();
+ private CuratorFramework client;
+ private TreeCache cache;
+ private List<Throwable> exceptions;
+ private BlockingQueue<TreeCacheEvent> events;
+ private TreeCacheListener eventListener;
+
+ /**
+ * A TreeCache that records exceptions.
+ */
+ private TreeCache newTreeCache(String path, boolean cacheData)
+ {
+ return new TreeCache(client, path, cacheData)
+ {
+ @Override
+ protected void handleException(Throwable e)
+ {
+ exceptions.add(e);
+ }
+ };
+ }
+
+ @Override
+ @BeforeMethod
+ public void setup() throws Exception
+ {
+ super.setup();
+
+ exceptions = new ArrayList<Throwable>();
+ events = new LinkedBlockingQueue<TreeCacheEvent>();
+ eventListener = new TreeCacheListener()
+ {
+ @Override
+ public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception
+ {
+ events.add(event);
+ }
+ };
+
+ client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1));
+ client.start();
+ client.getUnhandledErrorListenable().addListener(new UnhandledErrorListener()
+ {
+ @Override
+ public void unhandledError(String message, Throwable e)
+ {
+ exceptions.add(e);
+ }
+ }
+ );
+ cache = newTreeCache("/test", true);
+ cache.getListenable().addListener(eventListener);
+ }
+
+ @Override
+ @AfterMethod
+ public void teardown() throws Exception
+ {
+ try
+ {
+ try
+ {
+ for ( Throwable exception : exceptions )
+ {
+ Assert.fail("Exception was thrown", exception);
+ }
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache);
+ CloseableUtils.closeQuietly(client);
+ }
+ }
+ finally
+ {
+ super.teardown();
+ }
+ }
+
+ @Test
+ public void testStartup() throws Exception
+ {
+ client.create().forPath("/test");
+ client.create().forPath("/test/1", "one".getBytes());
+ client.create().forPath("/test/2", "two".getBytes());
+ client.create().forPath("/test/3", "three".getBytes());
+ client.create().forPath("/test/2/sub", "two-sub".getBytes());
+
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/1", "one".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/2", "two".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/3", "three".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/2/sub", "two-sub".getBytes());
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+ assertNoMoreEvents();
+
+ Assert.assertEquals(cache.getCurrentChildren("/test"), ImmutableSortedSet.of("1", "2", "3"));
+ Assert.assertEquals(cache.getCurrentChildren("/test/1"), Collections.emptySet());
+ Assert.assertEquals(cache.getCurrentChildren("/test/2"), ImmutableSortedSet.of("sub"));
+ Assert.assertNull(cache.getCurrentChildren("/test/non_exist"));
+ }
+
+ @Test
+ public void testStartEmpty() throws Exception
+ {
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+
+ client.create().forPath("/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testAsyncInitialPopulation() throws Exception
+ {
+ client.create().forPath("/test");
+ client.create().forPath("/test/one", "hey there".getBytes());
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/one");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testSyncInitialPopulation() throws Exception
+ {
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+ client.create().forPath("/test");
+ client.create().forPath("/test/one", "hey there".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/one");
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testChildrenInitialized() throws Exception
+ {
+ client.create().forPath("/test", "".getBytes());
+ client.create().forPath("/test/1", "1".getBytes());
+ client.create().forPath("/test/2", "2".getBytes());
+ client.create().forPath("/test/3", "3".getBytes());
+
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/1");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/2");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/3");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testUpdateWhenNotCachingData() throws Exception
+ {
+ cache = newTreeCache("/test", false);
+ cache.getListenable().addListener(eventListener);
+
+ client.create().forPath("/test");
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+
+ client.create().forPath("/test/foo", "first".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/foo");
+
+ client.setData().forPath("/test/foo", "something new".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_UPDATED, "/test/foo");
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testDeleteThenCreate() throws Exception
+ {
+ client.create().forPath("/test");
+ client.create().forPath("/test/foo", "one".getBytes());
+ cache.start();
+
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/foo");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+
+ client.delete().forPath("/test/foo");
+ assertEvent(TreeCacheEvent.Type.NODE_REMOVED, "/test/foo");
+ client.create().forPath("/test/foo", "two".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/foo");
+
+ assertNoMoreEvents();
+ }
+
+ // see https://github.com/Netflix/curator/issues/27 - was caused by not comparing old->new data
+ @Test
+ public void testIssue27() throws Exception
+ {
+ client.create().forPath("/test");
+ client.create().forPath("/test/a");
+ client.create().forPath("/test/b");
+ client.create().forPath("/test/c");
+
+ client.getChildren().forPath("/test");
+
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/a");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/b");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/c");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+
+ client.delete().forPath("/test/a");
+ client.create().forPath("/test/a");
+ assertEvent(TreeCacheEvent.Type.NODE_REMOVED, "/test/a");
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/a");
+
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testKilledSession() throws Exception
+ {
+ client.create().forPath("/test");
+ cache.start();
+
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+
+ client.create().forPath("/test/foo", "foo".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/foo");
+ client.create().withMode(CreateMode.EPHEMERAL).forPath("/test/me", "data".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/me");
+
+ KillSession.kill(client.getZookeeperClient().getZooKeeper(), server.getConnectString());
+ assertEvent(TreeCacheEvent.Type.CONNECTION_SUSPENDED);
+ assertEvent(TreeCacheEvent.Type.CONNECTION_LOST);
+ assertEvent(TreeCacheEvent.Type.CONNECTION_RECONNECTED);
+ assertEvent(TreeCacheEvent.Type.NODE_REMOVED, "/test/me");
+
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testBasics() throws Exception
+ {
+ client.create().forPath("/test");
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+ Assert.assertEquals(cache.getCurrentChildren("/test"), Collections.emptySortedSet());
+
+ client.create().forPath("/test/one", "hey there".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/one");
+ Assert.assertEquals(cache.getCurrentChildren("/test"), ImmutableSortedSet.of("one"));
+ Assert.assertEquals(new String(cache.getCurrentData("/test/one").getData()), "hey there");
+
+ client.setData().forPath("/test/one", "sup!".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_UPDATED, "/test/one");
+ Assert.assertEquals(cache.getCurrentChildren("/test"), ImmutableSortedSet.of("one"));
+ Assert.assertEquals(new String(cache.getCurrentData("/test/one").getData()), "sup!");
+
+ client.delete().forPath("/test/one");
+ assertEvent(TreeCacheEvent.Type.NODE_REMOVED, "/test/one");
+ Assert.assertEquals(cache.getCurrentChildren("/test"), Collections.emptySortedSet());
+
+ assertNoMoreEvents();
+ }
+
+ @Test
+ public void testBasicsOnTwoCaches() throws Exception
+ {
+ TreeCache cache2 = newTreeCache("/test", true);
+
+ // Just ensures the same event count; enables test flow control on cache2.
+ final Semaphore semaphore = new Semaphore(0);
+ cache2.getListenable().addListener(new TreeCacheListener()
+ {
+ @Override
+ public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception
+ {
+ semaphore.release();
+ }
+ });
+
+ try
+ {
+ client.create().forPath("/test");
+ cache.start();
+ cache2.start();
+
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+ semaphore.acquire(2);
+
+ client.create().forPath("/test/one", "hey there".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/one");
+ Assert.assertEquals(new String(cache.getCurrentData("/test/one").getData()), "hey there");
+ semaphore.acquire();
+ Assert.assertEquals(new String(cache2.getCurrentData("/test/one").getData()), "hey there");
+
+ client.setData().forPath("/test/one", "sup!".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_UPDATED, "/test/one");
+ Assert.assertEquals(new String(cache.getCurrentData("/test/one").getData()), "sup!");
+ semaphore.acquire();
+ Assert.assertEquals(new String(cache2.getCurrentData("/test/one").getData()), "sup!");
+
+ client.delete().forPath("/test/one");
+ assertEvent(TreeCacheEvent.Type.NODE_REMOVED, "/test/one");
+ Assert.assertNull(cache.getCurrentData("/test/one"));
+ semaphore.acquire();
+ Assert.assertNull(cache2.getCurrentData("/test/one"));
+
+ assertNoMoreEvents();
+ Assert.assertEquals(semaphore.availablePermits(), 0);
+ }
+ finally
+ {
+ CloseableUtils.closeQuietly(cache2);
+ }
+ }
+
+ @Test
+ public void testDeleteNodeAfterCloseDoesntCallExecutor() throws Exception
+ {
+ client.create().forPath("/test");
+
+ cache.start();
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
+ assertEvent(TreeCacheEvent.Type.INITIALIZED);
+
+ client.create().forPath("/test/one", "hey there".getBytes());
+ assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/one");
+ Assert.assertEquals(new String(cache.getCurrentData("/test/one").getData()), "hey there");
+
+ cache.close();
+ assertNoMoreEvents();
+
+ client.delete().forPath("/test/one");
+ assertNoMoreEvents();
+ }
+
+ private void assertNoMoreEvents() throws InterruptedException
+ {
+ timing.sleepABit();
+ Assert.assertTrue(events.isEmpty());
+ }
+
+ private TreeCacheEvent assertEvent(TreeCacheEvent.Type expectedType) throws InterruptedException
+ {
+ return assertEvent(expectedType, null);
+ }
+
+ private TreeCacheEvent assertEvent(TreeCacheEvent.Type expectedType, String expectedPath) throws InterruptedException
+ {
+ return assertEvent(expectedType, expectedPath, null);
+ }
+
+ private TreeCacheEvent assertEvent(TreeCacheEvent.Type expectedType, String expectedPath, byte[] expectedData) throws InterruptedException
+ {
+ TreeCacheEvent event = events.poll(timing.forWaiting().seconds(), TimeUnit.SECONDS);
+ Assert.assertEquals(event.getType(), expectedType);
+ if ( expectedPath == null )
+ {
+ Assert.assertNull(event.getData());
+ }
+ else
+ {
+ Assert.assertEquals(event.getData().getPath(), expectedPath);
+ }
+ if ( expectedData != null )
+ {
+ Assert.assertEquals(event.getData().getData(), expectedData);
+ }
+ return event;
+ }
+}
[4/5] git commit: Use Splitter; remove println's.
Posted by ra...@apache.org.
Use Splitter; remove println's.
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/eb1bae5b
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/eb1bae5b
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/eb1bae5b
Branch: refs/heads/CURATOR-33
Commit: eb1bae5b3ed189dd38283ccdb2e0df6fd79dd2bd
Parents: 678e200
Author: Scott Blum <sc...@squareup.com>
Authored: Mon Jul 28 14:05:51 2014 -0400
Committer: Scott Blum <sc...@squareup.com>
Committed: Mon Jul 28 14:05:51 2014 -0400
----------------------------------------------------------------------
.../java/org/apache/curator/utils/ZKPaths.java | 21 ++++----------------
.../framework/recipes/cache/TreeCache.java | 3 ---
2 files changed, 4 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/eb1bae5b/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
----------------------------------------------------------------------
diff --git a/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java b/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
index 1ffd71b..1e17217 100644
--- a/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
+++ b/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
@@ -18,13 +18,13 @@
*/
package org.apache.curator.utils;
+import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -112,6 +112,8 @@ public class ZKPaths
return new PathAndNode(parentPath, node);
}
+ private static final Splitter PATH_SPLITTER = Splitter.on('/').omitEmptyStrings();
+
/**
* Given a full path, return the the individual parts, without slashes.
* The root path will return an empty list.
@@ -122,22 +124,7 @@ public class ZKPaths
public static List<String> split(String path)
{
PathUtils.validatePath(path);
- if (path.length() == 1) {
- return Collections.emptyList();
- }
- int lastPos = 1;
- List<String> parts = new ArrayList<String>();
- while (true)
- {
- int nextPos = path.indexOf('/', lastPos);
- if (nextPos < 0) {
- parts.add(path.substring(lastPos));
- break;
- }
- parts.add(path.substring(lastPos, nextPos));
- lastPos = nextPos + 1;
- }
- return parts;
+ return PATH_SPLITTER.splitToList(path);
}
/**
http://git-wip-us.apache.org/repos/asf/curator/blob/eb1bae5b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
index f71c21d..4781253 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
@@ -167,7 +167,6 @@ public class TreeCache implements Closeable
@Override
public void process(WatchedEvent event)
{
- System.out.println(event);
try
{
switch ( event.getType() )
@@ -196,7 +195,6 @@ public class TreeCache implements Closeable
@Override
public void processResult(CuratorFramework client, CuratorEvent event) throws Exception
{
- System.out.println(event);
switch ( event.getType() )
{
case EXISTS:
@@ -583,7 +581,6 @@ public class TreeCache implements Closeable
private void publishEvent(final TreeCacheEvent event)
{
- System.out.println(event);
if ( treeState.get() != TreeState.CLOSED )
{
executorService.submit(new Runnable()
[3/5] git commit: Fix event ordering on JDK 1.7
Posted by ra...@apache.org.
Fix event ordering on JDK 1.7
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/678e2002
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/678e2002
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/678e2002
Branch: refs/heads/CURATOR-33
Commit: 678e2002cdc060659e8ccd61f7ac95683781ab04
Parents: f31c1f9
Author: Scott Blum <sc...@squareup.com>
Authored: Thu Jul 24 06:24:37 2014 -0400
Committer: Scott Blum <sc...@squareup.com>
Committed: Thu Jul 24 06:24:37 2014 -0400
----------------------------------------------------------------------
.../framework/recipes/cache/TreeCache.java | 21 ++++++++++++++------
1 file changed, 15 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/678e2002/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
index 735d5e7..f71c21d 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
@@ -40,6 +40,7 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.SortedSet;
import java.util.concurrent.ConcurrentMap;
@@ -230,16 +231,24 @@ public class TreeCache implements Closeable
}
}
+ // Present new children in sorted order for test determinism.
+ List<String> newChildren = new ArrayList<String>();
for ( String child : event.getChildren() )
{
- String fullPath = ZKPaths.makePath(path, child);
if ( !childMap.containsKey(child) )
{
- TreeNode node = new TreeNode(fullPath, this);
- if ( childMap.putIfAbsent(child, node) == null )
- {
- node.wasCreated();
- }
+ newChildren.add(child);
+ }
+ }
+
+ Collections.sort(newChildren);
+ for ( String child : newChildren )
+ {
+ String fullPath = ZKPaths.makePath(path, child);
+ TreeNode node = new TreeNode(fullPath, this);
+ if ( childMap.putIfAbsent(child, node) == null )
+ {
+ node.wasCreated();
}
}
}
[2/5] git commit: Remove Java 8 Collections.emptySortedSet().
Posted by ra...@apache.org.
Remove Java 8 Collections.emptySortedSet().
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/f31c1f9a
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/f31c1f9a
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/f31c1f9a
Branch: refs/heads/CURATOR-33
Commit: f31c1f9a31c0c12ac2886ed1c7edf0dfd4fe0fb0
Parents: 547a7e2
Author: Scott Blum <sc...@squareup.com>
Authored: Wed Jul 23 21:10:35 2014 -0400
Committer: Scott Blum <sc...@squareup.com>
Committed: Wed Jul 23 21:10:35 2014 -0400
----------------------------------------------------------------------
.../org/apache/curator/framework/recipes/cache/TreeCache.java | 3 +--
.../apache/curator/framework/recipes/cache/TestTreeCache.java | 7 +++----
2 files changed, 4 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/f31c1f9a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
index 31bff1e..735d5e7 100644
--- a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/cache/TreeCache.java
@@ -40,7 +40,6 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.SortedSet;
import java.util.concurrent.ConcurrentMap;
@@ -470,7 +469,7 @@ public class TreeCache implements Closeable
SortedSet<String> result;
if ( map == null )
{
- result = Collections.emptySortedSet();
+ result = ImmutableSortedSet.of();
}
else
{
http://git-wip-us.apache.org/repos/asf/curator/blob/f31c1f9a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
index b8a009b..2922d40 100644
--- a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/cache/TestTreeCache.java
@@ -34,7 +34,6 @@ import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -141,7 +140,7 @@ public class TestTreeCache extends BaseClassForTests
assertNoMoreEvents();
Assert.assertEquals(cache.getCurrentChildren("/test"), ImmutableSortedSet.of("1", "2", "3"));
- Assert.assertEquals(cache.getCurrentChildren("/test/1"), Collections.emptySet());
+ Assert.assertEquals(cache.getCurrentChildren("/test/1"), ImmutableSortedSet.of());
Assert.assertEquals(cache.getCurrentChildren("/test/2"), ImmutableSortedSet.of("sub"));
Assert.assertNull(cache.getCurrentChildren("/test/non_exist"));
}
@@ -292,7 +291,7 @@ public class TestTreeCache extends BaseClassForTests
cache.start();
assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test");
assertEvent(TreeCacheEvent.Type.INITIALIZED);
- Assert.assertEquals(cache.getCurrentChildren("/test"), Collections.emptySortedSet());
+ Assert.assertEquals(cache.getCurrentChildren("/test"), ImmutableSortedSet.of());
client.create().forPath("/test/one", "hey there".getBytes());
assertEvent(TreeCacheEvent.Type.NODE_ADDED, "/test/one");
@@ -306,7 +305,7 @@ public class TestTreeCache extends BaseClassForTests
client.delete().forPath("/test/one");
assertEvent(TreeCacheEvent.Type.NODE_REMOVED, "/test/one");
- Assert.assertEquals(cache.getCurrentChildren("/test"), Collections.emptySortedSet());
+ Assert.assertEquals(cache.getCurrentChildren("/test"), ImmutableSortedSet.of());
assertNoMoreEvents();
}
[5/5] git commit: Merge remote-tracking branch 'temp/CURATOR-33' into
CURATOR-33
Posted by ra...@apache.org.
Merge remote-tracking branch 'temp/CURATOR-33' into CURATOR-33
Conflicts:
curator-client/src/test/java/org/apache/curator/utils/TestZKPaths.java
Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/fd5c8aa1
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/fd5c8aa1
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/fd5c8aa1
Branch: refs/heads/CURATOR-33
Commit: fd5c8aa1dcdc60b88f3f229aaba36304a4d90b9a
Parents: dcb44a1 eb1bae5
Author: randgalt <ra...@apache.org>
Authored: Thu Jul 31 10:03:10 2014 -0500
Committer: randgalt <ra...@apache.org>
Committed: Thu Jul 31 10:03:10 2014 -0500
----------------------------------------------------------------------
.../java/org/apache/curator/utils/ZKPaths.java | 16 +
.../framework/recipes/cache/TreeCache.java | 605 +++++++++++++++++++
.../framework/recipes/cache/TreeCacheEvent.java | 126 ++++
.../recipes/cache/TreeCacheListener.java | 37 ++
.../framework/recipes/cache/TestTreeCache.java | 420 +++++++++++++
5 files changed, 1204 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/curator/blob/fd5c8aa1/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
----------------------------------------------------------------------
diff --cc curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
index 820f45f,1e17217..352bfd6
--- a/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
+++ b/curator-client/src/main/java/org/apache/curator/utils/ZKPaths.java
@@@ -16,9 -16,9 +16,10 @@@
* specific language governing permissions and limitations
* under the License.
*/
+
package org.apache.curator.utils;
+ import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;