curator-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From randg...@apache.org
Subject curator git commit: First pass implementation of group membership
Date Sun, 27 Sep 2015 22:00:09 GMT
Repository: curator
Updated Branches:
  refs/heads/CURATOR-267 [created] 3029856c8


First pass implementation of group membership


Project: http://git-wip-us.apache.org/repos/asf/curator/repo
Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/3029856c
Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/3029856c
Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/3029856c

Branch: refs/heads/CURATOR-267
Commit: 3029856c894a72ed2c5762e0c6acd2c0a8cd3937
Parents: f8f05be
Author: randgalt <randgalt@apache.org>
Authored: Sun Sep 27 16:59:48 2015 -0500
Committer: randgalt <randgalt@apache.org>
Committed: Sun Sep 27 16:59:48 2015 -0500

----------------------------------------------------------------------
 .../framework/CuratorFrameworkFactory.java      |  31 ++--
 .../framework/recipes/nodes/GroupMember.java    | 140 +++++++++++++++++++
 .../src/site/confluence/group-member.confluence |  42 ++++++
 .../src/site/confluence/index.confluence        |   3 +-
 .../recipes/nodes/TestGroupMember.java          |  79 +++++++++++
 5 files changed, 281 insertions(+), 14 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
----------------------------------------------------------------------
diff --git a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
index dcb2ee6..41ff9cd 100644
--- a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
+++ b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java
@@ -100,6 +100,24 @@ public class CuratorFrameworkFactory
             build();
     }
 
+    /**
+     * Return the local address as bytes that can be used as a node payload
+     *
+     * @return local address bytes
+     */
+    public static byte[] getLocalAddress()
+    {
+        try
+        {
+            return InetAddress.getLocalHost().getHostAddress().getBytes();
+        }
+        catch ( UnknownHostException ignore )
+        {
+            // ignore
+        }
+        return new byte[0];
+    }
+
     public static class Builder
     {
         private EnsembleProvider ensembleProvider;
@@ -465,19 +483,6 @@ public class CuratorFrameworkFactory
         }
     }
 
-    private static byte[] getLocalAddress()
-    {
-        try
-        {
-            return InetAddress.getLocalHost().getHostAddress().getBytes();
-        }
-        catch ( UnknownHostException ignore )
-        {
-            // ignore
-        }
-        return new byte[0];
-    }
-
     private CuratorFrameworkFactory()
     {
     }

http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
new file mode 100644
index 0000000..5aa8ca2
--- /dev/null
+++ b/curator-recipes/src/main/java/org/apache/curator/framework/recipes/nodes/GroupMember.java
@@ -0,0 +1,140 @@
+/**
+ * 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.nodes;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableMap;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.recipes.cache.ChildData;
+import org.apache.curator.framework.recipes.cache.PathChildrenCache;
+import org.apache.curator.utils.CloseableUtils;
+import org.apache.curator.utils.ZKPaths;
+import java.io.Closeable;
+import java.util.Arrays;
+import java.util.Map;
+
+/**
+ * Group membership management. Adds this instance into a group and
+ * keeps a cache of members in the group
+ */
+public class GroupMember implements Closeable
+{
+    private final PersistentEphemeralNode pen;
+    private final PathChildrenCache cache;
+    private final String thisId;
+    private final byte[] payload;
+
+    /**
+     * @param client client
+     * @param membershipPath the path to use for membership
+     * @param thisId ID of this group member. MUST be unique for the group
+     */
+    public GroupMember(CuratorFramework client, String membershipPath, String thisId)
+    {
+        this(client, membershipPath, thisId, CuratorFrameworkFactory.getLocalAddress());
+    }
+
+    /**
+     * @param client client
+     * @param membershipPath the path to use for membership
+     * @param thisId ID of this group member. MUST be unique for the group
+     * @param payload the payload to write in our member node
+     */
+    public GroupMember(CuratorFramework client, String membershipPath, String thisId, byte[]
payload)
+    {
+        this.thisId = Preconditions.checkNotNull(thisId, "thisId cannot be null");
+        this.payload = Arrays.copyOf(payload, payload.length);
+
+        cache = newPathChildrenCache(client, membershipPath);
+        pen = newPersistentEphemeralNode(client, membershipPath, thisId, payload);
+    }
+
+    /**
+     * Start the group membership. Register thisId as a member and begin
+     * caching all members
+     */
+    public void start()
+    {
+        pen.start();
+        try
+        {
+            cache.start();
+        }
+        catch ( Exception e )
+        {
+            Throwables.propagate(e);
+        }
+    }
+
+    /**
+     * Have thisId leave the group and stop caching membership
+     */
+    @Override
+    public void close()
+    {
+        CloseableUtils.closeQuietly(cache);
+        CloseableUtils.closeQuietly(pen);
+    }
+
+    /**
+     * Return the current view of membership. The keys are the IDs
+     * of the members. The values are each member's payload
+     *
+     * @return membership
+     */
+    public Map<String, byte[]> getCurrentMembers()
+    {
+        ImmutableMap.Builder<String, byte[]> builder = ImmutableMap.builder();
+        boolean thisIdAdded = false;
+        for ( ChildData data : cache.getCurrentData() )
+        {
+            String id = idFromPath(data.getPath());
+            thisIdAdded = thisIdAdded || id.equals(thisId);
+            builder.put(id, data.getData());
+        }
+        if ( !thisIdAdded )
+        {
+            builder.put(thisId, payload);   // this instance is always a member
+        }
+        return builder.build();
+    }
+
+    /**
+     * Given a full ZNode path, return the member ID
+     *
+     * @param path full ZNode path
+     * @return id
+     */
+    public String idFromPath(String path)
+    {
+        return ZKPaths.getNodeFromPath(path);
+    }
+
+    protected PersistentEphemeralNode newPersistentEphemeralNode(CuratorFramework client,
String membershipPath, String thisId, byte[] payload)
+    {
+        return new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL,
ZKPaths.makePath(membershipPath, thisId), payload);
+    }
+
+    protected PathChildrenCache newPathChildrenCache(CuratorFramework client, String membershipPath)
+    {
+        return new PathChildrenCache(client, membershipPath, true);
+    }
+}

http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/site/confluence/group-member.confluence
----------------------------------------------------------------------
diff --git a/curator-recipes/src/site/confluence/group-member.confluence b/curator-recipes/src/site/confluence/group-member.confluence
new file mode 100644
index 0000000..dcb27b3
--- /dev/null
+++ b/curator-recipes/src/site/confluence/group-member.confluence
@@ -0,0 +1,42 @@
+h1. Group Member
+
+h2. Description
+Group membership management. Adds this instance into a group and keeps a cache of members
in the group.
+
+h2. Participating Classes
+* GroupMember
+
+h2. Usage
+h3. Creating a GroupMember
+{code}
+public GroupMember(CuratorFramework client,
+                        String membershipPath,
+                        String thisId,
+                        byte[] payload)
+Parameters:
+client - client instance
+membershipPath - the path to use for membership
+thisId - ID of this group member. MUST be unique for the group
+payload - the payload to write in our member node
+{code}
+
+h3. General Usage
+GroupMember must be started:
+{code}
+group.start();
+{code}
+
+When you are through with the GroupMember instance, you should call close:
+{code}
+group.close();
+{code}
+
+NOTE: this will remove the instance from the group
+
+You can get a current view of the members by calling:
+{code}
+group.getCurrentMembers();
+{code}
+
+h2. Error Handling
+GroupMember instances internally handle all error states recreating the node as necessary.

http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/site/confluence/index.confluence
----------------------------------------------------------------------
diff --git a/curator-recipes/src/site/confluence/index.confluence b/curator-recipes/src/site/confluence/index.confluence
index 4f3a032..71de8df 100644
--- a/curator-recipes/src/site/confluence/index.confluence
+++ b/curator-recipes/src/site/confluence/index.confluence
@@ -29,7 +29,8 @@ regarding "Curator Recipes Own Their ZNode/Paths".
 |[[Tree Cache|tree-cache.html]] \- 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.|
 
 ||Nodes||
-|[[Persistent Ephemeral Node|persistent-ephemeral-node.html]] \- An ephemeral node that attempts
to stay present in ZooKeeper, even through connection and session interruptions..|
+|[[Persistent Ephemeral Node|persistent-ephemeral-node.html]] \- An ephemeral node that attempts
to stay present in ZooKeeper, even through connection and session interruptions.|
+|[Group Member|group-member-node.html]] \- Group membership management. Adds this instance
into a group and keeps a cache of members in the group.|
 
 ||Queues||
 |[[Distributed Queue|distributed-queue.html]] \- An implementation of the Distributed Queue
ZK recipe. Items put into the queue are guaranteed to be ordered (by means of ZK's PERSISTENT_SEQUENTIAL
node). If a single consumer takes items out of the queue, they will be ordered FIFO. If ordering
is important, use a LeaderSelector to nominate a single consumer.|

http://git-wip-us.apache.org/repos/asf/curator/blob/3029856c/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
----------------------------------------------------------------------
diff --git a/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
new file mode 100644
index 0000000..3d4b951
--- /dev/null
+++ b/curator-recipes/src/test/java/org/apache/curator/framework/recipes/nodes/TestGroupMember.java
@@ -0,0 +1,79 @@
+/**
+ * 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.nodes;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.curator.test.BaseClassForTests;
+import org.apache.curator.test.Timing;
+import org.apache.curator.utils.CloseableUtils;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+import java.util.Map;
+
+public class TestGroupMember extends BaseClassForTests
+{
+    // NOTE - don't need many tests as this class is just a wrapper around two existing recipes
+
+    @Test
+    public void testBasic() throws Exception
+    {
+        Timing timing = new Timing();
+        GroupMember groupMember1 = null;
+        GroupMember groupMember2 = null;
+        CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(),
new RetryOneTime(1));
+        try
+        {
+            client.start();
+
+            groupMember1 = new GroupMember(client, "/member", "1");
+            Assert.assertTrue(groupMember1.getCurrentMembers().containsKey("1"));
+            groupMember1.start();
+
+            groupMember2 = new GroupMember(client, "/member", "2");
+            groupMember2.start();
+
+            timing.sleepABit();
+
+            Map<String, byte[]> currentMembers1 = groupMember1.getCurrentMembers();
+            Map<String, byte[]> currentMembers2 = groupMember2.getCurrentMembers();
+            Assert.assertEquals(currentMembers1.size(), 2);
+            Assert.assertEquals(currentMembers2.size(), 2);
+            Assert.assertEquals(currentMembers1, currentMembers2);
+            Assert.assertTrue(currentMembers1.containsKey("1"));
+            Assert.assertTrue(currentMembers1.containsKey("2"));
+
+            groupMember2.close();
+
+            timing.sleepABit();
+
+            currentMembers1 = groupMember1.getCurrentMembers();
+            Assert.assertEquals(currentMembers1.size(), 1);
+            Assert.assertTrue(currentMembers1.containsKey("1"));
+            Assert.assertFalse(currentMembers1.containsKey("2"));
+        }
+        finally
+        {
+            CloseableUtils.closeQuietly(groupMember1);
+            CloseableUtils.closeQuietly(groupMember2);
+            CloseableUtils.closeQuietly(client);
+        }
+    }
+}


Mime
View raw message