You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/03/10 17:50:38 UTC
[1/3] camel git commit: CAMEL-10986: camel-zookeeper-master -
Donation of the master component from fabric8 v1
Repository: camel
Updated Branches:
refs/heads/master 8e14a77d2 -> bf82a5ed1
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/GroupTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/GroupTest.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/GroupTest.java
new file mode 100644
index 0000000..6bc5757
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/GroupTest.java
@@ -0,0 +1,388 @@
+/**
+ * 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.camel.component.zookeepermaster.group;
+
+import java.io.File;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.component.zookeepermaster.group.internal.ChildData;
+import org.apache.camel.component.zookeepermaster.group.internal.ZooKeeperGroup;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryNTimes;
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+public class GroupTest {
+
+ private GroupListener listener = new GroupListener<NodeState>() {
+ @Override
+ public void groupEvent(Group<NodeState> group, GroupListener.GroupEvent event) {
+ boolean connected = group.isConnected();
+ boolean master = group.isMaster();
+ if (connected) {
+ Collection<NodeState> members = group.members().values();
+ System.err.println("GroupEvent: " + event + " (connected=" + connected + ", master=" + master + ", members=" + members + ")");
+ } else {
+ System.err.println("GroupEvent: " + event + " (connected=" + connected + ", master=false)");
+ }
+ }
+ };
+
+ private int findFreePort() throws Exception {
+ ServerSocket ss = new ServerSocket(0);
+ int port = ss.getLocalPort();
+ ss.close();
+ return port;
+ }
+
+ private NIOServerCnxnFactory startZooKeeper(int port) throws Exception {
+ ServerConfig cfg = new ServerConfig();
+ cfg.parse(new String[] {Integer.toString(port), "target/zk/data"});
+
+ ZooKeeperServer zkServer = new ZooKeeperServer();
+ FileTxnSnapLog ftxn = new FileTxnSnapLog(new File(cfg.getDataLogDir()), new File(cfg.getDataDir()));
+ zkServer.setTxnLogFactory(ftxn);
+ zkServer.setTickTime(cfg.getTickTime());
+ zkServer.setMinSessionTimeout(6000);
+ zkServer.setMaxSessionTimeout(9000);
+ NIOServerCnxnFactory cnxnFactory = new NIOServerCnxnFactory();
+ cnxnFactory.configure(cfg.getClientPortAddress(), cfg.getMaxClientCnxns());
+ cnxnFactory.startup(zkServer);
+ return cnxnFactory;
+ }
+
+
+ @Test
+ public void testOrder() throws Exception {
+ int port = findFreePort();
+
+ CuratorFramework curator = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .retryPolicy(new RetryNTimes(10, 100))
+ .build();
+ curator.start();
+
+
+ final String path = "/singletons/test/Order" + System.currentTimeMillis();
+ ArrayList<ZooKeeperGroup> members = new ArrayList<ZooKeeperGroup>();
+ for (int i = 0; i < 4; i++) {
+ ZooKeeperGroup<NodeState> group = new ZooKeeperGroup<NodeState>(curator, path, NodeState.class);
+ group.add(listener);
+ members.add(group);
+ }
+
+ for (ZooKeeperGroup group : members) {
+ assertFalse(group.isConnected());
+ assertFalse(group.isMaster());
+ }
+
+ NIOServerCnxnFactory cnxnFactory = startZooKeeper(port);
+
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+
+ // first to start should be master if members are ordered...
+ int i = 0;
+ for (ZooKeeperGroup group : members) {
+ group.start();
+ group.update(new NodeState("foo" + i));
+ i++;
+
+ // wait for registration
+ while (group.getId() == null) {
+ TimeUnit.MILLISECONDS.sleep(100);
+ }
+ }
+
+ boolean firsStartedIsMaster = members.get(0).isMaster();
+
+ for (ZooKeeperGroup group : members) {
+ group.close();
+ }
+ curator.close();
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+
+ assertTrue("first started is master", firsStartedIsMaster);
+
+ }
+
+ @Test
+ public void testJoinAfterConnect() throws Exception {
+ int port = findFreePort();
+
+ CuratorFramework curator = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .retryPolicy(new RetryNTimes(10, 100))
+ .build();
+ curator.start();
+
+ final Group<NodeState> group = new ZooKeeperGroup<NodeState>(curator, "/singletons/test" + System.currentTimeMillis(), NodeState.class);
+ group.add(listener);
+ group.start();
+
+ assertFalse(group.isConnected());
+ assertFalse(group.isMaster());
+
+ GroupCondition groupCondition = new GroupCondition();
+ group.add(groupCondition);
+
+ NIOServerCnxnFactory cnxnFactory = startZooKeeper(port);
+
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+
+ assertTrue(groupCondition.waitForConnected(5, TimeUnit.SECONDS));
+ assertFalse(group.isMaster());
+
+ group.update(new NodeState("foo"));
+ assertTrue(groupCondition.waitForMaster(5, TimeUnit.SECONDS));
+
+
+ group.close();
+ curator.close();
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+ }
+
+ @Test
+ public void testJoinBeforeConnect() throws Exception {
+ int port = findFreePort();
+
+ CuratorFramework curator = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .retryPolicy(new RetryNTimes(10, 100))
+ .build();
+ curator.start();
+
+ Group<NodeState> group = new ZooKeeperGroup<NodeState>(curator, "/singletons/test" + System.currentTimeMillis(), NodeState.class);
+ group.add(listener);
+ group.start();
+
+ GroupCondition groupCondition = new GroupCondition();
+ group.add(groupCondition);
+
+ assertFalse(group.isConnected());
+ assertFalse(group.isMaster());
+ group.update(new NodeState("foo"));
+
+ NIOServerCnxnFactory cnxnFactory = startZooKeeper(port);
+
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+
+ assertTrue(groupCondition.waitForConnected(5, TimeUnit.SECONDS));
+ assertTrue(groupCondition.waitForMaster(5, TimeUnit.SECONDS));
+
+
+ group.close();
+ curator.close();
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+ }
+
+ @Test
+ public void testRejoinAfterDisconnect() throws Exception {
+ int port = findFreePort();
+
+ CuratorFramework curator = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .retryPolicy(new RetryNTimes(10, 100))
+ .build();
+ curator.start();
+
+ NIOServerCnxnFactory cnxnFactory = startZooKeeper(port);
+ Group<NodeState> group = new ZooKeeperGroup<NodeState>(curator, "/singletons/test" + System.currentTimeMillis(), NodeState.class);
+ group.add(listener);
+ group.update(new NodeState("foo"));
+ group.start();
+
+ GroupCondition groupCondition = new GroupCondition();
+ group.add(groupCondition);
+
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+ assertTrue(groupCondition.waitForConnected(5, TimeUnit.SECONDS));
+ assertTrue(groupCondition.waitForMaster(5, TimeUnit.SECONDS));
+
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+
+ groupCondition.waitForDisconnected(5, TimeUnit.SECONDS);
+ group.remove(groupCondition);
+
+ assertFalse(group.isConnected());
+ assertFalse(group.isMaster());
+
+ groupCondition = new GroupCondition();
+ group.add(groupCondition);
+
+ cnxnFactory = startZooKeeper(port);
+
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+ assertTrue(groupCondition.waitForConnected(5, TimeUnit.SECONDS));
+ assertTrue(groupCondition.waitForMaster(5, TimeUnit.SECONDS));
+
+
+ group.close();
+ curator.close();
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+ }
+
+ //Tests that if close() is executed right after start(), there are no left over entries.
+ //(see https://github.com/jboss-fuse/fuse/issues/133)
+ @Test
+ public void testGroupClose() throws Exception {
+ int port = findFreePort();
+ NIOServerCnxnFactory cnxnFactory = startZooKeeper(port);
+
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .connectionTimeoutMs(6000)
+ .sessionTimeoutMs(6000)
+ .retryPolicy(new RetryNTimes(10, 100));
+ CuratorFramework curator = builder.build();
+ curator.start();
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+ String groupNode = "/singletons/test" + System.currentTimeMillis();
+ curator.create().creatingParentsIfNeeded().forPath(groupNode);
+
+ for (int i = 0; i < 100; i++) {
+ ZooKeeperGroup<NodeState> group = new ZooKeeperGroup<NodeState>(curator, groupNode, NodeState.class);
+ group.add(listener);
+ group.update(new NodeState("foo"));
+ group.start();
+ group.close();
+ List<String> entries = curator.getChildren().forPath(groupNode);
+ assertTrue(entries.isEmpty() || group.isUnstable());
+ if (group.isUnstable()) {
+ // let's wait for session timeout
+ curator.close();
+ curator = builder.build();
+ curator.start();
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+ }
+ }
+
+ curator.close();
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+ }
+
+ @Test
+ public void testAddFieldIgnoredOnParse() throws Exception {
+
+ int port = findFreePort();
+ NIOServerCnxnFactory cnxnFactory = startZooKeeper(port);
+
+ CuratorFramework curator = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .retryPolicy(new RetryNTimes(10, 100))
+ .build();
+ curator.start();
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+ String groupNode = "/singletons/test" + System.currentTimeMillis();
+ curator.create().creatingParentsIfNeeded().forPath(groupNode);
+
+ curator.getZookeeperClient().blockUntilConnectedOrTimedOut();
+
+ final ZooKeeperGroup<NodeState> group = new ZooKeeperGroup<NodeState>(curator, groupNode, NodeState.class);
+ group.add(listener);
+ group.start();
+
+ GroupCondition groupCondition = new GroupCondition();
+ group.add(groupCondition);
+
+ group.update(new NodeState("foo"));
+
+ assertTrue(groupCondition.waitForConnected(5, TimeUnit.SECONDS));
+ assertTrue(groupCondition.waitForMaster(5, TimeUnit.SECONDS));
+
+ ChildData currentData = group.getCurrentData().get(0);
+ final int version = currentData.getStat().getVersion();
+
+ NodeState lastState = group.getLastState();
+ String json = lastState.toString();
+ System.err.println("JSON:" + json);
+
+ String newValWithNewField = json.substring(0, json.lastIndexOf('}')) + ",\"Rubbish\":\"Rubbish\"}";
+ curator.getZookeeperClient().getZooKeeper().setData(group.getId(), newValWithNewField.getBytes(), version);
+
+ assertTrue(group.isMaster());
+
+ int attempts = 0;
+ while (attempts++ < 5 && version == group.getCurrentData().get(0).getStat().getVersion()) {
+ TimeUnit.SECONDS.sleep(1);
+ }
+
+ assertNotEquals("We see the updated version", version, group.getCurrentData().get(0).getStat().getVersion());
+
+ System.err.println("CurrentData:" + group.getCurrentData());
+
+ group.close();
+ curator.close();
+ cnxnFactory.shutdown();
+ cnxnFactory.join();
+ }
+
+ private class GroupCondition implements GroupListener<NodeState> {
+ private CountDownLatch connected = new CountDownLatch(1);
+ private CountDownLatch master = new CountDownLatch(1);
+ private CountDownLatch disconnected = new CountDownLatch(1);
+
+ @Override
+ public void groupEvent(Group<NodeState> group, GroupEvent event) {
+ switch (event) {
+ case CONNECTED:
+ case CHANGED:
+ connected.countDown();
+ if (group.isMaster()) {
+ master.countDown();
+ }
+ break;
+ case DISCONNECTED:
+ disconnected.countDown();
+ break;
+ default:
+ // noop
+ }
+ }
+
+ public boolean waitForConnected(long time, TimeUnit timeUnit) throws InterruptedException {
+ return connected.await(time, timeUnit);
+ }
+
+ public boolean waitForDisconnected(long time, TimeUnit timeUnit) throws InterruptedException {
+ return disconnected.await(time, timeUnit);
+ }
+
+ public boolean waitForMaster(long time, TimeUnit timeUnit) throws InterruptedException {
+ return master.await(time, timeUnit);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupTest.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupTest.java
new file mode 100644
index 0000000..9e00621
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupTest.java
@@ -0,0 +1,232 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.List;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.RetryOneTime;
+import org.apache.zookeeper.data.Stat;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.junit.Assert.assertThat;
+
+public class ZooKeeperGroupTest {
+
+ private static final String PATH = "/singletons/test/" + ZooKeeperGroupTest.class.getSimpleName();
+
+ private CuratorFramework curator;
+ private ZooKeeperGroup<NodeState> group;
+
+ private int findFreePort() throws Exception {
+ ServerSocket ss = new ServerSocket(0);
+ int port = ss.getLocalPort();
+ ss.close();
+ return port;
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ int port = findFreePort();
+ curator = CuratorFrameworkFactory.builder()
+ .connectString("localhost:" + port)
+ .retryPolicy(new RetryOneTime(1))
+ .build();
+ //curator.start();
+ group = new ZooKeeperGroup<>(curator, PATH, NodeState.class);
+ //group.start();
+ // Starting curator and group is not necessary for the current tests.
+ }
+
+ @After
+ public void tearDown() throws IOException {
+ group.close();
+ curator.close();
+ group = null;
+ curator = null;
+ }
+
+ private static void putChildData(ZooKeeperGroup<NodeState> group, String path, String container) throws Exception {
+ NodeState node = new NodeState("test", container);
+ ByteArrayOutputStream data = new ByteArrayOutputStream();
+ new ObjectMapper().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES).writeValue(data, node);
+ ChildData<NodeState> child = new ChildData<>(path, new Stat(), data.toByteArray(), node);
+ group.putCurrentData(path, child);
+ }
+
+ @Test
+ public void testMembers() throws Exception {
+ putChildData(group, PATH + "/001", "container1");
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container3");
+
+ Map<String, NodeState> members = group.members();
+ assertThat(members.size(), equalTo(3));
+ assertThat(members.get(PATH + "/001").getContainer(), equalTo("container1"));
+ assertThat(members.get(PATH + "/002").getContainer(), equalTo("container2"));
+ assertThat(members.get(PATH + "/003").getContainer(), equalTo("container3"));
+ }
+
+ @Test
+ public void testMembersWithStaleNodes() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container1");
+ putChildData(group, PATH + "/003", "container2"); // stale
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container2");
+ putChildData(group, PATH + "/006", "container3");
+
+ Map<String, NodeState> members = group.members();
+ assertThat(members.size(), equalTo(3));
+ assertThat(members.get(PATH + "/002").getContainer(), equalTo("container1"));
+ assertThat(members.get(PATH + "/005").getContainer(), equalTo("container2"));
+ assertThat(members.get(PATH + "/006").getContainer(), equalTo("container3"));
+ }
+
+ @Test
+ public void testIsMaster() throws Exception {
+ putChildData(group, PATH + "/001", "container1");
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container3");
+
+ group.setId(PATH + "/001");
+ assertThat(group.isMaster(), equalTo(true));
+ group.setId(PATH + "/002");
+ assertThat(group.isMaster(), equalTo(false));
+ }
+
+ @Test
+ public void testIsMasterWithStaleNodes1() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container1");
+ putChildData(group, PATH + "/003", "container2"); // stale
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container2");
+ putChildData(group, PATH + "/006", "container3");
+
+ group.setId(PATH + "/002");
+ assertThat(group.isMaster(), equalTo(true));
+ group.setId(PATH + "/005");
+ assertThat(group.isMaster(), equalTo(false));
+ }
+
+ @Test
+ public void testIsMasterWithStaleNodes2() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container1");
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container3");
+
+ group.setId(PATH + "/002");
+ assertThat(group.isMaster(), equalTo(true));
+ group.setId(PATH + "/003");
+ assertThat(group.isMaster(), equalTo(false));
+ }
+
+ @Test
+ public void testMaster() throws Exception {
+ putChildData(group, PATH + "/001", "container1");
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container3");
+
+ NodeState master = group.master();
+ assertThat(master, notNullValue());
+ assertThat(master.getContainer(), equalTo("container1"));
+ }
+
+ @Test
+ public void testMasterWithStaleNodes1() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container1");
+ putChildData(group, PATH + "/003", "container2"); // stale
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container2");
+ putChildData(group, PATH + "/006", "container3");
+
+ NodeState master = group.master();
+ assertThat(master, notNullValue());
+ assertThat(master.getContainer(), equalTo("container1"));
+ }
+
+ @Test
+ public void testMasterWithStaleNodes2() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container1");
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container3");
+
+ NodeState master = group.master();
+ assertThat(master, notNullValue());
+ assertThat(master.getContainer(), equalTo("container2"));
+ }
+
+ @Test
+ public void testSlaves() throws Exception {
+ putChildData(group, PATH + "/001", "container1");
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container3");
+
+ List<NodeState> slaves = group.slaves();
+ assertThat(slaves.size(), equalTo(2));
+ assertThat(slaves.get(0).getContainer(), equalTo("container2"));
+ assertThat(slaves.get(1).getContainer(), equalTo("container3"));
+ }
+
+ @Test
+ public void testSlavesWithStaleNodes1() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container1");
+ putChildData(group, PATH + "/003", "container2"); // stale
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container2");
+ putChildData(group, PATH + "/006", "container3");
+
+ List<NodeState> slaves = group.slaves();
+ assertThat(slaves.size(), equalTo(2));
+ assertThat(slaves.get(0).getContainer(), equalTo("container2"));
+ assertThat(slaves.get(1).getContainer(), equalTo("container3"));
+ }
+
+ @Test
+ public void testSlavesWithStaleNodes2() throws Exception {
+ putChildData(group, PATH + "/001", "container1"); // stale
+ putChildData(group, PATH + "/002", "container2");
+ putChildData(group, PATH + "/003", "container1");
+ putChildData(group, PATH + "/004", "container3"); // stale
+ putChildData(group, PATH + "/005", "container3");
+
+ List<NodeState> slaves = group.slaves();
+ assertThat(slaves.size(), equalTo(2));
+ assertThat(slaves.get(0).getContainer(), equalTo("container1"));
+ assertThat(slaves.get(1).getContainer(), equalTo("container3"));
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/resources/log4j2.properties
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/resources/log4j2.properties b/components/camel-zookeeper-master/src/test/resources/log4j2.properties
new file mode 100644
index 0000000..409eaf7
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/resources/log4j2.properties
@@ -0,0 +1,45 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+
+appender.file.type = File
+appender.file.name = file
+appender.file.fileName = target/camel-zookeeper-master-test.log
+appender.file.layout.type = PatternLayout
+appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
+appender.out.type = Console
+appender.out.name = out
+appender.out.layout.type = PatternLayout
+# appender.out.layout.pattern = %highlight{%d [%t] %-5level: %msg%n%throwable}{FATAL=red, ERROR=red, WARN=blue, INFO=black, DEBUG=grey, TRACE=blue}
+appender.out.layout.pattern = [%t] %c{1} %-5p %m%n
+
+logger.zookeeper.name = org.apache.zookeeper
+logger.zookeeper.level = INFO
+logger.camel-zookeeper.name = org.apache.camel.component.zookeepermaster
+logger.camel-zookeeper.level = INFO
+logger.camel-support.name = org.apache.camel.support
+logger.camel-support.level = INFO
+logger.camel.name = org.apache.camel
+logger.camel.level = INFO
+
+
+logger.springframework.name = org.springframework
+logger.springframework.level = WARN
+rootLogger.level = INFO
+#rootLogger.appenderRef.stdout.ref = out
+rootLogger.appenderRef.file.ref = file
+
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterEndpointTest-context.xml
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterEndpointTest-context.xml b/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterEndpointTest-context.xml
new file mode 100644
index 0000000..1e1ca47
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterEndpointTest-context.xml
@@ -0,0 +1,43 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
+
+ <!-- Since this is a test case lets run a local ZK server -->
+ <bean id="zkServer" class="org.apache.camel.component.zookeepermaster.ZKServerFactoryBean">
+ <property name="port" value="9003"/>
+ </bean>
+
+ <bean id="curator" class="org.apache.camel.component.zookeepermaster.CuratorFactoryBean" depends-on="zkServer">
+ <property name="timeout" value="3000"/>
+ <property name="connectString" value="localhost:9003"/>
+ </bean>
+
+ <camelContext xmlns="http://camel.apache.org/schema/spring" depends-on="curator">
+
+ <route>
+ <from uri="zookeeper-master:master-000:seda:bar"/>
+ <to uri="mock:results"/>
+ </route>
+
+ </camelContext>
+
+ <!-- some other stuff here... -->
+
+</beans>
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest-context.xml
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest-context.xml b/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest-context.xml
new file mode 100644
index 0000000..fe549e3
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/resources/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest-context.xml
@@ -0,0 +1,45 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+ http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
+
+ <!-- Since this is a test case lets run a local ZK server -->
+ <bean id="zkServer" class="org.apache.camel.component.zookeepermaster.ZKServerFactoryBean">
+ <property name="port" value="9003"/>
+ </bean>
+
+
+ <bean id="curator" class="org.apache.camel.component.zookeepermaster.CuratorFactoryBean" depends-on="zkServer">
+ <property name="timeout" value="3000"/>
+ <property name="connectString" value="localhost:9003"/>
+ </bean>
+
+ <camelContext xmlns="http://camel.apache.org/schema/spring" depends-on="curator">
+
+ <route>
+ <from uri="zookeeper-master:master-000:quartz2://masterTest?cron=0/2+0/1+*+1/1+*+?+*"/>
+
+ <to uri="mock:results"/>
+ </route>
+
+ </camelContext>
+
+ <!-- some other stuff here... -->
+
+</beans>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/readme.adoc
----------------------------------------------------------------------
diff --git a/components/readme.adoc b/components/readme.adoc
index 0d26279..35e9644 100644
--- a/components/readme.adoc
+++ b/components/readme.adoc
@@ -2,7 +2,7 @@ Components
^^^^^^^^^^
// components: START
-Number of Components: 221 in 175 JAR artifacts (13 deprecated)
+Number of Components: 222 in 176 JAR artifacts (13 deprecated)
[width="100%",cols="4,1,5",options="header"]
|=======================================================================
@@ -632,7 +632,7 @@ Number of Components: 221 in 175 JAR artifacts (13 deprecated)
| link:camel-twitter/src/main/docs/twitter-component.adoc[Twitter] (camel-twitter) +
`twitter:kind` | 2.10 | This component integrates with Twitter to send tweets or search for tweets and more.
-| link:camel-undertow/src/main/docs/undertow-component.adoc[Undertow] (camel-undertow) +
+| link:null/src/main/docs/undertow-component.adoc[Undertow] (null) +
`undertow:httpURI` | 2.16 | The undertow component provides HTTP-based endpoints for consuming and producing HTTP requests.
| link:../camel-core/src/main/docs/validator-component.adoc[Validator] (camel-core) +
@@ -671,6 +671,9 @@ Number of Components: 221 in 175 JAR artifacts (13 deprecated)
| link:camel-zookeeper/src/main/docs/zookeeper-component.adoc[ZooKeeper] (camel-zookeeper) +
`zookeeper:serverUrls/path` | 2.9 | The zookeeper component allows interaction with a ZooKeeper cluster.
+| link:camel-zookeeper-master/src/main/docs/zookeeper-master-component.adoc[ZooKeeper Master] (camel-zookeeper-master) +
+`zookeeper-master:name:endpoint` | 2.19 | Represents an endpoint which only becomes active when it obtains the master lock
+
|=======================================================================
// components: END
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/docs/user-manual/en/SUMMARY.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index 1edd5ca..e8bd83b 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -320,6 +320,7 @@
* [XQuery](xquery-component.adoc)
* [Yammer](yammer-component.adoc)
* [ZooKeeper](zookeeper-component.adoc)
+ * [ZooKeeper Master](zookeeper-master-component.adoc)
<!-- components: END -->
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index 0e95441..ee6d05a 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -2037,6 +2037,11 @@
<artifactId>camel-zookeeper</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-zookeeper-master</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<!-- camel catalog -->
<dependency>
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/pom.xml
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/pom.xml b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/pom.xml
new file mode 100644
index 0000000..a7abe74
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/pom.xml
@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>components-starter</artifactId>
+ <version>2.19.0-SNAPSHOT</version>
+ </parent>
+ <artifactId>camel-zookeeper-master-starter</artifactId>
+ <packaging>jar</packaging>
+ <name>Spring-Boot Starter :: Camel :: Zookeeper Master</name>
+ <description>Spring-Boot Starter for Camel Zookeeper Master Support</description>
+ <dependencies>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter</artifactId>
+ <version>${spring-boot-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-zookeeper-master</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <!--START OF GENERATED CODE-->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core-starter</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-spring-boot-starter</artifactId>
+ </dependency>
+ <!--END OF GENERATED CODE-->
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentAutoConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentAutoConfiguration.java b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentAutoConfiguration.java
new file mode 100644
index 0000000..5afd8cc
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentAutoConfiguration.java
@@ -0,0 +1,110 @@
+/**
+ * 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.camel.component.zookeepermaster.springboot;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.zookeepermaster.MasterComponent;
+import org.apache.camel.util.IntrospectionSupport;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionMessage;
+import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.SpringBootCondition;
+import org.springframework.boot.bind.RelaxedPropertyResolver;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.ConditionContext;
+import org.springframework.context.annotation.Conditional;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.core.type.AnnotatedTypeMetadata;
+
+/**
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@Configuration
+@ConditionalOnBean(type = "org.apache.camel.spring.boot.CamelAutoConfiguration")
+@Conditional(MasterComponentAutoConfiguration.Condition.class)
+@AutoConfigureAfter(name = "org.apache.camel.spring.boot.CamelAutoConfiguration")
+@EnableConfigurationProperties(MasterComponentConfiguration.class)
+public class MasterComponentAutoConfiguration {
+
+ @Lazy
+ @Bean(name = "zookeeper-master-component")
+ @ConditionalOnClass(CamelContext.class)
+ @ConditionalOnMissingBean(MasterComponent.class)
+ public MasterComponent configureMasterComponent(CamelContext camelContext,
+ MasterComponentConfiguration configuration) throws Exception {
+ MasterComponent component = new MasterComponent();
+ component.setCamelContext(camelContext);
+ Map<String, Object> parameters = new HashMap<>();
+ IntrospectionSupport.getProperties(configuration, parameters, null,
+ false);
+ for (Map.Entry<String, Object> entry : parameters.entrySet()) {
+ Object value = entry.getValue();
+ Class<?> paramClass = value.getClass();
+ if (paramClass.getName().endsWith("NestedConfiguration")) {
+ Class nestedClass = null;
+ try {
+ nestedClass = (Class) paramClass.getDeclaredField(
+ "CAMEL_NESTED_CLASS").get(null);
+ HashMap<String, Object> nestedParameters = new HashMap<>();
+ IntrospectionSupport.getProperties(value, nestedParameters,
+ null, false);
+ Object nestedProperty = nestedClass.newInstance();
+ IntrospectionSupport.setProperties(camelContext,
+ camelContext.getTypeConverter(), nestedProperty,
+ nestedParameters);
+ entry.setValue(nestedProperty);
+ } catch (NoSuchFieldException e) {
+ }
+ }
+ }
+ IntrospectionSupport.setProperties(camelContext,
+ camelContext.getTypeConverter(), component, parameters);
+ return component;
+ }
+
+ public static class Condition extends SpringBootCondition {
+ @Override
+ public ConditionOutcome getMatchOutcome(
+ ConditionContext conditionContext,
+ AnnotatedTypeMetadata annotatedTypeMetadata) {
+ boolean groupEnabled = isEnabled(conditionContext,
+ "camel.component.", true);
+ ConditionMessage.Builder message = ConditionMessage
+ .forCondition("camel.component.zookeeper-master");
+ if (isEnabled(conditionContext,
+ "camel.component.zookeeper-master.", groupEnabled)) {
+ return ConditionOutcome.match(message.because("enabled"));
+ }
+ return ConditionOutcome.noMatch(message.because("not enabled"));
+ }
+
+ private boolean isEnabled(
+ org.springframework.context.annotation.ConditionContext context,
+ java.lang.String prefix, boolean defaultValue) {
+ RelaxedPropertyResolver resolver = new RelaxedPropertyResolver(
+ context.getEnvironment(), prefix);
+ return resolver.getProperty("enabled", Boolean.class, defaultValue);
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentConfiguration.java
new file mode 100644
index 0000000..8408447
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/java/org/apache/camel/component/zookeepermaster/springboot/MasterComponentConfiguration.java
@@ -0,0 +1,111 @@
+/**
+ * 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.camel.component.zookeepermaster.springboot;
+
+import org.apache.curator.framework.CuratorFramework;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.NestedConfigurationProperty;
+
+/**
+ * Represents an endpoint which only becomes active when it obtains the master
+ * lock
+ *
+ * Generated by camel-package-maven-plugin - do not edit this file!
+ */
+@ConfigurationProperties(prefix = "camel.component.zookeeper-master")
+public class MasterComponentConfiguration {
+
+ /**
+ * The root path to use in zookeeper where information is stored which nodes
+ * are master/slave etc. Will by default use:
+ * /camel/zookeepermaster/clusters/master
+ */
+ private String zkRoot = "/camel/zookeepermaster/clusters/master";
+ /**
+ * To use a custom configured CuratorFramework as connection to zookeeper
+ * ensemble.
+ */
+ @NestedConfigurationProperty
+ private CuratorFramework curator;
+ /**
+ * Timeout in millis to use when connecting to the zookeeper ensemble
+ */
+ private Integer maximumConnectionTimeout = 10000;
+ /**
+ * The url for the zookeeper ensemble
+ */
+ private String zooKeeperUrl = "localhost:2181";
+ /**
+ * The password to use when connecting to the zookeeper ensemble
+ */
+ private String zooKeeperPassword;
+ /**
+ * Whether the component should resolve property placeholders on itself when
+ * starting. Only properties which are of String type can use property
+ * placeholders.
+ */
+ private Boolean resolvePropertyPlaceholders = true;
+
+ public String getZkRoot() {
+ return zkRoot;
+ }
+
+ public void setZkRoot(String zkRoot) {
+ this.zkRoot = zkRoot;
+ }
+
+ public CuratorFramework getCurator() {
+ return curator;
+ }
+
+ public void setCurator(CuratorFramework curator) {
+ this.curator = curator;
+ }
+
+ public Integer getMaximumConnectionTimeout() {
+ return maximumConnectionTimeout;
+ }
+
+ public void setMaximumConnectionTimeout(Integer maximumConnectionTimeout) {
+ this.maximumConnectionTimeout = maximumConnectionTimeout;
+ }
+
+ public String getZooKeeperUrl() {
+ return zooKeeperUrl;
+ }
+
+ public void setZooKeeperUrl(String zooKeeperUrl) {
+ this.zooKeeperUrl = zooKeeperUrl;
+ }
+
+ public String getZooKeeperPassword() {
+ return zooKeeperPassword;
+ }
+
+ public void setZooKeeperPassword(String zooKeeperPassword) {
+ this.zooKeeperPassword = zooKeeperPassword;
+ }
+
+ public Boolean getResolvePropertyPlaceholders() {
+ return resolvePropertyPlaceholders;
+ }
+
+ public void setResolvePropertyPlaceholders(
+ Boolean resolvePropertyPlaceholders) {
+ this.resolvePropertyPlaceholders = resolvePropertyPlaceholders;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/LICENSE.txt b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ 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.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/NOTICE.txt b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
new file mode 100644
index 0000000..0b8cff5
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/additional-spring-configuration-metadata.json
@@ -0,0 +1,10 @@
+{
+ "properties": [
+ {
+ "defaultValue": true,
+ "name": "camel.component.zookeeper-master.enabled",
+ "description": "Enable zookeeper-master component",
+ "type": "java.lang.Boolean"
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.factories
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.factories b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.factories
new file mode 100644
index 0000000..4dee811
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.apache.camel.component.zookeepermaster.springboot.MasterComponentAutoConfiguration
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.provides
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.provides b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.provides
new file mode 100644
index 0000000..d50cd0a
--- /dev/null
+++ b/platforms/spring-boot/components-starter/camel-zookeeper-master-starter/src/main/resources/META-INF/spring.provides
@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+provides: camel-zookeeper-master
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/platforms/spring-boot/components-starter/pom.xml
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/pom.xml b/platforms/spring-boot/components-starter/pom.xml
index e0e5fca..efb907f 100644
--- a/platforms/spring-boot/components-starter/pom.xml
+++ b/platforms/spring-boot/components-starter/pom.xml
@@ -304,6 +304,7 @@
<module>camel-yammer-starter</module>
<module>camel-zipfile-starter</module>
<module>camel-zipkin-starter</module>
+ <module>camel-zookeeper-master-starter</module>
<module>camel-zookeeper-starter</module>
</modules>
</project>
[3/3] camel git commit: CAMEL-10986: camel-zookeeper-master -
Donation of the master component from fabric8 v1
Posted by da...@apache.org.
CAMEL-10986: camel-zookeeper-master - Donation of the master component from fabric8 v1
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/bf82a5ed
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/bf82a5ed
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/bf82a5ed
Branch: refs/heads/master
Commit: bf82a5ed1980b96eaf2cc012e8b97ef78ffd2229
Parents: 8e14a77
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Mar 10 18:49:53 2017 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Mar 10 18:49:53 2017 +0100
----------------------------------------------------------------------
apache-camel/pom.xml | 4 +
.../src/main/descriptors/common-bin.xml | 1 +
components/camel-zookeeper-master/pom.xml | 157 +++++
.../main/docs/zookeeper-master-component.adoc | 86 +++
.../zookeepermaster/CamelNodeState.java | 37 +
.../zookeepermaster/MasterComponent.java | 76 +++
.../zookeepermaster/MasterConsumer.java | 157 +++++
.../zookeepermaster/MasterEndpoint.java | 98 +++
.../zookeepermaster/ZKComponentSupport.java | 185 +++++
.../component/zookeepermaster/group/Group.java | 100 +++
.../zookeepermaster/group/GroupFactory.java | 31 +
.../zookeepermaster/group/GroupListener.java | 32 +
.../zookeepermaster/group/MultiGroup.java | 26 +
.../zookeepermaster/group/NodeState.java | 69 ++
.../group/internal/ChildData.java | 129 ++++
.../group/internal/CompositeOperation.java | 56 ++
.../group/internal/DelegateZooKeeperGroup.java | 180 +++++
.../internal/DelegateZooKeeperMultiGroup.java | 44 ++
.../group/internal/EventOperation.java | 39 ++
.../group/internal/GetDataOperation.java | 61 ++
.../group/internal/ManagedGroupFactory.java | 28 +
.../internal/ManagedGroupFactoryBuilder.java | 50 ++
.../group/internal/Operation.java | 23 +
.../group/internal/RefreshOperation.java | 61 ++
.../group/internal/SequenceComparator.java | 27 +
.../internal/StaticManagedGroupFactory.java | 66 ++
.../group/internal/UpdateOperation.java | 55 ++
.../group/internal/ZooKeeperGroup.java | 671 +++++++++++++++++++
.../group/internal/ZooKeeperGroupFactory.java | 56 ++
.../group/internal/ZooKeeperMultiGroup.java | 58 ++
.../internal/osgi/OsgiManagedGroupFactory.java | 183 +++++
.../internal/osgi/TrackingZooKeeperGroup.java | 65 ++
.../src/main/resources/META-INF/LICENSE.txt | 203 ++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
.../org/apache/camel/component/zookeeper-master | 17 +
.../zookeepermaster/CuratorFactoryBean.java | 88 +++
.../MasterEndpointFailoverTest.java | 147 ++++
.../zookeepermaster/MasterEndpointTest.java | 100 +++
.../MasterQuartz2EndpointTest.java | 76 +++
.../zookeepermaster/ZKServerFactoryBean.java | 225 +++++++
.../zookeepermaster/group/GroupTest.java | 388 +++++++++++
.../group/internal/ZooKeeperGroupTest.java | 232 +++++++
.../src/test/resources/log4j2.properties | 45 ++
.../MasterEndpointTest-context.xml | 43 ++
.../MasterQuartz2EndpointTest-context.xml | 45 ++
components/readme.adoc | 7 +-
docs/user-manual/en/SUMMARY.md | 1 +
parent/pom.xml | 5 +
.../camel-zookeeper-master-starter/pom.xml | 51 ++
.../MasterComponentAutoConfiguration.java | 110 +++
.../MasterComponentConfiguration.java | 111 +++
.../src/main/resources/META-INF/LICENSE.txt | 203 ++++++
.../src/main/resources/META-INF/NOTICE.txt | 11 +
...dditional-spring-configuration-metadata.json | 10 +
.../main/resources/META-INF/spring.factories | 19 +
.../src/main/resources/META-INF/spring.provides | 18 +
.../spring-boot/components-starter/pom.xml | 1 +
57 files changed, 5076 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/apache-camel/pom.xml
----------------------------------------------------------------------
diff --git a/apache-camel/pom.xml b/apache-camel/pom.xml
index e05f958..db37cc5 100644
--- a/apache-camel/pom.xml
+++ b/apache-camel/pom.xml
@@ -1068,6 +1068,10 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-zookeeper</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-zookeeper-master</artifactId>
+ </dependency>
<!-- camel starters -->
<dependency>
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/apache-camel/src/main/descriptors/common-bin.xml
----------------------------------------------------------------------
diff --git a/apache-camel/src/main/descriptors/common-bin.xml b/apache-camel/src/main/descriptors/common-bin.xml
index 672ed0e..ad89366 100644
--- a/apache-camel/src/main/descriptors/common-bin.xml
+++ b/apache-camel/src/main/descriptors/common-bin.xml
@@ -278,6 +278,7 @@
<include>org.apache.camel:camel-zipkin</include>
<include>org.apache.camel:camel-zipkin-starter</include>
<include>org.apache.camel:camel-zookeeper</include>
+ <include>org.apache.camel:camel-zookeeper-master</include>
</includes>
</dependencySet>
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/pom.xml b/components/camel-zookeeper-master/pom.xml
new file mode 100644
index 0000000..493a266
--- /dev/null
+++ b/components/camel-zookeeper-master/pom.xml
@@ -0,0 +1,157 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>components</artifactId>
+ <groupId>org.apache.camel</groupId>
+ <version>2.19.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>camel-zookeeper-master</artifactId>
+ <packaging>jar</packaging>
+ <name>Camel :: Zookeeper Master</name>
+ <description>Camel Zookeeper Master Support</description>
+
+ <properties>
+ <camel.osgi.export.pkg>org.apache.camel.component.zookeepermaster.*</camel.osgi.export.pkg>
+ <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=zookeeper-master</camel.osgi.export.service>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-core</artifactId>
+ </dependency>
+
+ <!-- zookeeper -->
+ <dependency>
+ <groupId>org.apache.zookeeper</groupId>
+ <artifactId>zookeeper</artifactId>
+ <version>${zookeeper-version}</version>
+ <!-- Just exclude the old version of netty to avoid netty version conflict-->
+ <exclusions>
+ <exclusion>
+ <groupId>org.jboss.netty</groupId>
+ <artifactId>netty</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty</artifactId>
+ <version>${netty3-version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-framework</artifactId>
+ <version>${curator-version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.curator</groupId>
+ <artifactId>curator-recipes</artifactId>
+ <version>${curator-version}</version>
+ </dependency>
+
+ <!-- json state -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+
+ <!-- optional OSGi support -->
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>org.osgi.core</artifactId>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
+ <groupId>org.osgi</groupId>
+ <artifactId>org.osgi.compendium</artifactId>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
+ <!-- test dependencies -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-quartz2</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-spring</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <childDelegation>false</childDelegation>
+ <useFile>true</useFile>
+ <forkCount>1</forkCount>
+ <reuseForks>true</reuseForks>
+ <forkedProcessTimeoutInSeconds>600</forkedProcessTimeoutInSeconds>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+
+</project>
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/docs/zookeeper-master-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/docs/zookeeper-master-component.adoc b/components/camel-zookeeper-master/src/main/docs/zookeeper-master-component.adoc
new file mode 100644
index 0000000..be3530f
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/docs/zookeeper-master-component.adoc
@@ -0,0 +1,86 @@
+## ZooKeeper Master Component
+
+*Available as of Camel version 2.19*
+
+The **zookeeper-master:** endpoint provides a way to ensure only a single consumer in a cluster consumes from a given endpoint;
+with automatic failover if that JVM dies.
+
+This can be very useful if you need to consume from some legacy back end which either doesn't support concurrent
+consumption or due to commercial or stability reasons you can only have a single connection at any point in time.
+
+### Using the master endpoint
+
+Just prefix any camel endpoint with **zookeeper-master:someName:** where _someName_ is a logical name and is
+used to acquire the master lock. e.g.
+
+```
+from("zookeeper-master:cheese:jms:foo").to("activemq:wine");
+```
+The above simulates the [Exclusive Consumers](http://activemq.apache.org/exclusive-consumer.html) type feature in
+ActiveMQ; but on any third party JMS provider which maybe doesn't support exclusive consumers.
+
+
+### URI format
+
+[source]
+----
+zookeeper-master:name:endpoint[?options]
+----
+
+Where endpoint is any Camel endpoint you want to run in master/slave mode.
+
+
+### Options
+
+// component options: START
+The ZooKeeper Master component supports 6 options which are listed below.
+
+
+
+[width="100%",cols="2,1,1m,1m,5",options="header"]
+|=======================================================================
+| Name | Group | Default | Java Type | Description
+| zkRoot | consumer | /camel/zookeepermaster/clusters/master | String | The root path to use in zookeeper where information is stored which nodes are master/slave etc. Will by default use: /camel/zookeepermaster/clusters/master
+| curator | advanced | | CuratorFramework | To use a custom configured CuratorFramework as connection to zookeeper ensemble.
+| maximumConnectionTimeout | consumer | 10000 | int | Timeout in millis to use when connecting to the zookeeper ensemble
+| zooKeeperUrl | consumer | localhost:2181 | String | The url for the zookeeper ensemble
+| zooKeeperPassword | security | | String | The password to use when connecting to the zookeeper ensemble
+| resolvePropertyPlaceholders | advanced | true | boolean | Whether the component should resolve property placeholders on itself when starting. Only properties which are of String type can use property placeholders.
+|=======================================================================
+// component options: END
+
+// endpoint options: START
+The ZooKeeper Master endpoint is configured using URI syntax:
+
+ zookeeper-master:name:endpoint
+
+with the following path and query parameters:
+
+#### Path Parameters (2 parameters):
+
+[width="100%",cols="2,1,1m,6",options="header"]
+|=======================================================================
+| Name | Default | Java Type | Description
+| name | | String | *Required* The name of the cluster group to use
+| endpoint | | String | *Required* The Camel endpoint to use in master/slave mode
+|=======================================================================
+
+#### Query Parameters (4 parameters):
+
+[width="100%",cols="2,1,1m,1m,5",options="header"]
+|=======================================================================
+| Name | Group | Default | Java Type | Description
+| bridgeErrorHandler | consumer | false | boolean | Allows for bridging the consumer to the Camel routing Error Handler which mean any exceptions occurred while the consumer is trying to pickup incoming messages or the likes will now be processed as a message and handled by the routing Error Handler. By default the consumer will use the org.apache.camel.spi.ExceptionHandler to deal with exceptions that will be logged at WARN or ERROR level and ignored.
+| exceptionHandler | consumer (advanced) | | ExceptionHandler | To let the consumer use a custom ExceptionHandler. Notice if the option bridgeErrorHandler is enabled then this options is not in use. By default the consumer will deal with exceptions that will be logged at WARN or ERROR level and ignored.
+| exchangePattern | consumer (advanced) | | ExchangePattern | Sets the exchange pattern when the consumer creates an exchange.
+| synchronous | advanced | false | boolean | Sets whether synchronous processing should be strictly used or Camel is allowed to use asynchronous processing (if supported).
+|=======================================================================
+// endpoint options: END
+
+
+### See Also
+
+* link:configuring-camel.html[Configuring Camel]
+* link:component.html[Component]
+* link:endpoint.html[Endpoint]
+* link:getting-started.html[Getting Started]
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/CamelNodeState.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/CamelNodeState.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/CamelNodeState.java
new file mode 100644
index 0000000..d501e12
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/CamelNodeState.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.camel.component.zookeepermaster;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+
+public class CamelNodeState extends NodeState {
+
+ @JsonProperty
+ String consumer;
+
+ @JsonProperty
+ boolean started;
+
+ public CamelNodeState() {
+ }
+
+ public CamelNodeState(String id) {
+ super(id);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterComponent.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterComponent.java
new file mode 100644
index 0000000..f2d5d63
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterComponent.java
@@ -0,0 +1,76 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import java.util.Map;
+
+import org.apache.camel.Endpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.URISupport;
+
+/**
+ * The zookeeper-master camel component ensures that only a single endpoint in a cluster is active at any
+ * point in time with all other JVMs being hot standbys which wait until the master JVM dies before
+ * taking over to provide high availability of a single consumer.
+ */
+public class MasterComponent extends ZKComponentSupport {
+
+ @Metadata(defaultValue = "/camel/zookeepermaster/clusters/master")
+ private String zkRoot = "/camel/zookeepermaster/clusters/master";
+
+ public String getZkRoot() {
+ return zkRoot;
+ }
+
+ /**
+ * The root path to use in zookeeper where information is stored which nodes are master/slave etc.
+ * Will by default use: /camel/zookeepermaster/clusters/master
+ */
+ public void setZkRoot(String zkRoot) {
+ this.zkRoot = zkRoot;
+ }
+
+ // Implementation methods
+ //-------------------------------------------------------------------------
+
+ @Override
+ protected Endpoint createEndpoint(String uri, String remaining, Map<String, Object> params) throws Exception {
+ int idx = remaining.indexOf(':');
+ if (idx <= 0) {
+ throw new IllegalArgumentException("Missing : in URI so cannot split the group name from the actual URI for '" + remaining + "'");
+ }
+ // we are registering a regular endpoint
+ String name = remaining.substring(0, idx);
+ String childUri = remaining.substring(idx + 1);
+ // we need to apply the params here
+ if (params != null && params.size() > 0) {
+ childUri = childUri + "?" + URISupport.createQueryString(params);
+ }
+ MasterEndpoint answer = new MasterEndpoint(uri, this, name, childUri);
+ return answer;
+ }
+
+ protected String getCamelClusterPath(String name) {
+ String path = name;
+ if (ObjectHelper.isNotEmpty(zkRoot)) {
+ path = zkRoot + "/" + name;
+ }
+ return path;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
new file mode 100644
index 0000000..ab3608c
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
@@ -0,0 +1,157 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.SuspendableService;
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.GroupListener;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.util.ServiceHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A consumer which is only really active while it holds the master lock
+ */
+public class MasterConsumer extends DefaultConsumer implements GroupListener {
+ private static final transient Logger LOG = LoggerFactory.getLogger(MasterConsumer.class);
+
+ private final MasterEndpoint endpoint;
+ private final Processor processor;
+ private Consumer delegate;
+ private SuspendableService delegateService;
+ private final Group<CamelNodeState> singleton;
+
+ public MasterConsumer(MasterEndpoint endpoint, Processor processor) {
+ super(endpoint, processor);
+ this.endpoint = endpoint;
+ this.processor = processor;
+ MasterComponent component = endpoint.getComponent();
+ String path = component.getCamelClusterPath(endpoint.getSingletonId());
+ this.singleton = component.createGroup(path);
+ this.singleton.add(this);
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ singleton.start();
+ LOG.info("Attempting to become master for endpoint: " + endpoint + " in " + endpoint.getCamelContext() + " with singletonID: " + endpoint.getSingletonId());
+ singleton.update(createNodeState());
+ }
+
+ private CamelNodeState createNodeState() {
+ CamelNodeState state = new CamelNodeState(endpoint.getSingletonId());
+ state.consumer = endpoint.getChildEndpoint().getEndpointUri();
+ return state;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ try {
+ stopConsumer();
+ } finally {
+ singleton.close();
+ }
+ }
+
+ protected void stopConsumer() throws Exception {
+ ServiceHelper.stopAndShutdownServices(delegate);
+ ServiceHelper.stopAndShutdownServices(endpoint.getChildEndpoint());
+ delegate = null;
+ delegateService = null;
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ if (delegateService != null) {
+ delegateService.resume();
+ }
+ super.doResume();
+ }
+
+ @Override
+ protected void doSuspend() throws Exception {
+ if (delegateService != null) {
+ delegateService.suspend();
+ }
+ super.doSuspend();
+ }
+
+ @Override
+ public void groupEvent(Group group, GroupEvent event) {
+ switch (event) {
+ case CONNECTED:
+ break;
+ case CHANGED:
+ if (singleton.isConnected()) {
+ if (singleton.isMaster()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Master/Standby endpoint is Master for: " + endpoint + " in " + endpoint.getCamelContext());
+ }
+ onLockOwned();
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Master/Standby endpoint is Standby for: " + endpoint + " in " + endpoint.getCamelContext());
+ }
+ }
+ }
+ break;
+ case DISCONNECTED:
+ try {
+ LOG.info("Disconnecting as Master. Stopping consumer: {}", endpoint.getChildEndpoint());
+ stopConsumer();
+ } catch (Exception e) {
+ LOG.error("Failed to stop master consumer for: " + endpoint + ". Reason: " + e, e);
+ }
+ break;
+ default:
+ // noop
+ }
+
+ }
+
+ protected void onLockOwned() {
+ if (delegate == null) {
+ try {
+ // ensure endpoint is also started
+ LOG.info("Elected as master. Starting consumer: {}", endpoint.getChildEndpoint());
+ ServiceHelper.startService(endpoint.getChildEndpoint());
+
+ delegate = endpoint.getChildEndpoint().createConsumer(processor);
+ delegateService = null;
+ if (delegate instanceof SuspendableService) {
+ delegateService = (SuspendableService) delegate;
+ }
+
+ // Lets show we are starting the consumer.
+ CamelNodeState nodeState = createNodeState();
+ nodeState.started = true;
+ singleton.update(nodeState);
+
+ ServiceHelper.startService(delegate);
+ } catch (Exception e) {
+ LOG.error("Failed to start master consumer for: " + endpoint + ". Reason: " + e, e);
+ }
+
+ LOG.info("Elected as master. Consumer started: {}", endpoint.getChildEndpoint());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterEndpoint.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterEndpoint.java
new file mode 100644
index 0000000..6e034ed
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterEndpoint.java
@@ -0,0 +1,98 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.DelegateEndpoint;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.Metadata;
+import org.apache.camel.spi.UriEndpoint;
+import org.apache.camel.spi.UriPath;
+
+/**
+ * Represents an endpoint which only becomes active when it obtains the master lock
+ */
+@UriEndpoint(firstVersion = "2.19.0", scheme = "zookeeper-master", syntax = "zookeeper-master:name:endpoint", consumerClass = MasterConsumer.class, consumerOnly = true,
+ title = "ZooKeeper Master", lenientProperties = true, label = "clustering")
+public class MasterEndpoint extends DefaultEndpoint implements DelegateEndpoint {
+
+ private final MasterComponent component;
+
+ private final Endpoint childEndpoint;
+
+ @UriPath(name = "name", description = "The name of the cluster group to use")
+ @Metadata(required = "true")
+ private final String singletonId;
+
+ @UriPath(name = "endpoint", description = "The Camel endpoint to use in master/slave mode")
+ @Metadata(required = "true")
+ private final String child;
+
+ public MasterEndpoint(String uri, MasterComponent component, String singletonId, String child) {
+ super(uri, component);
+ this.component = component;
+ this.singletonId = singletonId;
+ this.child = child;
+ this.childEndpoint = getCamelContext().getEndpoint(child);
+ }
+
+ public String getSingletonId() {
+ return singletonId;
+ }
+
+ @Override
+ public Producer createProducer() throws Exception {
+ throw new UnsupportedOperationException("Cannot produce from this endpoint");
+ }
+
+ public Consumer createConsumer(Processor processor) throws Exception {
+ return new MasterConsumer(this, processor);
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ @Override
+ public boolean isLenientProperties() {
+ // to allow properties to be propagated to the child endpoint
+ return true;
+ }
+
+ // Properties
+ //-------------------------------------------------------------------------
+ public MasterComponent getComponent() {
+ return component;
+ }
+
+ public String getChild() {
+ return child;
+ }
+
+ // Implementation methods
+ //-------------------------------------------------------------------------
+ Endpoint getChildEndpoint() {
+ return childEndpoint;
+ }
+
+ public Endpoint getEndpoint() {
+ return getChildEndpoint();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/ZKComponentSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/ZKComponentSupport.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/ZKComponentSupport.java
new file mode 100644
index 0000000..30d9e04
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/ZKComponentSupport.java
@@ -0,0 +1,185 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import java.util.concurrent.Callable;
+
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.internal.ManagedGroupFactory;
+import org.apache.camel.component.zookeepermaster.group.internal.ManagedGroupFactoryBuilder;
+import org.apache.camel.impl.DefaultComponent;
+import org.apache.camel.spi.Metadata;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.state.ConnectionState;
+import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.retry.RetryOneTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class ZKComponentSupport extends DefaultComponent implements Callable<CuratorFramework>, ConnectionStateListener {
+ private static final transient Logger LOG = LoggerFactory.getLogger(ZKComponentSupport.class);
+
+ private static final String ZOOKEEPER_URL = "zookeeper.url";
+ private static final String ZOOKEEPER_PASSWORD = "zookeeper.password";
+ private static final String ZOOKEEPER_URL_ENV = "ZOOKEEPER_URL";
+ private static final String ZOOKEEPER_HOST_ENV = "ZK_CLIENT_SERVICE_HOST";
+ private static final String ZOOKEEPER_PORT_ENV = "ZK_CLIENT_SERVICE_PORT";
+
+ private ManagedGroupFactory managedGroupFactory;
+
+ @Metadata(label = "advanced")
+ private CuratorFramework curator;
+ @Metadata(defaultValue = "10000")
+ private int maximumConnectionTimeout = 10 * 1000;
+ @Metadata(defaultValue = "localhost:2181")
+ private String zooKeeperUrl;
+ @Metadata(label = "security", secret = true)
+ private String zooKeeperPassword;
+
+ public CuratorFramework getCurator() {
+ if (managedGroupFactory == null) {
+ throw new IllegalStateException("Component is not started");
+ }
+ return managedGroupFactory.getCurator();
+ }
+
+ public Group<CamelNodeState> createGroup(String path) {
+ if (managedGroupFactory == null) {
+ throw new IllegalStateException("Component is not started");
+ }
+ return managedGroupFactory.createGroup(path, CamelNodeState.class);
+ }
+
+ /**
+ * To use a custom configured CuratorFramework as connection to zookeeper ensemble.
+ */
+ public void setCurator(CuratorFramework curator) {
+ this.curator = curator;
+ registerAsListener();
+ }
+
+ public int getMaximumConnectionTimeout() {
+ return maximumConnectionTimeout;
+ }
+
+ /**
+ * Timeout in millis to use when connecting to the zookeeper ensemble
+ */
+ public void setMaximumConnectionTimeout(int maximumConnectionTimeout) {
+ this.maximumConnectionTimeout = maximumConnectionTimeout;
+ }
+
+ public String getZooKeeperUrl() {
+ return zooKeeperUrl;
+ }
+
+ /**
+ * The url for the zookeeper ensemble
+ */
+ public void setZooKeeperUrl(String zooKeeperUrl) {
+ this.zooKeeperUrl = zooKeeperUrl;
+ }
+
+ public String getZooKeeperPassword() {
+ return zooKeeperPassword;
+ }
+
+ /**
+ * The password to use when connecting to the zookeeper ensemble
+ */
+ public void setZooKeeperPassword(String zooKeeperPassword) {
+ this.zooKeeperPassword = zooKeeperPassword;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ if (curator == null) {
+ try {
+ CuratorFramework aCurator = (CuratorFramework) getCamelContext().getRegistry().lookupByName("curator");
+ if (aCurator != null) {
+ setCurator(aCurator);
+ }
+ if (curator != null) {
+ LOG.debug("Zookeeper client found in camel registry. " + curator);
+ }
+ } catch (Exception exception) {
+ // ignore
+ }
+ }
+
+ // will auto create curator if needed
+ managedGroupFactory = ManagedGroupFactoryBuilder.create(curator, getClass().getClassLoader(), getCamelContext().getClassResolver(), this);
+ }
+
+ public CuratorFramework call() throws Exception {
+ String connectString = getZooKeeperUrl();
+ if (connectString == null) {
+ connectString = System.getenv(ZOOKEEPER_URL_ENV);
+ }
+ if (connectString == null) {
+ String zkHost = System.getenv(ZOOKEEPER_HOST_ENV);
+ if (zkHost != null) {
+ String zkPort = System.getenv(ZOOKEEPER_PORT_ENV);
+ connectString = zkHost + ":" + (zkPort == null ? "2181" : zkPort);
+ }
+ }
+ if (connectString == null) {
+ connectString = System.getProperty(ZOOKEEPER_URL, "localhost:2181");
+ }
+ String password = getZooKeeperPassword();
+ if (password == null) {
+ System.getProperty(ZOOKEEPER_PASSWORD);
+ }
+ LOG.debug("CuratorFramework not found in Camel registry, creating new with connection " + connectString);
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+ .connectString(connectString)
+ .retryPolicy(new RetryOneTime(1000))
+ .connectionTimeoutMs(getMaximumConnectionTimeout());
+
+ if (password != null && !password.isEmpty()) {
+ builder.authorization("digest", ("fabric:" + password).getBytes());
+ }
+
+ curator = builder.build();
+ LOG.debug("Starting curator " + curator);
+ curator.start();
+ return curator;
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ super.doStop();
+ if (managedGroupFactory != null) {
+ managedGroupFactory.close();
+ managedGroupFactory = null;
+ }
+ }
+
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState) {
+ LOG.info("Curator Connection new state: " + newState);
+ }
+
+ protected void registerAsListener() {
+ if (curator != null) {
+ curator.getConnectionStateListenable().addListener(this);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/Group.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/Group.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/Group.java
new file mode 100644
index 0000000..647e8c1
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/Group.java
@@ -0,0 +1,100 @@
+/**
+ * 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.camel.component.zookeepermaster.group;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+public interface Group<T extends NodeState> extends Closeable {
+
+ /**
+ * Are we connected with the cluster?
+ */
+ boolean isConnected();
+
+ /**
+ * Start this member
+ */
+ void start();
+
+ /**
+ * A member should be closed to release acquired resources used
+ * to monitor the group membership.
+ *
+ * When the member is closed, any memberships registered via this
+ * Group will be removed from the group.
+ */
+ void close() throws IOException;
+
+ /**
+ * Registers a listener which will be called
+ * when the cluster membership changes or
+ * the group is connected or disconnected.
+ */
+ void add(GroupListener<T> listener);
+
+ /**
+ * Removes a previously added listener.
+ */
+ void remove(GroupListener<T> listener);
+
+ /**
+ * Update the state of this group member.
+ * If the state is null, the member will leave the group.
+ *
+ * This method can be called even if the group is not started,
+ * in which case the state will be stored and updated
+ * when the group becomes started.
+ *
+ * @param state the new state of this group member
+ */
+ void update(T state);
+
+ /**
+ * Get the list of members connected to this group.
+ */
+ Map<String, T> members();
+
+ /**
+ * Check if we are the master.
+ */
+ boolean isMaster();
+
+ /**
+ * Retrieve the master node.
+ */
+ T master();
+
+ /**
+ * Retrieve the list of slaves.
+ */
+ List<T> slaves();
+
+ /**
+ * Gets the last state.
+ * <p/>
+ * This can be used by clients to access that last state, such as when the clients is being added
+ * as a {@link #add(GroupListener) listener} but wants to retrieve the last state to be up to date when the
+ * client is added.
+ *
+ * @return the state, or <tt>null</tt> if no last state yet.
+ */
+ T getLastState();
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupFactory.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupFactory.java
new file mode 100644
index 0000000..2298e40
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupFactory.java
@@ -0,0 +1,31 @@
+/**
+ * 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.camel.component.zookeepermaster.group;
+
+import java.util.concurrent.ThreadFactory;
+
+public interface GroupFactory {
+
+ <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz);
+
+ <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz, ThreadFactory threadFactory);
+
+ <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz);
+
+ <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz, ThreadFactory threadFactory);
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupListener.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupListener.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupListener.java
new file mode 100644
index 0000000..7d98c6e
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/GroupListener.java
@@ -0,0 +1,32 @@
+/**
+ * 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.camel.component.zookeepermaster.group;
+
+/**
+ * Callback interface used to get notifications of changes to a cluster group.
+ */
+public interface GroupListener<T extends NodeState> {
+
+ enum GroupEvent {
+ CONNECTED,
+ CHANGED,
+ DISCONNECTED
+ }
+
+ void groupEvent(Group<T> group, GroupEvent event);
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/MultiGroup.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/MultiGroup.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/MultiGroup.java
new file mode 100644
index 0000000..b7d94a5
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/MultiGroup.java
@@ -0,0 +1,26 @@
+/**
+ * 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.camel.component.zookeepermaster.group;
+
+public interface MultiGroup<T extends NodeState> extends Group<T> {
+
+ /**
+ * Check if we are the master.
+ */
+ boolean isMaster(String id);
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/NodeState.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/NodeState.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/NodeState.java
new file mode 100644
index 0000000..7188201
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/NodeState.java
@@ -0,0 +1,69 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.zookeepermaster.group;
+
+import java.util.UUID;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class NodeState {
+
+ @JsonProperty
+ public String id;
+
+ @JsonProperty
+ public final String container;
+
+ @JsonProperty
+ public String uuid; // internal use to suppress duplicates
+
+ public NodeState() {
+ this(null);
+ }
+
+ public NodeState(String id) {
+ this(id, System.getProperty("runtime.id", UUID.randomUUID().toString()));
+ }
+
+ public NodeState(String id, String container) {
+ this.id = id;
+ this.container = container;
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ public void setId(String id) {
+ this.id = id;
+ }
+
+ public String getContainer() {
+ return container;
+ }
+
+ @Override
+ public String toString() {
+ try {
+ return new ObjectMapper().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES).writeValueAsString(this);
+ } catch (Exception e) {
+ return super.toString();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ChildData.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ChildData.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ChildData.java
new file mode 100644
index 0000000..7874673
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ChildData.java
@@ -0,0 +1,129 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.util.Arrays;
+
+import org.apache.zookeeper.data.Stat;
+
+public class ChildData<T> implements Comparable<ChildData> {
+ private final String path;
+ private final Stat stat;
+ private final byte[] data;
+ private final T node;
+
+ ChildData(String path, Stat stat, byte[] data, T node) {
+ this.path = path;
+ this.stat = stat;
+ this.data = data;
+ this.node = node;
+ }
+
+ /**
+ * @inheritDoc
+ *
+ * Note: this class has a natural ordering that is inconsistent with equals.
+ */
+ @Override
+ public int compareTo(ChildData rhs) {
+ if (this == rhs) {
+ return 0;
+ }
+ if (rhs == null || getClass() != rhs.getClass()) {
+ return -1;
+ }
+
+ return path.compareTo(rhs.path);
+ }
+
+ @SuppressWarnings("RedundantIfStatement")
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ ChildData childData = (ChildData) o;
+
+ if (!Arrays.equals(data, childData.data)) {
+ return false;
+ }
+ if (path != null ? !path.equals(childData.path) : childData.path != null) {
+ return false;
+ }
+ if (stat != null ? !stat.equals(childData.stat) : childData.stat != null) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = path != null ? path.hashCode() : 0;
+ result = 31 * result + (stat != null ? stat.hashCode() : 0);
+ result = 31 * result + (data != null ? Arrays.hashCode(data) : 0);
+ return result;
+ }
+
+ /**
+ * Returns the full path of the this child
+ *
+ * @return full path
+ */
+ public String getPath() {
+ return path;
+ }
+
+ /**
+ * Returns the stat data for this child
+ *
+ * @return stat or null
+ */
+ public Stat getStat() {
+ return stat;
+ }
+
+ /**
+ * <p>Returns the node data for this child when the cache mode is set to cache data.</p>
+ *
+ * <p><b>NOTE:</b> the byte array returned is the raw reference of this instance's field. If you change
+ * the values in the array any other callers to this method will see the change.</p>
+ *
+ * @return node data or null
+ */
+ public byte[] getData() {
+ return data;
+ }
+
+ /**
+ * <p>Returns the node for this group member.</p>
+ *
+ * @return the node or null
+ */
+ public T getNode() {
+ return node;
+ }
+
+ @Override
+ public String toString() {
+ return "ChildData{path='" + path + '\'' + ", stat=" + stat + ", data=" + Arrays.toString(data) + '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/CompositeOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/CompositeOperation.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/CompositeOperation.java
new file mode 100644
index 0000000..7073264
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/CompositeOperation.java
@@ -0,0 +1,56 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Operation that aggregates several {@link Operation operations} to be performed inside single task passed
+ * to thread executor
+ */
+public class CompositeOperation implements Operation {
+
+ public static final Logger LOG = LoggerFactory.getLogger(CompositeOperation.class);
+
+ private Operation[] operations;
+
+ public CompositeOperation(Operation... operations) {
+ this.operations = operations;
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ for (Operation op : operations) {
+ try {
+ op.invoke();
+ if (Thread.currentThread().isInterrupted()) {
+ LOG.debug("Interrupting composite operation");
+ Thread.currentThread().interrupt();
+ break;
+ }
+ } catch (InterruptedException e) {
+ LOG.debug("Interrupting composite operation");
+ Thread.currentThread().interrupt();
+ break;
+ } catch (Exception e) {
+ LOG.error(e.getMessage(), e);
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperGroup.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperGroup.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperGroup.java
new file mode 100644
index 0000000..f0e3f64
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperGroup.java
@@ -0,0 +1,180 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.GroupListener;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+
+import static org.apache.curator.utils.CloseableUtils.closeQuietly;
+
+public class DelegateZooKeeperGroup<T extends NodeState> implements Group<T> {
+
+ private final String path;
+ private final Class<T> clazz;
+ private final List<GroupListener<T>> listeners;
+ private Group<T> group;
+ private T state;
+ private AtomicBoolean started = new AtomicBoolean();
+
+ public DelegateZooKeeperGroup(String path, Class<T> clazz) {
+ this.listeners = new ArrayList<GroupListener<T>>();
+ this.path = path;
+ this.clazz = clazz;
+ }
+
+ public void useCurator(CuratorFramework curator) {
+ Group<T> group = this.group;
+ if (group != null) {
+ closeQuietly(group);
+ }
+ if (curator != null) {
+ group = createGroup(curator, path, clazz);
+ group.update(state);
+ for (GroupListener<T> listener : listeners) {
+ group.add(listener);
+ }
+ if (started.get()) {
+ group.start();
+ }
+ this.group = group;
+ }
+ }
+
+ protected Group<T> createGroup(CuratorFramework client, String path, Class<T> clazz) {
+ return new ZooKeeperGroup<T>(client, path, clazz);
+ }
+
+ @Override
+ public void add(GroupListener<T> listener) {
+ listeners.add(listener);
+ Group<T> group = this.group;
+ if (group != null) {
+ group.add(listener);
+ }
+ }
+
+ @Override
+ public void remove(GroupListener<T> listener) {
+ listeners.remove(listener);
+ Group<T> group = this.group;
+ if (group != null) {
+ group.remove(listener);
+ }
+ }
+
+ @Override
+ public boolean isConnected() {
+ Group<T> group = this.group;
+ if (group != null) {
+ return group.isConnected();
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public void start() {
+ if (started.compareAndSet(false, true)) {
+ doStart();
+ }
+ }
+
+ protected void doStart() {
+ if (group != null) {
+ group.start();
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (started.compareAndSet(true, false)) {
+ doStop();
+ }
+ }
+
+ protected void doStop() throws IOException {
+ closeQuietly(group);
+ }
+
+ @Override
+ public void update(T state) {
+ this.state = state;
+ Group<T> group = this.group;
+ if (group != null) {
+ group.update(state);
+ }
+ }
+
+ @Override
+ public Map<String, T> members() {
+ Group<T> group = this.group;
+ if (group != null) {
+ return group.members();
+ } else {
+ return Collections.emptyMap();
+ }
+ }
+
+ @Override
+ public boolean isMaster() {
+ Group<T> group = this.group;
+ if (group != null) {
+ return group.isMaster();
+ } else {
+ return false;
+ }
+ }
+
+ @Override
+ public T master() {
+ Group<T> group = this.group;
+ if (group != null) {
+ return group.master();
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public List<T> slaves() {
+ Group<T> group = this.group;
+ if (group != null) {
+ return group.slaves();
+ } else {
+ return Collections.emptyList();
+ }
+ }
+
+ public Group<T> getGroup() {
+ return group;
+ }
+
+ @Override
+ public T getLastState() {
+ return group != null ? group.getLastState() : null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperMultiGroup.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperMultiGroup.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperMultiGroup.java
new file mode 100644
index 0000000..16afc2d
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/DelegateZooKeeperMultiGroup.java
@@ -0,0 +1,44 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.MultiGroup;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+
+public class DelegateZooKeeperMultiGroup<T extends NodeState> extends DelegateZooKeeperGroup<T> implements MultiGroup<T> {
+
+ public DelegateZooKeeperMultiGroup(String path, Class<T> clazz) {
+ super(path, clazz);
+ }
+
+ protected Group<T> createGroup(CuratorFramework client, String path, Class<T> clazz) {
+ return new ZooKeeperMultiGroup<T>(client, path, clazz);
+ }
+
+ @Override
+ public boolean isMaster(String id) {
+ Group<T> group = this.getGroup();
+ if (group != null) {
+ return ((MultiGroup) group).isMaster(id);
+ } else {
+ return false;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/EventOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/EventOperation.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/EventOperation.java
new file mode 100644
index 0000000..72ad142
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/EventOperation.java
@@ -0,0 +1,39 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import org.apache.camel.component.zookeepermaster.group.GroupListener;
+
+class EventOperation implements Operation {
+ private final ZooKeeperGroup cache;
+ private final GroupListener.GroupEvent event;
+
+ EventOperation(ZooKeeperGroup cache, GroupListener.GroupEvent event) {
+ this.cache = cache;
+ this.event = event;
+ }
+
+ @Override
+ public void invoke() {
+ cache.callListeners(event);
+ }
+
+ @Override
+ public String toString() {
+ return "EventOperation{event=" + event + '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/GetDataOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/GetDataOperation.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/GetDataOperation.java
new file mode 100644
index 0000000..fc7f610
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/GetDataOperation.java
@@ -0,0 +1,61 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+class GetDataOperation implements Operation {
+
+ private final ZooKeeperGroup cache;
+ private final String fullPath;
+
+ GetDataOperation(ZooKeeperGroup cache, String fullPath) {
+ this.cache = cache;
+ this.fullPath = fullPath;
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ cache.getDataAndStat(fullPath);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ GetDataOperation that = (GetDataOperation) o;
+
+ if (!fullPath.equals(that.fullPath)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return fullPath.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return "GetDataOperation{fullPath='" + fullPath + '\'' + '}';
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactory.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactory.java
new file mode 100644
index 0000000..b812209
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactory.java
@@ -0,0 +1,28 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import org.apache.camel.component.zookeepermaster.group.GroupFactory;
+import org.apache.curator.framework.CuratorFramework;
+
+public interface ManagedGroupFactory extends GroupFactory {
+
+ CuratorFramework getCurator();
+
+ void close();
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactoryBuilder.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactoryBuilder.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactoryBuilder.java
new file mode 100644
index 0000000..393c205
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ManagedGroupFactoryBuilder.java
@@ -0,0 +1,50 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.util.concurrent.Callable;
+
+import org.apache.camel.spi.ClassResolver;
+import org.apache.camel.util.IntrospectionSupport;
+import org.apache.curator.framework.CuratorFramework;
+
+public final class ManagedGroupFactoryBuilder {
+
+ private ManagedGroupFactoryBuilder() {
+ }
+
+ public static ManagedGroupFactory create(CuratorFramework curator,
+ ClassLoader loader,
+ ClassResolver resolver,
+ Callable<CuratorFramework> factory) throws Exception {
+ if (curator != null) {
+ return new StaticManagedGroupFactory(curator, false);
+ }
+ try {
+ Class<?> clazz = resolver.resolveClass("org.apache.camel.component.zookeepermaster.group.internal.osgi.OsgiManagedGroupFactory");
+ if (clazz != null) {
+ Object instance = clazz.newInstance();
+ IntrospectionSupport.setProperty(instance, "classLoader", loader);
+ return (ManagedGroupFactory) instance;
+ }
+ } catch (Throwable e) {
+ // Ignore if we'e not in OSGi
+ }
+ return new StaticManagedGroupFactory(factory.call(), true);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/Operation.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/Operation.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/Operation.java
new file mode 100644
index 0000000..d50e51a
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/Operation.java
@@ -0,0 +1,23 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+interface Operation {
+
+ void invoke() throws Exception;
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/RefreshOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/RefreshOperation.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/RefreshOperation.java
new file mode 100644
index 0000000..ab52530
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/RefreshOperation.java
@@ -0,0 +1,61 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+class RefreshOperation implements Operation {
+ private final ZooKeeperGroup cache;
+ private final ZooKeeperGroup.RefreshMode mode;
+
+ RefreshOperation(ZooKeeperGroup cache, ZooKeeperGroup.RefreshMode mode) {
+ this.cache = cache;
+ this.mode = mode;
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ cache.refresh(mode);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ RefreshOperation that = (RefreshOperation) o;
+
+ //noinspection RedundantIfStatement
+ if (mode != that.mode) {
+ return false;
+ }
+
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return mode.hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return "RefreshOperation(" + mode + "){}";
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/SequenceComparator.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/SequenceComparator.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/SequenceComparator.java
new file mode 100644
index 0000000..42b33a2
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/SequenceComparator.java
@@ -0,0 +1,27 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.util.Comparator;
+
+public class SequenceComparator implements Comparator<ChildData> {
+
+ @Override
+ public int compare(ChildData left, ChildData right) {
+ return left.getPath().compareTo(right.getPath());
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/StaticManagedGroupFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/StaticManagedGroupFactory.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/StaticManagedGroupFactory.java
new file mode 100644
index 0000000..3b91ccd
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/StaticManagedGroupFactory.java
@@ -0,0 +1,66 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+
+public class StaticManagedGroupFactory implements ManagedGroupFactory {
+
+ private final CuratorFramework curator;
+ private final boolean shouldClose;
+
+ StaticManagedGroupFactory(CuratorFramework curator, boolean shouldClose) {
+ this.curator = curator;
+ this.shouldClose = shouldClose;
+ }
+
+ @Override
+ public CuratorFramework getCurator() {
+ return curator;
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz) {
+ return new ZooKeeperGroup<T>(curator, path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ return new ZooKeeperGroup<T>(curator, path, clazz, threadFactory);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz) {
+ return new ZooKeeperMultiGroup<T>(curator, path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ return new ZooKeeperMultiGroup<T>(curator, path, clazz, threadFactory);
+ }
+
+ @Override
+ public void close() {
+ if (shouldClose) {
+ curator.close();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/UpdateOperation.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/UpdateOperation.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/UpdateOperation.java
new file mode 100644
index 0000000..687bfbf
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/UpdateOperation.java
@@ -0,0 +1,55 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+
+class UpdateOperation<T extends NodeState> implements Operation {
+ private final ZooKeeperGroup<T> cache;
+ private final T node;
+
+ UpdateOperation(ZooKeeperGroup cache, T node) {
+ this.cache = cache;
+ this.node = node;
+ }
+
+ @Override
+ public void invoke() throws Exception {
+ cache.doUpdate(node);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public int hashCode() {
+ return getClass().hashCode();
+ }
+
+ @Override
+ public String toString() {
+ return "UpdateOperation{node='" + node + '\'' + '}';
+ }
+}
[2/3] camel git commit: CAMEL-10986: camel-zookeeper-master -
Donation of the master component from fabric8 v1
Posted by da...@apache.org.
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroup.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroup.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroup.java
new file mode 100644
index 0000000..6e58f50
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroup.java
@@ -0,0 +1,671 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.GroupListener;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+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.EnsurePath;
+import org.apache.curator.utils.ZKPaths;
+import org.apache.zookeeper.CreateMode;
+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;
+
+/**
+ * <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><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 ZooKeeperGroup<T extends NodeState> implements Group<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperGroup.class);
+ private static ObjectMapper mapper = new ObjectMapper().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+
+ private final Class<T> clazz;
+ private final CuratorFramework client;
+ private final String path;
+ private final ExecutorService executorService;
+ private final EnsurePath ensurePath;
+ private final BlockingQueue<Operation> operations = new LinkedBlockingQueue<Operation>();
+ private final ListenerContainer<GroupListener<T>> listeners = new ListenerContainer<GroupListener<T>>();
+ private final ConcurrentMap<String, ChildData<T>> currentData = Maps.newConcurrentMap();
+ private final AtomicBoolean started = new AtomicBoolean();
+ private final AtomicBoolean connected = new AtomicBoolean();
+ private final SequenceComparator sequenceComparator = new SequenceComparator();
+ private final String uuid = UUID.randomUUID().toString();
+
+ private volatile String id;
+ // to help detecting whether ZK Group update failed
+ private final AtomicBoolean creating = new AtomicBoolean();
+ // flag indicating that ephemeral node could be created in registry, but exact sequence ID is uknown
+ // this status means we may have (temporary - for the period of ZK session) duplication of nodes
+ private final AtomicBoolean unstable = new AtomicBoolean();
+ private volatile T state;
+
+ private final Watcher childrenWatcher = new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getType() != Event.EventType.None) {
+ // only interested in real change events, eg no refresh on Keeper.Disconnect
+ offerOperation(new RefreshOperation(ZooKeeperGroup.this, RefreshMode.STANDARD));
+ }
+ }
+ };
+
+ private final Watcher dataWatcher = new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ try {
+ if (event.getType() == Event.EventType.NodeDeleted) {
+ remove(event.getPath());
+ } else if (event.getType() == Event.EventType.NodeDataChanged) {
+ offerOperation(new GetDataOperation(ZooKeeperGroup.this, event.getPath()));
+ }
+ } catch (Exception e) {
+ handleException(e);
+ }
+ }
+ };
+
+ private final ConnectionStateListener connectionStateListener = new ConnectionStateListener() {
+ @Override
+ public void stateChanged(CuratorFramework client, ConnectionState newState) {
+ handleStateChange(newState);
+ }
+ };
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ */
+ public ZooKeeperGroup(CuratorFramework client, String path, Class<T> clazz) {
+ this(client, path, clazz, Executors.newSingleThreadExecutor());
+ }
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param threadFactory factory to use when creating internal threads
+ */
+ public ZooKeeperGroup(CuratorFramework client, String path, Class<T> clazz, ThreadFactory threadFactory) {
+ this(client, path, clazz, Executors.newSingleThreadExecutor(threadFactory));
+ }
+
+ /**
+ * @param client the client
+ * @param path path to watch
+ * @param executorService ExecutorService to use for the ZooKeeperGroup's background thread
+ */
+ public ZooKeeperGroup(CuratorFramework client, String path, Class<T> clazz, final ExecutorService executorService) {
+ LOG.info("Creating ZK Group for path \"" + path + "\"");
+ this.client = client;
+ this.path = path;
+ this.clazz = clazz;
+ this.executorService = executorService;
+ ensurePath = client.newNamespaceAwareEnsurePath(path);
+ }
+
+ /**
+ * Start the cache. The cache is not started automatically. You must call this method.
+ */
+ public void start() {
+ LOG.info("Starting ZK Group for path \"" + path + "\"");
+ if (started.compareAndSet(false, true)) {
+ connected.set(client.getZookeeperClient().isConnected());
+
+ if (isConnected()) {
+ handleStateChange(ConnectionState.CONNECTED);
+ }
+
+ client.getConnectionStateListenable().addListener(connectionStateListener);
+ executorService.execute(new Runnable() {
+ @Override
+ public void run() {
+ mainLoop();
+ }
+ });
+ }
+ }
+
+ /**
+ * Close/end the cache
+ *
+ * @throws IOException errors
+ */
+ @Override
+ public void close() throws IOException {
+ LOG.debug(this + ".close, connected:" + connected);
+ if (started.compareAndSet(true, false)) {
+ client.getConnectionStateListenable().removeListener(connectionStateListener);
+ executorService.shutdownNow();
+ try {
+ executorService.awaitTermination(5, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ throw (IOException) new InterruptedIOException().initCause(e);
+ }
+ try {
+ doUpdate(null);
+ if (isConnected()) {
+ callListeners(GroupListener.GroupEvent.DISCONNECTED);
+ }
+ } catch (Exception e) {
+ handleException(e);
+ }
+ listeners.clear();
+ mapper.getTypeFactory().clearCache();
+ mapper = new ObjectMapper().disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
+
+ client.clearWatcherReferences(childrenWatcher);
+ client.clearWatcherReferences(dataWatcher);
+ }
+ }
+
+ @Override
+ public boolean isConnected() {
+ return connected.get();
+ }
+
+ @Override
+ public void add(GroupListener<T> listener) {
+ listeners.addListener(listener);
+ }
+
+ @Override
+ public void remove(GroupListener<T> listener) {
+ listeners.removeListener(listener);
+ }
+
+ @Override
+ public void update(T state) {
+ T oldState = this.state;
+ this.state = state;
+
+ if (started.get()) {
+ boolean update = state == null && oldState != null
+ || state != null && oldState == null
+ || !Arrays.equals(encode(state), encode(oldState));
+ if (update) {
+ offerOperation(new CompositeOperation(
+ new RefreshOperation(this, RefreshMode.FORCE_GET_DATA_AND_STAT),
+ new UpdateOperation<T>(this, state)
+ ));
+ }
+ }
+ }
+
+ protected void doUpdate(T state) throws Exception {
+ if (LOG.isTraceEnabled()) {
+ // state.toString() invokes Jackson ObjectMapper serialization
+ LOG.trace(this + " doUpdate, state:" + state + " id:" + id);
+ }
+ if (state == null) {
+ if (id != null) {
+ try {
+ if (isConnected()) {
+ client.delete().guaranteed().forPath(id);
+ unstable.set(false);
+ }
+ } catch (KeeperException.NoNodeException e) {
+ // Ignore
+ } finally {
+ id = null;
+ }
+ } else if (creating.get()) {
+ LOG.warn("Ephemeral node could be created in the registry, but ZooKeeper group didn't record its id");
+ unstable.set(true);
+ }
+ } else if (isConnected()) {
+ // We could have created the sequence, but then have crashed and our entry is already registered.
+ // However, we ignore old ephemeral nodes, and create new ones. We can have double nodes for a bit,
+ // but the old ones should be deleted by the server when session is invalidated.
+ // See: https://issues.jboss.org/browse/FABRIC-1238
+ if (id == null) {
+ id = createEphemeralNode(state);
+ } else {
+ try {
+ updateEphemeralNode(state);
+ } catch (KeeperException.NoNodeException e) {
+ id = createEphemeralNode(state);
+ }
+ }
+ }
+ }
+
+ private String createEphemeralNode(T state) throws Exception {
+ state.uuid = uuid;
+ creating.set(true);
+ String pathId = client.create().creatingParentsIfNeeded()
+ .withMode(CreateMode.EPHEMERAL_SEQUENTIAL)
+ .forPath(path + "/0", encode(state));
+ creating.set(false);
+ unstable.set(false);
+ if (LOG.isTraceEnabled()) {
+ // state.toString() invokes Jackson ObjectMapper serialization
+ LOG.trace(this + ", state:" + state + ", new ephemeralSequential path:" + pathId);
+ }
+ prunePartialState(state, pathId);
+ state.uuid = null;
+ return pathId;
+ }
+
+ private void updateEphemeralNode(T state) throws Exception {
+ state.uuid = uuid;
+ client.setData().forPath(id, encode(state));
+ state.uuid = null;
+ }
+
+ // remove ephemeral sequential nodes created on server but not visible on client
+ private void prunePartialState(final T ourState, final String pathId) throws Exception {
+ if (ourState.uuid != null) {
+ clearAndRefresh(true, true);
+ List<ChildData<T>> children = new ArrayList<ChildData<T>>(currentData.values());
+ for (ChildData<T> child : children) {
+ if (ourState.uuid.equals(child.getNode().uuid) && !child.getPath().equals(pathId)) {
+ LOG.debug("Deleting partially created znode: " + child.getPath());
+ client.delete().guaranteed().forPath(child.getPath());
+ }
+ }
+ }
+ }
+
+ @Override
+ public Map<String, T> members() {
+ List<ChildData<T>> children = getActiveChildren();
+ Collections.sort(children, sequenceComparator);
+ Map<String, T> members = new LinkedHashMap<String, T>();
+ for (ChildData<T> child : children) {
+ members.put(child.getPath(), child.getNode());
+ }
+ return members;
+ }
+
+ @Override
+ public boolean isMaster() {
+ List<ChildData<T>> children = getActiveChildren();
+ Collections.sort(children, sequenceComparator);
+ return !children.isEmpty() && children.get(0).getPath().equals(id);
+ }
+
+ @Override
+ public T master() {
+ List<ChildData<T>> children = getActiveChildren();
+ Collections.sort(children, sequenceComparator);
+ if (children.isEmpty()) {
+ return null;
+ }
+ return children.get(0).getNode();
+ }
+
+ @Override
+ public List<T> slaves() {
+ List<ChildData<T>> children = getActiveChildren();
+ Collections.sort(children, sequenceComparator);
+ List<T> slaves = new ArrayList<T>();
+ for (int i = 1; i < children.size(); i++) {
+ slaves.add(children.get(i).getNode());
+ }
+ return slaves;
+ }
+
+ /**
+ * Filter stale nodes and return only active children from the current data.
+ *
+ * @return list of active children and data
+ */
+ protected List<ChildData<T>> getActiveChildren() {
+ Map<String, ChildData<T>> filtered = new HashMap<>();
+ for (ChildData<T> child : currentData.values()) {
+ T node = child.getNode();
+ if (!filtered.containsKey(node.getContainer())
+ || filtered.get(node.getContainer()).getPath().compareTo(child.getPath()) < 0) {
+ filtered.put(node.getContainer(), child);
+ }
+ }
+ return new ArrayList<>(filtered.values());
+ }
+
+ @Override
+ public T getLastState() {
+ return this.state;
+ }
+
+ public SequenceComparator getSequenceComparator() {
+ return sequenceComparator;
+ }
+
+ /**
+ * Return the cache listenable
+ *
+ * @return listenable
+ */
+ public ListenerContainer<GroupListener<T>> getListenable() {
+ return listeners;
+ }
+
+ /**
+ * Return the current data. There are no guarantees of accuracy. This is
+ * merely the most recent view of the data. The data is returned in sorted order.
+ *
+ * @return list of children and data
+ */
+ public List<ChildData> getCurrentData() {
+ return ImmutableList.copyOf(Sets.<ChildData>newTreeSet(currentData.values()));
+ }
+
+ /**
+ * Used for testing purpose
+ */
+ void putCurrentData(String key, ChildData value) {
+ currentData.put(key, value);
+ }
+
+ /**
+ * 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 or null
+ */
+ public ChildData getCurrentData(String fullPath) {
+ return currentData.get(fullPath);
+ }
+
+ /**
+ * Clear out current data and begin a new query on the path
+ *
+ * @throws Exception errors
+ */
+ public void clearAndRefresh() throws Exception {
+ clearAndRefresh(false, false);
+ }
+
+ /**
+ * Clear out current data and begin a new query on the path
+ *
+ * @param force - whether to force clear and refresh to trigger updates
+ * @param sync - whether to run this synchronously (block current thread) or asynchronously
+ * @throws Exception errors
+ */
+ public void clearAndRefresh(boolean force, boolean sync) throws Exception {
+ RefreshMode mode = force ? RefreshMode.FORCE_GET_DATA_AND_STAT : RefreshMode.STANDARD;
+ currentData.clear();
+ if (sync) {
+ this.refresh(mode);
+ } else {
+ offerOperation(new RefreshOperation(this, mode));
+ }
+ }
+
+ /**
+ * Clears the current data without beginning a new query and without generating any events
+ * for listeners.
+ */
+ public void clear() {
+ currentData.clear();
+ }
+
+ enum RefreshMode {
+ STANDARD,
+ FORCE_GET_DATA_AND_STAT
+ }
+
+ void refresh(final RefreshMode mode) throws Exception {
+ try {
+ ensurePath.ensure(client.getZookeeperClient());
+ List<String> children = client.getChildren().usingWatcher(childrenWatcher).forPath(path);
+ Collections.sort(children, new Comparator<String>() {
+ @Override
+ public int compare(String left, String right) {
+ return left.compareTo(right);
+ }
+ });
+ processChildren(children, mode);
+ } catch (Exception e) {
+ handleException(e);
+ }
+ }
+
+ void callListeners(final GroupListener.GroupEvent event) {
+ listeners.forEach(new Function<GroupListener<T>, Void>() {
+ @Override
+ public Void apply(GroupListener<T> listener) {
+ try {
+ listener.groupEvent(ZooKeeperGroup.this, event);
+ } catch (Exception e) {
+ handleException(e);
+ }
+ return null;
+ }
+ }
+ );
+ }
+
+ void getDataAndStat(final String fullPath) throws Exception {
+ Stat stat = new Stat();
+ byte[] data = client.getData().storingStatIn(stat).usingWatcher(dataWatcher).forPath(fullPath);
+ applyNewData(fullPath, KeeperException.Code.OK.intValue(), stat, data);
+ }
+
+ /**
+ * Default behavior is just to log the exception
+ *
+ * @param e the exception
+ */
+ protected void handleException(Throwable e) {
+ if (e instanceof IllegalStateException && "Client is not started".equals(e.getMessage())) {
+ LOG.debug("", e);
+ } else {
+ LOG.error("", e);
+ }
+ }
+
+ @VisibleForTesting
+ protected void remove(String fullPath) {
+ ChildData data = currentData.remove(fullPath);
+ if (data != null) {
+ offerOperation(new EventOperation(this, GroupListener.GroupEvent.CHANGED));
+ }
+ }
+
+ private void internalRebuildNode(String fullPath) throws Exception {
+ try {
+ Stat stat = new Stat();
+ byte[] bytes = client.getData().storingStatIn(stat).forPath(fullPath);
+ currentData.put(fullPath, new ChildData<T>(fullPath, stat, bytes, decode(bytes)));
+ } catch (KeeperException.NoNodeException ignore) {
+ // node no longer exists - remove it
+ currentData.remove(fullPath);
+ }
+ }
+
+ private void handleStateChange(ConnectionState newState) {
+ switch (newState) {
+ case SUSPENDED:
+ case LOST: {
+ connected.set(false);
+ clear();
+ EventOperation op = new EventOperation(this, GroupListener.GroupEvent.DISCONNECTED);
+ op.invoke();
+ break;
+ }
+
+ case CONNECTED:
+ case RECONNECTED: {
+ connected.set(true);
+ offerOperation(new CompositeOperation(
+ new RefreshOperation(this, RefreshMode.FORCE_GET_DATA_AND_STAT),
+ new UpdateOperation<T>(this, state),
+ new EventOperation(this, GroupListener.GroupEvent.CONNECTED)
+ ));
+ break;
+ }
+ default:
+ // noop
+ }
+ }
+
+ private void processChildren(List<String> children, RefreshMode mode) throws Exception {
+ List<String> fullPaths = Lists.newArrayList(Lists.transform(
+ children,
+ new Function<String, String>() {
+ @Override
+ public String apply(String child) {
+ return ZKPaths.makePath(path, child);
+ }
+ }
+ ));
+ Set<String> removedNodes = Sets.newHashSet(currentData.keySet());
+ removedNodes.removeAll(fullPaths);
+
+ for (String fullPath : removedNodes) {
+ remove(fullPath);
+ }
+
+ for (String name : children) {
+ String fullPath = ZKPaths.makePath(path, name);
+
+ if ((mode == RefreshMode.FORCE_GET_DATA_AND_STAT) || !currentData.containsKey(fullPath)) {
+ try {
+ getDataAndStat(fullPath);
+ } catch (KeeperException.NoNodeException ignore) {
+ }
+ }
+ }
+ }
+
+ private void applyNewData(String fullPath, int resultCode, Stat stat, byte[] bytes) {
+ if (resultCode == KeeperException.Code.OK.intValue()) {
+ // otherwise - node must have dropped or something - we should be getting another event
+ ChildData<T> data = new ChildData<T>(fullPath, stat, bytes, decode(bytes));
+ ChildData<T> previousData = currentData.put(fullPath, data);
+ if (previousData == null || previousData.getStat().getVersion() != stat.getVersion()) {
+ offerOperation(new EventOperation(this, GroupListener.GroupEvent.CHANGED));
+ }
+ }
+ }
+
+ private void mainLoop() {
+ while (started.get() && !Thread.currentThread().isInterrupted()) {
+ try {
+ operations.take().invoke();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ break;
+ } catch (Exception e) {
+ handleException(e);
+ }
+ }
+ }
+
+ private byte[] encode(T state) {
+ try {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ mapper.writeValue(baos, state);
+ return baos.toByteArray();
+ } catch (IOException e) {
+ throw new IllegalStateException("Unable to decode data", e);
+ }
+ }
+
+ private T decode(byte[] data) {
+ try {
+ return mapper.readValue(data, clazz);
+ } catch (IOException e) {
+ throw new IllegalStateException("Unable to decode data", e);
+ }
+ }
+
+ private void offerOperation(Operation operation) {
+ if (!operations.contains(operation)) {
+ operations.offer(operation);
+ }
+// operations.remove(operation); // avoids herding for refresh operations
+ }
+
+ public static <T> Map<String, T> members(ObjectMapper mapper, CuratorFramework curator, String path, Class<T> clazz) throws Exception {
+ Map<String, T> map = new TreeMap<String, T>();
+ List<String> nodes = curator.getChildren().forPath(path);
+ for (String node : nodes) {
+ byte[] data = curator.getData().forPath(path + "/" + node);
+ T val = mapper.readValue(data, clazz);
+ map.put(node, val);
+ }
+ return map;
+ }
+
+ public String getId() {
+ return id;
+ }
+
+ @VisibleForTesting
+ void setId(String id) {
+ this.id = id;
+ }
+
+ /**
+ * Returns an indication that the sequential, ephemeral node may be registered more than once for this group
+ * @return
+ */
+ public boolean isUnstable() {
+ return unstable.get();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupFactory.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupFactory.java
new file mode 100644
index 0000000..bcf2e70
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperGroupFactory.java
@@ -0,0 +1,56 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.GroupFactory;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+
+/**
+ */
+public class ZooKeeperGroupFactory implements GroupFactory {
+
+ private CuratorFramework curator;
+
+ public ZooKeeperGroupFactory(CuratorFramework curator) {
+ this.curator = curator;
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz) {
+ return new ZooKeeperGroup<T>(curator, path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ return new ZooKeeperGroup<T>(curator, path, clazz, threadFactory);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz) {
+ return new ZooKeeperMultiGroup<T>(curator, path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ return new ZooKeeperMultiGroup<T>(curator, path, clazz, threadFactory);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperMultiGroup.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperMultiGroup.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperMultiGroup.java
new file mode 100644
index 0000000..5b53220
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/ZooKeeperMultiGroup.java
@@ -0,0 +1,58 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.camel.component.zookeepermaster.group.MultiGroup;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.curator.framework.CuratorFramework;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ZooKeeperMultiGroup<T extends NodeState> extends ZooKeeperGroup<T> implements MultiGroup<T> {
+ private final Logger log = LoggerFactory.getLogger(getClass());
+
+ public ZooKeeperMultiGroup(CuratorFramework client, String path, Class<T> clazz) {
+ super(client, path, clazz);
+ }
+
+ public ZooKeeperMultiGroup(CuratorFramework client, String path, Class<T> clazz, ExecutorService executorService) {
+ super(client, path, clazz, executorService);
+ }
+
+ public ZooKeeperMultiGroup(CuratorFramework client, String path, Class<T> clazz, ThreadFactory threadFactory) {
+ super(client, path, clazz, threadFactory);
+ }
+
+ @Override
+ public boolean isMaster(String name) {
+ List<ChildData<T>> children = getActiveChildren();
+ Collections.sort(children, getSequenceComparator());
+ for (ChildData child : children) {
+ NodeState node = (NodeState) child.getNode();
+ if (node.id.equals(name)) {
+ return child.getPath().equals(getId());
+ }
+ }
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/OsgiManagedGroupFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/OsgiManagedGroupFactory.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/OsgiManagedGroupFactory.java
new file mode 100644
index 0000000..2507225
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/OsgiManagedGroupFactory.java
@@ -0,0 +1,183 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal.osgi;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.camel.component.zookeepermaster.group.internal.DelegateZooKeeperGroup;
+import org.apache.camel.component.zookeepermaster.group.internal.DelegateZooKeeperMultiGroup;
+import org.apache.camel.component.zookeepermaster.group.internal.ManagedGroupFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.BundleReference;
+import org.osgi.framework.ServiceReference;
+import org.osgi.util.tracker.ServiceTracker;
+import org.osgi.util.tracker.ServiceTrackerCustomizer;
+
+public class OsgiManagedGroupFactory implements ManagedGroupFactory {
+
+ private ManagedGroupFactory delegate;
+
+ public OsgiManagedGroupFactory() {
+ }
+
+ public void setClassLoader(ClassLoader loader) {
+ this.delegate = new OsgiTrackingManagedGroupFactory(loader);
+ }
+
+ @Override
+ public CuratorFramework getCurator() {
+ return delegate.getCurator();
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz) {
+ return delegate.createGroup(path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ return delegate.createGroup(path, clazz, threadFactory);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz) {
+ throw new IllegalStateException("not supported");
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ throw new IllegalStateException("not supported");
+ }
+
+ @Override
+ public void close() {
+ delegate.close();
+ }
+
+ static class OsgiTrackingManagedGroupFactory implements ManagedGroupFactory, ServiceTrackerCustomizer<CuratorFramework, CuratorFramework> {
+
+ private final BundleContext bundleContext;
+ private final ServiceTracker<CuratorFramework, CuratorFramework> tracker;
+ private CuratorFramework curator;
+ private final List<DelegateZooKeeperGroup<?>> groups = new ArrayList<DelegateZooKeeperGroup<?>>();
+
+ OsgiTrackingManagedGroupFactory(ClassLoader loader) {
+ this(getBundleContext(loader));
+ }
+
+ OsgiTrackingManagedGroupFactory(BundleContext bundleContext) {
+ this.bundleContext = bundleContext;
+ this.tracker = new ServiceTracker<CuratorFramework, CuratorFramework>(
+ bundleContext, CuratorFramework.class, this);
+ this.tracker.open();
+ }
+
+ static BundleContext getBundleContext(ClassLoader loader) {
+ if (!(loader instanceof BundleReference)) {
+ throw new IllegalStateException("Not an OSGi ClassLoader");
+ }
+ return ((BundleReference) loader).getBundle().getBundleContext();
+ }
+
+ @Override
+ public CuratorFramework addingService(ServiceReference<CuratorFramework> reference) {
+ CuratorFramework curator = OsgiTrackingManagedGroupFactory.this.bundleContext.getService(reference);
+ useCurator(curator);
+ return curator;
+ }
+
+ @Override
+ public void modifiedService(ServiceReference<CuratorFramework> reference, CuratorFramework service) {
+ }
+
+ @Override
+ public void removedService(ServiceReference<CuratorFramework> reference, CuratorFramework service) {
+ useCurator(null);
+ OsgiTrackingManagedGroupFactory.this.bundleContext.ungetService(reference);
+ }
+
+ protected void useCurator(CuratorFramework curator) {
+ this.curator = curator;
+ for (DelegateZooKeeperGroup<?> group : groups) {
+ group.useCurator(curator);
+ }
+ }
+
+ @Override
+ public CuratorFramework getCurator() {
+ return curator;
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz) {
+ return new DelegateZooKeeperGroup<T>(path, clazz) {
+ @Override
+ public void start() {
+ useCurator(curator);
+ groups.add(this);
+ super.start();
+ }
+
+ @Override
+ public void close() throws IOException {
+ groups.remove(this);
+ super.close();
+ }
+ };
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ throw new IllegalStateException("not supported");
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz) {
+ return new DelegateZooKeeperMultiGroup<T>(path, clazz) {
+ @Override
+ public void start() {
+ useCurator(curator);
+ groups.add(this);
+ super.start();
+ }
+
+ @Override
+ public void close() throws IOException {
+ groups.remove(this);
+ super.close();
+ }
+ };
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path, Class<T> clazz, ThreadFactory threadFactory) {
+ throw new IllegalStateException("not supported");
+ }
+
+ @Override
+ public void close() {
+ this.tracker.close();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/TrackingZooKeeperGroup.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/TrackingZooKeeperGroup.java b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/TrackingZooKeeperGroup.java
new file mode 100644
index 0000000..26df150
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/group/internal/osgi/TrackingZooKeeperGroup.java
@@ -0,0 +1,65 @@
+/**
+ * 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.camel.component.zookeepermaster.group.internal.osgi;
+
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.camel.component.zookeepermaster.group.internal.DelegateZooKeeperGroup;
+import org.apache.curator.framework.CuratorFramework;
+import org.osgi.framework.BundleContext;
+import org.osgi.framework.ServiceReference;
+import org.osgi.util.tracker.ServiceTracker;
+import org.osgi.util.tracker.ServiceTrackerCustomizer;
+
+public class TrackingZooKeeperGroup<T extends NodeState> extends DelegateZooKeeperGroup<T> {
+
+ private final BundleContext bundleContext;
+ private final ServiceTracker<CuratorFramework, CuratorFramework> tracker;
+
+ public TrackingZooKeeperGroup(BundleContext bundleContext, String path, Class<T> clazz) {
+ super(path, clazz);
+ this.bundleContext = bundleContext;
+ this.tracker = new ServiceTracker<CuratorFramework, CuratorFramework>(bundleContext, CuratorFramework.class, new ServiceTrackerCustomizer<CuratorFramework, CuratorFramework>() {
+ @Override
+ public CuratorFramework addingService(ServiceReference<CuratorFramework> reference) {
+ CuratorFramework curator = TrackingZooKeeperGroup.this.bundleContext.getService(reference);
+ useCurator(curator);
+ return curator;
+ }
+
+ @Override
+ public void modifiedService(ServiceReference<CuratorFramework> reference, CuratorFramework service) {
+ }
+
+ @Override
+ public void removedService(ServiceReference<CuratorFramework> reference, CuratorFramework service) {
+ useCurator(null);
+ TrackingZooKeeperGroup.this.bundleContext.ungetService(reference);
+ }
+ });
+ }
+
+ @Override
+ protected void doStart() {
+ tracker.open();
+ }
+
+ @Override
+ protected void doStop() {
+ tracker.close();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/resources/META-INF/LICENSE.txt
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/resources/META-INF/LICENSE.txt b/components/camel-zookeeper-master/src/main/resources/META-INF/LICENSE.txt
new file mode 100644
index 0000000..6b0b127
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/resources/META-INF/LICENSE.txt
@@ -0,0 +1,203 @@
+
+ Apache License
+ Version 2.0, January 2004
+ http://www.apache.org/licenses/
+
+ TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+ 1. Definitions.
+
+ "License" shall mean the terms and conditions for use, reproduction,
+ and distribution as defined by Sections 1 through 9 of this document.
+
+ "Licensor" shall mean the copyright owner or entity authorized by
+ the copyright owner that is granting the License.
+
+ "Legal Entity" shall mean the union of the acting entity and all
+ other entities that control, are controlled by, or are under common
+ control with that entity. For the purposes of this definition,
+ "control" means (i) the power, direct or indirect, to cause the
+ direction or management of such entity, whether by contract or
+ otherwise, or (ii) ownership of fifty percent (50%) or more of the
+ outstanding shares, or (iii) beneficial ownership of such entity.
+
+ "You" (or "Your") shall mean an individual or Legal Entity
+ exercising permissions granted by this License.
+
+ "Source" form shall mean the preferred form for making modifications,
+ including but not limited to software source code, documentation
+ source, and configuration files.
+
+ "Object" form shall mean any form resulting from mechanical
+ transformation or translation of a Source form, including but
+ not limited to compiled object code, generated documentation,
+ and conversions to other media types.
+
+ "Work" shall mean the work of authorship, whether in Source or
+ Object form, made available under the License, as indicated by a
+ copyright notice that is included in or attached to the work
+ (an example is provided in the Appendix below).
+
+ "Derivative Works" shall mean any work, whether in Source or Object
+ form, that is based on (or derived from) the Work and for which the
+ editorial revisions, annotations, elaborations, or other modifications
+ represent, as a whole, an original work of authorship. For the purposes
+ of this License, Derivative Works shall not include works that remain
+ separable from, or merely link (or bind by name) to the interfaces of,
+ the Work and Derivative Works thereof.
+
+ "Contribution" shall mean any work of authorship, including
+ the original version of the Work and any modifications or additions
+ to that Work or Derivative Works thereof, that is intentionally
+ submitted to Licensor for inclusion in the Work by the copyright owner
+ or by an individual or Legal Entity authorized to submit on behalf of
+ the copyright owner. For the purposes of this definition, "submitted"
+ means any form of electronic, verbal, or written communication sent
+ to the Licensor or its representatives, including but not limited to
+ communication on electronic mailing lists, source code control systems,
+ and issue tracking systems that are managed by, or on behalf of, the
+ Licensor for the purpose of discussing and improving the Work, but
+ excluding communication that is conspicuously marked or otherwise
+ designated in writing by the copyright owner as "Not a Contribution."
+
+ "Contributor" shall mean Licensor and any individual or Legal Entity
+ on behalf of whom a Contribution has been received by Licensor and
+ subsequently incorporated within the Work.
+
+ 2. Grant of Copyright License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ copyright license to reproduce, prepare Derivative Works of,
+ publicly display, publicly perform, sublicense, and distribute the
+ Work and such Derivative Works in Source or Object form.
+
+ 3. Grant of Patent License. Subject to the terms and conditions of
+ this License, each Contributor hereby grants to You a perpetual,
+ worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+ (except as stated in this section) patent license to make, have made,
+ use, offer to sell, sell, import, and otherwise transfer the Work,
+ where such license applies only to those patent claims licensable
+ by such Contributor that are necessarily infringed by their
+ Contribution(s) alone or by combination of their Contribution(s)
+ with the Work to which such Contribution(s) was submitted. If You
+ institute patent litigation against any entity (including a
+ cross-claim or counterclaim in a lawsuit) alleging that the Work
+ or a Contribution incorporated within the Work constitutes direct
+ or contributory patent infringement, then any patent licenses
+ granted to You under this License for that Work shall terminate
+ as of the date such litigation is filed.
+
+ 4. Redistribution. You may reproduce and distribute copies of the
+ Work or Derivative Works thereof in any medium, with or without
+ modifications, and in Source or Object form, provided that You
+ meet the following conditions:
+
+ (a) You must give any other recipients of the Work or
+ Derivative Works a copy of this License; and
+
+ (b) You must cause any modified files to carry prominent notices
+ stating that You changed the files; and
+
+ (c) You must retain, in the Source form of any Derivative Works
+ that You distribute, all copyright, patent, trademark, and
+ attribution notices from the Source form of the Work,
+ excluding those notices that do not pertain to any part of
+ the Derivative Works; and
+
+ (d) If the Work includes a "NOTICE" text file as part of its
+ distribution, then any Derivative Works that You distribute must
+ include a readable copy of the attribution notices contained
+ within such NOTICE file, excluding those notices that do not
+ pertain to any part of the Derivative Works, in at least one
+ of the following places: within a NOTICE text file distributed
+ as part of the Derivative Works; within the Source form or
+ documentation, if provided along with the Derivative Works; or,
+ within a display generated by the Derivative Works, if and
+ wherever such third-party notices normally appear. The contents
+ of the NOTICE file are for informational purposes only and
+ do not modify the License. You may add Your own attribution
+ notices within Derivative Works that You distribute, alongside
+ or as an addendum to the NOTICE text from the Work, provided
+ that such additional attribution notices cannot be construed
+ as modifying the License.
+
+ You may add Your own copyright statement to Your modifications and
+ may provide additional or different license terms and conditions
+ for use, reproduction, or distribution of Your modifications, or
+ for any such Derivative Works as a whole, provided Your use,
+ reproduction, and distribution of the Work otherwise complies with
+ the conditions stated in this License.
+
+ 5. Submission of Contributions. Unless You explicitly state otherwise,
+ any Contribution intentionally submitted for inclusion in the Work
+ by You to the Licensor shall be under the terms and conditions of
+ this License, without any additional terms or conditions.
+ Notwithstanding the above, nothing herein shall supersede or modify
+ the terms of any separate license agreement you may have executed
+ with Licensor regarding such Contributions.
+
+ 6. Trademarks. This License does not grant permission to use the trade
+ names, trademarks, service marks, or product names of the Licensor,
+ except as required for reasonable and customary use in describing the
+ origin of the Work and reproducing the content of the NOTICE file.
+
+ 7. Disclaimer of Warranty. Unless required by applicable law or
+ agreed to in writing, Licensor provides the Work (and each
+ Contributor provides its Contributions) on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ implied, including, without limitation, any warranties or conditions
+ of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+ PARTICULAR PURPOSE. You are solely responsible for determining the
+ appropriateness of using or redistributing the Work and assume any
+ risks associated with Your exercise of permissions under this License.
+
+ 8. Limitation of Liability. In no event and under no legal theory,
+ whether in tort (including negligence), contract, or otherwise,
+ unless required by applicable law (such as deliberate and grossly
+ negligent acts) or agreed to in writing, shall any Contributor be
+ liable to You for damages, including any direct, indirect, special,
+ incidental, or consequential damages of any character arising as a
+ result of this License or out of the use or inability to use the
+ Work (including but not limited to damages for loss of goodwill,
+ work stoppage, computer failure or malfunction, or any and all
+ other commercial damages or losses), even if such Contributor
+ has been advised of the possibility of such damages.
+
+ 9. Accepting Warranty or Additional Liability. While redistributing
+ the Work or Derivative Works thereof, You may choose to offer,
+ and charge a fee for, acceptance of support, warranty, indemnity,
+ or other liability obligations and/or rights consistent with this
+ License. However, in accepting such obligations, You may act only
+ on Your own behalf and on Your sole responsibility, not on behalf
+ of any other Contributor, and only if You agree to indemnify,
+ defend, and hold each Contributor harmless for any liability
+ incurred by, or claims asserted against, such Contributor by reason
+ of your accepting any such warranty or additional liability.
+
+ END OF TERMS AND CONDITIONS
+
+ APPENDIX: How to apply the Apache License to your work.
+
+ To apply the Apache License to your work, attach the following
+ boilerplate notice, with the fields enclosed by brackets "[]"
+ replaced with your own identifying information. (Don't include
+ the brackets!) The text should be enclosed in the appropriate
+ comment syntax for the file format. We also recommend that a
+ file or class name and description of purpose be included on the
+ same "printed page" as the copyright notice for easier
+ identification within third-party archives.
+
+ Copyright [yyyy] [name of copyright owner]
+
+ 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.
+
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/resources/META-INF/NOTICE.txt
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/resources/META-INF/NOTICE.txt b/components/camel-zookeeper-master/src/main/resources/META-INF/NOTICE.txt
new file mode 100644
index 0000000..2e215bf
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/resources/META-INF/NOTICE.txt
@@ -0,0 +1,11 @@
+ =========================================================================
+ == NOTICE file corresponding to the section 4 d of ==
+ == the Apache License, Version 2.0, ==
+ == in this case for the Apache Camel distribution. ==
+ =========================================================================
+
+ This product includes software developed by
+ The Apache Software Foundation (http://www.apache.org/).
+
+ Please read the different LICENSE files present in the licenses directory of
+ this distribution.
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/main/resources/META-INF/services/org/apache/camel/component/zookeeper-master
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/main/resources/META-INF/services/org/apache/camel/component/zookeeper-master b/components/camel-zookeeper-master/src/main/resources/META-INF/services/org/apache/camel/component/zookeeper-master
new file mode 100644
index 0000000..7cec288
--- /dev/null
+++ b/components/camel-zookeeper-master/src/main/resources/META-INF/services/org/apache/camel/component/zookeeper-master
@@ -0,0 +1,17 @@
+#
+# 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.
+#
+class=org.apache.camel.component.zookeepermaster.MasterComponent
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/CuratorFactoryBean.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/CuratorFactoryBean.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/CuratorFactoryBean.java
new file mode 100644
index 0000000..6af5f26
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/CuratorFactoryBean.java
@@ -0,0 +1,88 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.FactoryBean;
+
+/**
+ * Spring {@link FactoryBean} to make using {@link CuratorFramework} easier to setup
+ * in Spring XML files also.
+ */
+public class CuratorFactoryBean implements FactoryBean<CuratorFramework>, DisposableBean {
+
+ private static final transient Log LOG = LogFactory.getLog(CuratorFactoryBean.class);
+
+ private String connectString = "localhost:2181";
+ private int timeout = 30000;
+ private CuratorFramework curator;
+
+ public String getConnectString() {
+ return connectString;
+ }
+
+ public void setConnectString(String connectString) {
+ this.connectString = connectString;
+ }
+
+ public int getTimeout() {
+ return timeout;
+ }
+
+ public void setTimeout(int timeout) {
+ this.timeout = timeout;
+ }
+
+ // FactoryBean interface
+ //-------------------------------------------------------------------------
+ public CuratorFramework getObject() throws Exception {
+ LOG.debug("Connecting to ZooKeeper on " + connectString);
+
+ CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
+ .connectString(connectString)
+ .retryPolicy(new ExponentialBackoffRetry(5, 10))
+ .connectionTimeoutMs(getTimeout());
+
+ this.curator = builder.build();
+ LOG.debug("Starting curator " + curator);
+ curator.start();
+ return curator;
+ }
+
+ public Class<?> getObjectType() {
+ return CuratorFramework.class;
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public void destroy() throws Exception {
+ if (curator != null) {
+ // Note we cannot use zkClient.close()
+ // since you cannot currently close a client which is not connected
+ curator.close();
+ curator = null;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointFailoverTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointFailoverTest.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointFailoverTest.java
new file mode 100644
index 0000000..887bef1
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointFailoverTest.java
@@ -0,0 +1,147 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.impl.SimpleRegistry;
+import org.apache.camel.util.ServiceHelper;
+import org.apache.curator.framework.CuratorFramework;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MasterEndpointFailoverTest {
+ private static final transient Logger LOG = LoggerFactory.getLogger(MasterEndpointFailoverTest.class);
+
+ protected ProducerTemplate template;
+ protected CamelContext producerContext;
+ protected CamelContext consumerContext1;
+ protected CamelContext consumerContext2;
+ protected MockEndpoint result1Endpoint;
+ protected MockEndpoint result2Endpoint;
+ protected AtomicInteger messageCounter = new AtomicInteger(1);
+ protected ZKServerFactoryBean serverFactoryBean = new ZKServerFactoryBean();
+ protected CuratorFactoryBean zkClientBean = new CuratorFactoryBean();
+
+ @Before
+ public void beforeRun() throws Exception {
+ System.out.println("Starting ZK server!");
+ serverFactoryBean.setPort(9004);
+ serverFactoryBean.afterPropertiesSet();
+
+ // Create the zkClientBean
+ zkClientBean.setConnectString("localhost:9004");
+ CuratorFramework client = zkClientBean.getObject();
+
+ // Need to bind the zookeeper client with the name "curator"
+ SimpleRegistry registry = new SimpleRegistry();
+ registry.put("curator", client);
+
+ producerContext = new DefaultCamelContext(registry);
+ // Add the vm:start endpoint to avoid the NPE before starting the consumerContext1
+ producerContext.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start").to("vm:start");
+ }
+ });
+
+ template = producerContext.createProducerTemplate();
+
+ consumerContext1 = new DefaultCamelContext(registry);
+ consumerContext1.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("zookeeper-master:MasterEndpointFailoverTest:vm:start")
+ .to("log:result1")
+ .to("mock:result1");
+ }
+ });
+ consumerContext2 = new DefaultCamelContext(registry);
+ consumerContext2.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("zookeeper-master:MasterEndpointFailoverTest:vm:start")
+ .to("log:result2")
+ .to("mock:result2");
+ }
+ });
+ // Need to start at less one consumerContext to enable the vm queue for producerContext
+ ServiceHelper.startServices(consumerContext1);
+ ServiceHelper.startServices(producerContext);
+ result1Endpoint = consumerContext1.getEndpoint("mock:result1", MockEndpoint.class);
+ result2Endpoint = consumerContext2.getEndpoint("mock:result2", MockEndpoint.class);
+ }
+
+ @After
+ public void afterRun() throws Exception {
+ ServiceHelper.stopServices(consumerContext1);
+ ServiceHelper.stopServices(consumerContext2);
+ ServiceHelper.stopServices(producerContext);
+ zkClientBean.destroy();
+ serverFactoryBean.destroy();
+ }
+
+ @Test
+ public void testEndpoint() throws Exception {
+ System.out.println("Starting consumerContext1");
+
+ ServiceHelper.startServices(consumerContext1);
+ assertMessageReceived(result1Endpoint, result2Endpoint);
+
+ System.out.println("Starting consumerContext2");
+ ServiceHelper.startServices(consumerContext2);
+ assertMessageReceivedLoop(result1Endpoint, result2Endpoint, 3);
+
+ System.out.println("Stopping consumerContext1");
+ ServiceHelper.stopService(consumerContext1);
+ assertMessageReceivedLoop(result2Endpoint, result1Endpoint, 3);
+ }
+
+ protected void assertMessageReceivedLoop(MockEndpoint masterEndpoint, MockEndpoint standbyEndpoint, int count) throws Exception {
+ for (int i = 0; i < count; i++) {
+ Thread.sleep(1000);
+ assertMessageReceived(masterEndpoint, standbyEndpoint);
+ }
+ }
+
+ protected void assertMessageReceived(MockEndpoint masterEndpoint, MockEndpoint standbyEndpoint) throws InterruptedException {
+ masterEndpoint.reset();
+ standbyEndpoint.reset();
+
+ String expectedBody = createNextExpectedBody();
+ masterEndpoint.expectedBodiesReceived(expectedBody);
+ standbyEndpoint.expectedMessageCount(0);
+
+ template.sendBody("direct:start", expectedBody);
+
+ LOG.info("Expecting master: " + masterEndpoint + " and standby: " + standbyEndpoint);
+ MockEndpoint.assertIsSatisfied(masterEndpoint, standbyEndpoint);
+ }
+
+ protected String createNextExpectedBody() {
+ return "body:" + messageCounter.incrementAndGet();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointTest.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointTest.java
new file mode 100644
index 0000000..b287428
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterEndpointTest.java
@@ -0,0 +1,100 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import java.util.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.Produce;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.Route;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.util.ServiceHelper;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
+
+import static org.junit.Assert.assertEquals;
+
+@ContextConfiguration
+public class MasterEndpointTest extends AbstractJUnit4SpringContextTests {
+
+ protected static ZKServerFactoryBean lastServerBean;
+
+ protected static CuratorFactoryBean lastClientBean;
+
+ @Autowired
+ protected CamelContext camelContext;
+
+ @EndpointInject(uri = "mock:results")
+ protected MockEndpoint resultEndpoint;
+
+ @Produce(uri = "seda:bar")
+ protected ProducerTemplate template;
+
+ // Yeah this sucks.. why does the spring context not get shutdown
+ // after each test case? Not sure!
+ @Autowired
+ protected ZKServerFactoryBean zkServerBean;
+
+ @Autowired
+ protected CuratorFactoryBean zkClientBean;
+
+ @Before
+ public void startService() throws Exception {
+ ServiceHelper.startService(camelContext);
+ ServiceHelper.startService(template);
+ }
+
+ @After
+ public void afterRun() throws Exception {
+ lastServerBean = zkServerBean;
+ lastClientBean = zkClientBean;
+ ServiceHelper.stopServices(camelContext);
+ }
+
+ @AfterClass
+ public static void shutDownZK() throws Exception {
+ lastClientBean.destroy();
+ lastServerBean.destroy();
+ }
+
+ @Test
+ public void testEndpoint() throws Exception {
+ // check the endpoint configuration
+ List<Route> registeredRoutes = camelContext.getRoutes();
+ assertEquals("number of routes", 1, registeredRoutes.size());
+ MasterEndpoint endpoint = (MasterEndpoint) registeredRoutes.get(0).getEndpoint();
+ assertEquals("wrong endpoint uri", "seda:bar", endpoint.getChild());
+
+ String expectedBody = "<matched/>";
+
+ resultEndpoint.expectedBodiesReceived(expectedBody);
+
+ // lets wait for the entry to be registered...
+ Thread.sleep(5000);
+
+ template.sendBodyAndHeader(expectedBody, "foo", "bar");
+
+ MockEndpoint.assertIsSatisfied(camelContext);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest.java
new file mode 100644
index 0000000..f4c77a8
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterQuartz2EndpointTest.java
@@ -0,0 +1,76 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.util.ServiceHelper;
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
+
+@ContextConfiguration
+public class MasterQuartz2EndpointTest extends AbstractJUnit4SpringContextTests {
+
+ protected static ZKServerFactoryBean lastServerBean;
+
+ protected static CuratorFactoryBean lastClientBean;
+
+ @Autowired
+ protected CamelContext camelContext;
+
+ @EndpointInject(uri = "mock:results")
+ protected MockEndpoint resultEndpoint;
+
+ // Yeah this sucks.. why does the spring context not get shutdown
+ // after each test case? Not sure!
+ @Autowired
+ protected ZKServerFactoryBean zkServerBean;
+
+ @Autowired
+ protected CuratorFactoryBean zkClientBean;
+
+ @Before
+ public void startService() throws Exception {
+ ServiceHelper.startService(camelContext);
+ }
+
+ @After
+ public void afterRun() throws Exception {
+ lastServerBean = zkServerBean;
+ lastClientBean = zkClientBean;
+ ServiceHelper.stopServices(camelContext);
+ }
+
+ @AfterClass
+ public static void shutDownZK() throws Exception {
+ lastClientBean.destroy();
+ lastServerBean.destroy();
+ }
+
+ @Test
+ public void testEndpoint() throws Exception {
+ resultEndpoint.expectedMinimumMessageCount(2);
+
+ MockEndpoint.assertIsSatisfied(camelContext);
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/bf82a5ed/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/ZKServerFactoryBean.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/ZKServerFactoryBean.java b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/ZKServerFactoryBean.java
new file mode 100644
index 0000000..0b0fe6b
--- /dev/null
+++ b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/ZKServerFactoryBean.java
@@ -0,0 +1,225 @@
+/**
+ * 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.camel.component.zookeepermaster;
+
+import java.io.File;
+import java.net.InetSocketAddress;
+
+import org.apache.zookeeper.server.NIOServerCnxnFactory;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.FactoryBean;
+import org.springframework.beans.factory.InitializingBean;
+
+/**
+ * A simple ZK server for testing ZK related code in unit tests
+ */
+public class ZKServerFactoryBean implements FactoryBean<ZooKeeperServer>, InitializingBean, DisposableBean {
+ private ZooKeeperServer zooKeeperServer = new ZooKeeperServer();
+ private NIOServerCnxnFactory connectionFactory;
+ private File dataLogDir;
+ private File dataDir;
+ private boolean purge;
+ private int tickTime = ZooKeeperServer.DEFAULT_TICK_TIME;
+
+ /**
+ * defaults to -1 if not set explicitly
+ */
+ private int minSessionTimeout = -1;
+
+ /**
+ * defaults to -1 if not set explicitly
+ */
+ private int maxSessionTimeout = -1;
+ private InetSocketAddress clientPortAddress;
+ private int maxClientConnections;
+ private int port = 2181;
+
+ public ZooKeeperServer getObject() throws Exception {
+ return zooKeeperServer;
+ }
+
+ public Class<ZooKeeperServer> getObjectType() {
+ return ZooKeeperServer.class;
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public void afterPropertiesSet() throws Exception {
+ if (purge) {
+ deleteFilesInDir(getDataLogDir());
+ deleteFilesInDir(getDataDir());
+ }
+ FileTxnSnapLog ftxn = new FileTxnSnapLog(getDataLogDir(), getDataDir());
+ zooKeeperServer.setTxnLogFactory(ftxn);
+ zooKeeperServer.setTickTime(getTickTime());
+ zooKeeperServer.setMinSessionTimeout(getMinSessionTimeout());
+ zooKeeperServer.setMaxSessionTimeout(getMaxSessionTimeout());
+ connectionFactory = new NIOServerCnxnFactory();
+ connectionFactory.configure(getClientPortAddress(), getMaxClientConnections());
+ connectionFactory.startup(zooKeeperServer);
+ }
+
+ private void deleteFilesInDir(File dir) {
+ File[] files = dir.listFiles();
+ if (files != null) {
+ for (File file : files) {
+ if (file.isDirectory()) {
+ deleteFilesInDir(file);
+ } else {
+ file.delete();
+ }
+ }
+ }
+ }
+
+ public void destroy() throws Exception {
+ shutdown();
+ }
+
+ protected void shutdown() {
+ if (connectionFactory != null) {
+ connectionFactory.shutdown();
+ try {
+ connectionFactory.join();
+ } catch (InterruptedException e) {
+ // Ignore
+ }
+ connectionFactory = null;
+ }
+ if (zooKeeperServer != null) {
+ zooKeeperServer.shutdown();
+ zooKeeperServer = null;
+ }
+ }
+
+ // Properties
+ //-------------------------------------------------------------------------
+
+ public ZooKeeperServer getZooKeeperServer() {
+ return zooKeeperServer;
+ }
+
+ public NIOServerCnxnFactory getConnectionFactory() {
+ return connectionFactory;
+ }
+
+ public File getDataLogDir() {
+ if (dataLogDir == null) {
+ dataLogDir = new File(getZKOutputDir(), "log");
+ dataLogDir.mkdirs();
+ }
+ return dataLogDir;
+ }
+
+ public File getDataDir() {
+ if (dataDir == null) {
+ dataDir = new File(getZKOutputDir(), "data");
+ dataDir.mkdirs();
+ }
+ return dataDir;
+ }
+
+ public int getTickTime() {
+ return tickTime;
+ }
+
+ public int getMinSessionTimeout() {
+ return minSessionTimeout;
+ }
+
+ public int getMaxSessionTimeout() {
+ return maxSessionTimeout;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public InetSocketAddress getClientPortAddress() {
+ if (clientPortAddress == null) {
+ clientPortAddress = new InetSocketAddress(port);
+ }
+ return clientPortAddress;
+ }
+
+ public int getMaxClientConnections() {
+ return maxClientConnections;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ public void setClientPortAddress(InetSocketAddress clientPortAddress) {
+ this.clientPortAddress = clientPortAddress;
+ }
+
+ public void setConnectionFactory(NIOServerCnxnFactory connectionFactory) {
+ this.connectionFactory = connectionFactory;
+ }
+
+ public void setDataDir(File dataDir) {
+ this.dataDir = dataDir;
+ }
+
+ public void setDataLogDir(File dataLogDir) {
+ this.dataLogDir = dataLogDir;
+ }
+
+ public void setMaxClientConnections(int maxClientConnections) {
+ this.maxClientConnections = maxClientConnections;
+ }
+
+ public void setMaxSessionTimeout(int maxSessionTimeout) {
+ this.maxSessionTimeout = maxSessionTimeout;
+ }
+
+ public void setMinSessionTimeout(int minSessionTimeout) {
+ this.minSessionTimeout = minSessionTimeout;
+ }
+
+ public void setTickTime(int tickTime) {
+ this.tickTime = tickTime;
+ }
+
+ public void setZooKeeperServer(ZooKeeperServer zooKeeperServer) {
+ this.zooKeeperServer = zooKeeperServer;
+ }
+
+ public boolean isPurge() {
+ return purge;
+ }
+
+ public void setPurge(boolean purge) {
+ this.purge = purge;
+ }
+
+ // Implementation methods
+ //-------------------------------------------------------------------------
+ protected File getZKOutputDir() {
+ String baseDir = System.getProperty("basedir", ".");
+ File dir = new File(baseDir + "/target/zk");
+ dir.mkdirs();
+ return dir;
+ }
+
+
+}