svn commit: r1161306 [1/3] - in /camel/trunk: components/ components/camel-zookeeper/ components/camel-zookeeper/src/ components/camel-zookeeper/src/main/ components/camel-zookeeper/src/main/java/ components/camel-zookeeper/src/main/java/org/ component...

[ha...@apache.org]

Author: hadrian
Date: Wed Aug 24 22:31:37 2011
New Revision: 1161306

URL: http://svn.apache.org/viewvc?rev=1161306&view=rev
Log:
CAMEL-2472. Patch applied with thanks to Stephen Gargan

Added:
    camel/trunk/components/camel-zookeeper/
    camel/trunk/components/camel-zookeeper/pom.xml
    camel/trunk/components/camel-zookeeper/src/
    camel/trunk/components/camel-zookeeper/src/main/
    camel/trunk/components/camel-zookeeper/src/main/java/
    camel/trunk/components/camel-zookeeper/src/main/java/org/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/NaturalSortComparator.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/SequenceComparator.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperComponent.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConnectionManager.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperMessage.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZookeeperProducer.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/AnyOfOperations.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/CreateOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistsOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/GetChildrenOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/GetDataOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/OperationResult.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/SetDataOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ZooKeeperOperation.java
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/
    camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/ZooKeeperRoutePolicy.java
    camel/trunk/components/camel-zookeeper/src/main/resources/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/org/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/org/apache/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/org/apache/camel/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/org/apache/camel/component/
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/org/apache/camel/component/zoo
    camel/trunk/components/camel-zookeeper/src/main/resources/META-INF/services/org/apache/camel/component/zookeeper
    camel/trunk/components/camel-zookeeper/src/test/
    camel/trunk/components/camel-zookeeper/src/test/java/
    camel/trunk/components/camel-zookeeper/src/test/java/org/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ConsumeChildrenTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ConsumeDataTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/NaturalSortComparatorTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperEndpointTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperTestSupport.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZooKeeperUtilsTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZookeeperConnectionManagerTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/ZookeeperProducerTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/AnyOfOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/CreateOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/DataChangedOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/ExistenceOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/operations/SetDataOperationTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverRoutePolicyTest.java
    camel/trunk/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/ZookeeperRoutePolicyTest.java
    camel/trunk/components/camel-zookeeper/src/test/resources/
    camel/trunk/components/camel-zookeeper/src/test/resources/log4j.properties
    camel/trunk/components/camel-zookeeper/src/test/resources/zkserver1.conf
    camel/trunk/components/camel-zookeeper/src/test/resources/zkserver2.conf
    camel/trunk/components/camel-zookeeper/src/test/resources/zkserver3.conf
Modified:
    camel/trunk/components/pom.xml
    camel/trunk/parent/pom.xml

Added: camel/trunk/components/camel-zookeeper/pom.xml
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/pom.xml?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/pom.xml (added)
+++ camel/trunk/components/camel-zookeeper/pom.xml Wed Aug 24 22:31:37 2011
@@ -0,0 +1,76 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project>
+<parent>
+    <artifactId>components</artifactId>
+    <groupId>org.apache.camel</groupId>
+    <version>2.9-SNAPSHOT</version>
+  </parent>
+  <modelVersion>4.0.0</modelVersion>
+  <artifactId>camel-zookeeper</artifactId>
+ 
+  <name>Camel :: Zookeeper</name>
+  <description>Camel Zookeeper Support</description>
+  
+  <properties>
+    <camel.osgi.export.pkg>org.apache.camel.component.zookeeper.*</camel.osgi.export.pkg>
+  </properties>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.zookeeper</groupId>
+      <artifactId>zookeeper</artifactId>
+    </dependency>  
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-core</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-core</artifactId>
+      <version>${pom.version}</version>
+      <scope>test</scope>      
+      <classifier>tests</classifier>
+    </dependency> 
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-spring</artifactId>
+    </dependency>    
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+    </dependency>   
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-log4j12</artifactId>
+    </dependency>      
+    <dependency>
+      <groupId>org.apache.camel</groupId>
+      <artifactId>camel-test</artifactId>      
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>log4j</groupId>
+      <artifactId>log4j</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+  <build>
+    <plugins>
+      <plugin>
+        <artifactId>maven-surefire-plugin</artifactId>
+        <configuration>
+          <forkMode>pertest</forkMode>
+          <excludes>
+            <!-- TODO FIXME ASAP -->
+            <exclude>**/ZooKeeperEndpointTest.*</exclude>
+          </excludes>
+        </configuration>
+      </plugin>
+    </plugins>
+  </build>  
+</project>

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ConnectionHolder.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,103 @@
+/**
+ * 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.zookeeper;
+
+import java.util.concurrent.CountDownLatch;
+
+import static java.lang.String.format;
+
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * <code>ConnectionHolder</code> watches for Connection based events from
+ * {@link ZooKeeper} and can be used to block until a connection has been
+ * established.
+ */
+public class ConnectionHolder implements Watcher {
+
+    private static final transient Log LOG = LogFactory.getLog(ConnectionHolder.class);
+
+    private CountDownLatch connectionLatch = new CountDownLatch(1);
+
+    private ZooKeeperConfiguration configuration;
+
+    private ZooKeeper zookeeper;
+
+    public ConnectionHolder(ZooKeeperConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    public ZooKeeper getZooKeeper() {
+        if (zookeeper != null) {
+            return zookeeper;
+        }
+        if (configuration.getConnectString() == null) {
+            throw new RuntimeCamelException("Cannot create ZooKeeper connection as connection string is null. Have servers been configured?");
+        }
+        try {
+            if (configuration.getSessionId() > 0 && configuration.getSessionPassword() != null) {
+                zookeeper = new ZooKeeper(configuration.getConnectString(), configuration.getTimeout(), this, configuration.getSessionId(), configuration.getSessionPassword());
+            } else {
+                zookeeper = new ZooKeeper(configuration.getConnectString(), configuration.getTimeout(), this);
+            }
+        } catch (Exception e) {
+            ObjectHelper.wrapRuntimeCamelException(e);
+        }
+        awaitConnection();
+        return zookeeper;
+    }
+
+    public boolean isConnected() {
+        return connectionLatch.getCount() == 0;
+    }
+
+    public void awaitConnection() {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug(format("Awaiting Connection event from Zookeeper cluster %s", configuration.getConnectString()));
+        }
+        try {
+            connectionLatch.await();
+        } catch (InterruptedException e) {
+            ObjectHelper.wrapRuntimeCamelException(e);
+        }
+    }
+
+    public void process(WatchedEvent event) {
+        if (event.getState() == KeeperState.SyncConnected) {
+            connectionLatch.countDown();
+        }
+        connectionLatch.countDown();
+    }
+
+    public void closeConnection() {
+        try {
+            zookeeper.close();
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(format("Shutting down connection to Zookeeper cluster %s", configuration.getConnectString()));
+            }
+        } catch (InterruptedException e) {
+            LOG.error("Error closing zookeeper connection.", e);
+        }
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/NaturalSortComparator.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/NaturalSortComparator.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/NaturalSortComparator.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/NaturalSortComparator.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,109 @@
+/**
+ * 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.zookeeper;
+
+import java.util.Comparator;
+
+/**
+ * <code>NaturalSortComparator</code> is a fast comparator for sorting a
+ * collection of Strings in a human readable fashion.
+ * <p>
+ * This implementation considers sequences of digits to be positive integer
+ * values, '.' does not indicate a decimal value nor '-' a negative one. As a
+ * result, 12345.12345 will sort higher than 12345.5432 and -12346 will sort
+ * higher than 12345.
+ *<p>
+ * it does work well for sorting software versions e.g. camel-2.2.0 sorting
+ * higher than camel-2.1.0
+ */
+public class NaturalSortComparator implements Comparator<CharSequence> {
+
+    public enum Order {
+        Ascending(1), Descending(-1);
+
+        int direction;
+
+        private Order(int direction) {
+            this.direction = direction;
+        }
+    }
+
+    private Order order;;
+
+    public NaturalSortComparator() {
+        this(Order.Ascending);
+    }
+
+    public NaturalSortComparator(Order order) {
+        if (order != null) {
+            this.order = order;
+        }
+    }
+
+    public int compare(CharSequence first, CharSequence second) {
+        if (first == null && second == null) {
+            return 0;
+        }
+        if (first != null && second == null) {
+            return 1;
+        }
+        if (first == null && second != null) {
+            return -1;
+        }
+
+        int compare = 0;
+        int fx = 0;
+        int sx = 0;
+        while (fx < first.length() && sx < second.length() && compare == 0) {
+            if (isDigit(first.charAt(fx)) && isDigit(second.charAt(sx))) {
+                int flen = getNumSequenceLength(first, fx);
+                int slen = getNumSequenceLength(second, sx);
+
+                if (flen == slen) {
+                    for (int x = 0; x < flen && compare == 0; x++) {
+                        /** the first difference in digit wins */
+                        compare = first.charAt(fx++) - second.charAt(sx++);
+                    }
+                } else {
+                    compare = flen - slen;
+                }
+            } else {
+                compare = first.charAt(fx) - second.charAt(sx);
+            }
+            fx++;
+            sx++;
+        }
+
+        if (compare == 0) {
+            compare = first.length() - second.length();
+        }
+        return order.direction * compare;
+    }
+
+    private boolean isDigit(char c) {
+        return c >= 48 && c < 57;
+    }
+
+    private int getNumSequenceLength(CharSequence sequence, int index) {
+        int x = index;
+        while (x < sequence.length() && isDigit(sequence.charAt(x))) {
+            x++;
+        }
+        return x - index;
+    }
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/SequenceComparator.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/SequenceComparator.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/SequenceComparator.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/SequenceComparator.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,49 @@
+/**
+ * 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.zookeeper;
+
+import org.apache.zookeeper.CreateMode;
+
+/**
+ * Nodes created with any of Sequential {@link CreateMode}s will have a 10
+ * character sequence attached to their node names.
+ * <code>SequenceComparator</code> is a Natural comparator used to compare lists
+ * of objects with these appended sequences.
+ */
+public class SequenceComparator extends NaturalSortComparator {
+
+    public static final int ZOOKEEPER_SEQUENCE_LENGTH = 10;
+
+    @Override
+    public int compare(CharSequence sequencedNode, CharSequence otherSequencedNode) {
+        if (sequencedNode == null && otherSequencedNode == null) {
+            return 0;
+        }
+        if (sequencedNode != null && otherSequencedNode == null) {
+            return 1;
+        }
+        if (sequencedNode == null && otherSequencedNode != null) {
+            return -1;
+        }
+        return super.compare(getZooKeeperSequenceNumber(sequencedNode), getZooKeeperSequenceNumber(otherSequencedNode));
+    }
+
+    private CharSequence getZooKeeperSequenceNumber(CharSequence sequencedNodeName) {
+        int len = sequencedNodeName.length();
+        return sequencedNodeName.subSequence(len - ZOOKEEPER_SEQUENCE_LENGTH, len);
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperComponent.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperComponent.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperComponent.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperComponent.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,75 @@
+/**
+ * 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.zookeeper;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Map;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelException;
+import org.apache.camel.Endpoint;
+import org.apache.camel.impl.DefaultComponent;
+
+/**
+ * Component that creates {@link ZooKeeperEndpoint}s for interacting with a ZooKeeper cluster.
+ */
+public class ZooKeeperComponent extends DefaultComponent {
+
+    private ZooKeeperConfiguration configuration;
+
+    public ZooKeeperComponent() {
+    }
+
+    public ZooKeeperComponent(CamelContext context) {
+        super(context);
+    }
+
+    public ZooKeeperComponent(ZooKeeperConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    @SuppressWarnings("all")
+    protected Endpoint createEndpoint(String uri, String remaining, Map parameters) throws Exception {
+        if (getCamelContext() == null) {
+            throw new CamelException("No Camel context has been provided to this zookeeper component");
+        }
+
+        ZooKeeperConfiguration config = getConfiguration();
+        extractConfigFromUri(uri, config);
+        setProperties(config, parameters);
+
+        return new ZooKeeperEndpoint(uri, this, config.copy());
+    }
+
+    private void extractConfigFromUri(String remaining, ZooKeeperConfiguration config) throws URISyntaxException {
+        URI u = new URI(remaining);
+        config.addZookeeperServer(u.getHost() + (u.getPort() != -1 ? ":" + u.getPort() : ""));
+        config.setPath(u.getPath());
+    }
+
+    public ZooKeeperConfiguration getConfiguration() {
+        if (configuration == null) {
+            configuration = new ZooKeeperConfiguration();
+        }
+        return configuration;
+    }
+
+    public void setConfiguration(ZooKeeperConfiguration configuration) {
+        this.configuration = configuration;
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConfiguration.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,166 @@
+/**
+ * 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.zookeeper;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.camel.RuntimeCamelException;
+
+/**
+ * <code>ZookeeperConfiguration</code> encapsulates the configuration used to
+ * interact with a ZooKeeper cluster. Most typically it is parsed from endpoint
+ * uri but may also be configured programatically and applied to a
+ * {@link ZooKeeperComponent}. A copy of this component's configuration will be
+ * injected into any {@link ZooKeeperEndpoint}s the component creates.
+ *
+ * @version $
+ */
+public class ZooKeeperConfiguration implements Cloneable {
+
+    private int timeout = 5000;
+    private long backoff = 5000;
+    private List<String> servers;
+    private boolean changed;
+    private int sessionId;
+    private byte[] password;
+    private String path;
+    private boolean awaitCreation = true;
+    private boolean repeat;
+    private boolean listChildren;
+    private boolean shouldCreate;
+    private String createMode;
+
+    public void addZookeeperServer(String server) {
+        if (servers == null) {
+            servers = new ArrayList<String>();
+        }
+        servers.add(server);
+        changed = true;
+    }
+
+    public List<String> getServers() {
+        return servers;
+    }
+
+    public void setServers(List<String> servers) {
+        this.servers = servers;
+    }
+
+    public int getTimeout() {
+        return timeout;
+    }
+
+    public void setTimeout(int timeout) {
+        this.timeout = timeout;
+        changed = true;
+    }
+
+    public boolean listChildren() {
+        return listChildren;
+    }
+
+    public void setListChildren(boolean listChildren) {
+        this.listChildren = listChildren;
+    }
+
+    public void clearChanged() {
+        changed = false;
+    }
+
+    public boolean isChanged() {
+        return changed;
+    }
+
+    public String getConnectString() {
+        // return StringHelper.toStringCommaSeparated(servers);
+        StringBuilder b = new StringBuilder();
+        for (String server : servers) {
+            b.append(server).append(",");
+        }
+        b.setLength(b.length() - 1);
+        return b.toString();
+
+    }
+
+    public byte[] getSessionPassword() {
+        return password;
+    }
+
+    public int getSessionId() {
+        return sessionId;
+    }
+
+    public void setPath(String path) {
+        this.path = path;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public boolean shouldRepeat() {
+        return repeat;
+    }
+
+    public void setRepeat(boolean repeat) {
+        this.repeat = repeat;
+    }
+
+    public ZooKeeperConfiguration copy() {
+        try {
+            return (ZooKeeperConfiguration)clone();
+        } catch (CloneNotSupportedException e) {
+            throw new RuntimeCamelException(e);
+        }
+    }
+
+    public boolean shouldAwaitExistence() {
+        return awaitCreation;
+    }
+
+    public void setAwaitExistance(boolean awaitCreation) {
+        this.awaitCreation = awaitCreation;
+    }
+
+    public long getBackoff() {
+        return backoff;
+    }
+
+    public void setBackoff(long backoff) {
+        this.backoff = backoff;
+    }
+
+    public void setCreate(boolean shouldCreate) {
+        this.shouldCreate = shouldCreate;
+    }
+
+    public boolean shouldCreate() {
+        return shouldCreate;
+    }
+
+    public String getCreateMode() {
+        return createMode;
+    }
+
+    public void setCreateMode(String createMode) {
+        this.createMode = createMode;
+    }
+    
+    
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConnectionManager.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConnectionManager.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConnectionManager.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConnectionManager.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,74 @@
+/**
+ * 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.zookeeper;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * <code>ZookeeperConnectionManager</code> is a simple class to manage
+ * {@link ZooKeeper} connections.
+ * 
+ * @author sgargan
+ */
+public class ZooKeeperConnectionManager {
+
+    private static final Log LOG = LogFactory.getLog(ZooKeeperConnectionManager.class);
+
+    private ZookeeperConnectionStrategy strategy;
+
+    public ZooKeeperConnectionManager(ZooKeeperEndpoint endpoint) {
+        strategy = new DefaultZookeeperConnectionStrategy(endpoint);
+    }
+
+    public ZooKeeper getConnection() {
+        return strategy.getConnection().getZooKeeper();
+    }
+
+    private interface ZookeeperConnectionStrategy {
+        ConnectionHolder getConnection();
+
+        void shutdown();
+    }
+
+    private class DefaultZookeeperConnectionStrategy implements ZookeeperConnectionStrategy {
+        private ConnectionHolder holder;
+        private ZooKeeperConfiguration configuration;
+
+        public DefaultZookeeperConnectionStrategy(ZooKeeperEndpoint endpoint) {
+            this.configuration = endpoint.getConfiguration();
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(String.format("Creating connection with static configuration of %s", configuration));
+            }
+            holder = new ConnectionHolder(configuration);
+        }
+
+        public ConnectionHolder getConnection() {
+            return holder;
+        }
+
+        public void shutdown() {
+            holder.closeConnection();
+        }
+    }
+
+    public void shutdown() {
+        strategy.shutdown();
+    }
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperConsumer.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,189 @@
+/**
+ * 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.zookeeper;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import static java.lang.String.format;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.component.zookeeper.operations.AnyOfOperations;
+import org.apache.camel.component.zookeeper.operations.ChildrenChangedOperation;
+import org.apache.camel.component.zookeeper.operations.DataChangedOperation;
+import org.apache.camel.component.zookeeper.operations.ExistenceChangedOperation;
+import org.apache.camel.component.zookeeper.operations.ExistsOperation;
+import org.apache.camel.component.zookeeper.operations.GetChildrenOperation;
+import org.apache.camel.component.zookeeper.operations.GetDataOperation;
+import org.apache.camel.component.zookeeper.operations.OperationResult;
+import org.apache.camel.component.zookeeper.operations.ZooKeeperOperation;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * <code>ZooKeeperConsumer</code> uses various {@link ZooKeeperOperation} to
+ * interact and consume data from a ZooKeeper cluster.
+ */
+@SuppressWarnings("rawtypes")
+public class ZooKeeperConsumer extends DefaultConsumer {
+
+    private ZooKeeperConnectionManager connectionManager;
+
+    private ZooKeeper connection;
+
+    private ZooKeeperConfiguration configuration;
+
+    private LinkedBlockingQueue<ZooKeeperOperation> operations = new LinkedBlockingQueue<ZooKeeperOperation>();
+
+    private boolean shuttingDown;
+
+    private ExecutorService executor;
+
+    public ZooKeeperConsumer(ZooKeeperEndpoint endpoint, Processor processor) {
+        super(endpoint, processor);
+        this.connectionManager = endpoint.getConnectionManager();
+        this.configuration = endpoint.getConfiguration();
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+        connection = connectionManager.getConnection();
+        if (log.isDebugEnabled()) {
+            log.debug(format("Connected to Zookeeper cluster %s", configuration.getConnectString()));
+        }
+
+        initializeConsumer();
+        executor = getEndpoint().getCamelContext().getExecutorServiceManager().newFixedThreadPool(configuration.getPath(), "Camel-Zookeeper Ops executor", 1);
+        OperationsExecutor opsService = new OperationsExecutor();
+        executor.execute(opsService);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        super.doStop();
+        shuttingDown = true;
+        connection = connectionManager.getConnection();
+        if (log.isTraceEnabled()) {
+            log.trace(format("Shutting down zookeeper consumer of '%s'", configuration.getPath()));
+        }
+        executor.shutdown();
+        connectionManager.shutdown();
+    }
+
+    private void initializeConsumer() {
+        String node = configuration.getPath();
+        if (configuration.listChildren()) {
+            initializeChildListingConsumer(node);
+        } else {
+            initializeDataConsumer(node);
+        }
+    }
+
+    private void initializeDataConsumer(String node) {
+        if (!shuttingDown) {
+            if (log.isDebugEnabled()) {
+                log.debug(String.format("Initailizing consumption of data on node '%s'", node));
+            }
+            addBasicDataConsumeSequence(node);
+        }
+    }
+
+    private void initializeChildListingConsumer(String node) {
+        if (!shuttingDown) {
+            if (log.isDebugEnabled()) {
+                log.debug(String.format("Initailizing child listing of node '%s'", node));
+            }
+            addBasicChildListingSequence(node);
+        }
+    }
+
+    private Exchange createExchange(String path, OperationResult result) {
+        Exchange e = getEndpoint().createExchange();
+        ZooKeeperMessage in = new ZooKeeperMessage(path, result.getStatistics());
+        e.setIn(in);
+        if (result.isOk()) {
+            in.setBody(result.getResult());
+        } else {
+            e.setException(result.getException());
+        }
+        return e;
+    }
+
+    private class OperationsExecutor implements Runnable {
+
+        private ZooKeeperOperation current;
+
+        public void run() {
+            while (isRunAllowed()) {
+
+                try {
+                    current = operations.take();
+                    if (log.isTraceEnabled()) {
+                        log.trace(format("Processing '%s' operation", current.getClass().getSimpleName()));
+                    }
+                } catch (InterruptedException e) {
+                    continue;
+                }
+                String node = current.getNode();
+                try {
+                    OperationResult result = current.get();
+                    if (result != null && current.shouldProduceExchange()) {
+                        getProcessor().process(createExchange(node, result));
+                    }
+                } catch (Exception e) {
+                    handleException(e);
+                    backoffAndThenRestart();
+                } finally {
+                    if (configuration.shouldRepeat()) {
+                        try {
+                            operations.offer(current.createCopy());
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                            backoffAndThenRestart();
+                        }
+                    }
+                }
+            }
+        }
+
+        private void backoffAndThenRestart() {
+            try {
+                if (isRunAllowed()) {
+                    Thread.sleep(configuration.getBackoff());
+                    initializeConsumer();
+                }
+            } catch (Exception e) {
+            }
+        }
+    }
+
+    private void addBasicDataConsumeSequence(String node) {
+        operations.clear();
+        operations.add(new AnyOfOperations(node, new ExistsOperation(connection, node), new ExistenceChangedOperation(connection, node)));
+        operations.add(new GetDataOperation(connection, node));
+        operations.add(new DataChangedOperation(connection, node, false));
+    }
+
+    private void addBasicChildListingSequence(String node) {
+        operations.clear();
+        operations.add(new AnyOfOperations(node, new ExistsOperation(connection, node), new ExistenceChangedOperation(connection, node)));
+        operations.add(new GetChildrenOperation(connection, node));
+        operations.add(new ChildrenChangedOperation(connection, node, false));
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperEndpoint.java Wed Aug 24 22:31:37 2011
@@ -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.zookeeper;
+
+import java.util.List;
+
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultEndpoint;
+import org.apache.camel.spi.management.ManagedAttribute;
+import org.apache.camel.spi.management.ManagedOperation;
+import org.apache.camel.spi.management.ManagedResource;
+
+/**
+ * <code>ZooKeeperEndpoint</code>
+ */
+@ManagedResource(description = "ZooKeeper Endpoint")
+public class ZooKeeperEndpoint extends DefaultEndpoint {
+    private ZooKeeperConfiguration configuration;
+    private ZooKeeperConnectionManager connectionManager;
+
+    public ZooKeeperEndpoint(String uri, ZooKeeperComponent component, ZooKeeperConfiguration configuration) {
+        super(uri, component);
+        this.configuration = configuration;
+        this.connectionManager = new ZooKeeperConnectionManager(this);
+    }
+
+    public Producer createProducer() throws Exception {
+        return new ZookeeperProducer(this);
+    }
+
+    public Consumer createConsumer(Processor processor) throws Exception {
+        return new ZooKeeperConsumer(this, processor);
+    }
+
+    public boolean isSingleton() {
+        return true;
+    }
+
+    public void setConfiguration(ZooKeeperConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    public ZooKeeperConfiguration getConfiguration() {
+        return configuration;
+    }
+
+    ZooKeeperConnectionManager getConnectionManager() {
+        return connectionManager;
+    }
+
+    @ManagedAttribute(description = "Session Password")
+    public byte[] getSessionPassword() {
+        return getConfiguration().getSessionPassword();
+    }
+
+    @ManagedAttribute
+    public int getSessionId() {
+        return getConfiguration().getSessionId();
+    }
+
+    @ManagedAttribute
+    public void setPath(String path) {
+        getConfiguration().setPath(path);
+    }
+
+    @ManagedAttribute
+    public String getPath() {
+        return getConfiguration().getPath();
+    }
+
+    @ManagedAttribute
+    public int getTimeout() {
+        return getConfiguration().getTimeout();
+    }
+
+    @ManagedAttribute
+    public void setTimeout(int timeout) {
+        getConfiguration().setTimeout(timeout);
+    }
+
+    @ManagedAttribute
+    public boolean getRepeat() {
+        return getConfiguration().shouldRepeat();
+    }
+
+    @ManagedAttribute
+    public void setRepeat(boolean shouldRepeat) {
+        getConfiguration().setRepeat(shouldRepeat);
+    }
+
+    @ManagedAttribute
+    public List<String> getServers() {
+        return getConfiguration().getServers();
+    }
+
+    @ManagedAttribute
+    public void setServers(List<String> servers) {
+        getConfiguration().setServers(servers);
+    }
+
+    // TODO: check how to allow should to indicate getter of an attribute pair
+//    public boolean shouldListChildren() {
+//        return getConfiguration().listChildren();
+//    }
+
+    @ManagedAttribute
+    public boolean getListChildren() {
+        return getConfiguration().listChildren();
+    }
+
+    @ManagedAttribute
+    public void setListChildren(boolean listChildren) {
+        getConfiguration().setListChildren(listChildren);
+    }
+
+//    public boolean shouldCreate() {
+//        return getConfiguration().shouldCreate();
+//    }
+
+    @ManagedAttribute
+    public boolean getCreate() {
+        return getConfiguration().shouldCreate();
+    }
+
+    @ManagedAttribute
+    public void setCreate(boolean shouldCreate) {
+        getConfiguration().setCreate(shouldCreate);
+    }
+
+    @ManagedAttribute
+    public long getBackoff() {
+        return getConfiguration().getBackoff();
+    }
+
+    @ManagedAttribute
+    public void setBackoff(long backoff) {
+        getConfiguration().setBackoff(backoff);
+    }
+
+//    public boolean shouldAwaitExistence() {
+//        return getConfiguration().shouldAwaitExistence();
+//    }
+
+    @ManagedAttribute
+    public boolean getAwaitExistence() {
+        return getConfiguration().shouldAwaitExistence();
+    }
+
+    @ManagedAttribute
+    public void setAwaitExistence(boolean awaitExistence) {
+        getConfiguration().setAwaitExistance(awaitExistence);
+    }
+
+    @ManagedOperation
+    public void addServer(String server) {
+        getConfiguration().addZookeeperServer(server);
+    }
+
+    @ManagedOperation
+    public void clearServers() {
+        getConfiguration().getServers().clear();
+    }
+
+    public Object getManagedObject(ZooKeeperEndpoint arg0) {
+        return this;
+    }
+
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperMessage.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperMessage.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperMessage.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperMessage.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,77 @@
+/**
+ * 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.zookeeper;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.camel.Message;
+import org.apache.camel.impl.DefaultMessage;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * <code>ZooKeeperMessage</code> is a {@link org.apache.camel.Message}
+ * representing interactions with a ZooKeeper service. It contains a number of
+ * optional Header Constants that are used by the Producer and consumer
+ * mechanisms to finely control these interactions.
+ * 
+ * @version $
+ */
+public class ZooKeeperMessage extends DefaultMessage {
+
+    public static final String ZOOKEEPER_NODE = "CamelZooKeeperNode";
+
+    public static final String ZOOKEEPER_NODE_VERSION = "CamelZooKeeperVersion";
+
+    public static final String ZOOKEEPER_ERROR_CODE = "CamelZooKeeperErrorCode";
+
+    public static final String ZOOKEEPER_ACL = "CamelZookeeperAcl";
+
+    public static final String ZOOKEEPER_CREATE_MODE = "CamelZookeeperCreateMode";
+
+    public static final String ZOOKEEPER_STATISTICS = "CamelZookeeperStatistics";
+
+    @SuppressWarnings("unchecked")
+    public ZooKeeperMessage(String node, Stat statistics) {
+        this(node, statistics, Collections.EMPTY_MAP);
+    }
+
+    public ZooKeeperMessage(String node, Stat statistics, Map<String, Object> headers) {
+        setHeaders(headers);
+        this.setHeader(ZOOKEEPER_NODE, node);
+        this.setHeader(ZOOKEEPER_STATISTICS, statistics);
+    }
+
+    public static Stat getStatistics(Message message) {
+        Stat stats = null;
+        if (message != null) {
+            stats = message.getHeader(ZOOKEEPER_STATISTICS, Stat.class);
+        }
+        return stats;
+    }
+
+    public static String getPath(Message message) {
+        String path = null;
+        if (message != null) {
+            path = message.getHeader(ZOOKEEPER_NODE, String.class);
+        }
+        return path;
+    }
+    
+    
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperUtils.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,105 @@
+/**
+ * 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.zookeeper;
+
+import java.util.List;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.util.ExchangeHelper;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.ACL;
+
+/**
+ * <code>ZooKeeperUtils</code> contains static utility functions mostly for
+ * retrieving optional message properties from Message headers.
+ */
+public final class ZooKeeperUtils {
+
+    private ZooKeeperUtils() { }
+    
+    /**
+     * Pulls a createMode flag from the header keyed by
+     * {@link ZooKeeperMessage.ZOOKEEPER_CREATE_MODE} in the given message and
+     * attemps to parse a {@link CreateMode} from it.
+     *
+     * @param message the message that may contain a ZOOKEEPER_CREATE_MODE
+     *            header.
+     * @return the parsed {@link CreateMode} or null if the header was null or
+     *         not a valid mode flag.
+     */
+    public static CreateMode getCreateMode(Message message, CreateMode defaultMode) {
+        CreateMode mode = null;
+
+        try {
+            mode = message.getHeader(ZooKeeperMessage.ZOOKEEPER_CREATE_MODE, CreateMode.class);
+        } catch (Exception e) {
+        }
+        
+        if (mode == null) {
+            String modeHeader = message.getHeader(ZooKeeperMessage.ZOOKEEPER_CREATE_MODE, String.class);
+            mode = getCreateModeFromString(modeHeader, defaultMode);
+        }
+        return mode == null ? defaultMode : mode;
+    }
+
+    public static CreateMode getCreateModeFromString(String modeHeader, CreateMode defaultMode) {
+        
+        CreateMode mode = null;
+        if (modeHeader != null) {
+            try {
+                mode = CreateMode.valueOf(modeHeader);
+            } catch (Exception e) {
+            }
+        }
+        return mode == null ? defaultMode : mode;
+    }
+
+    /**
+     * Pulls the target node from the header keyed by
+     * {@link ZooKeeperMessage.ZOOKEEPER_NODE}. This node is then typically used
+     * in place of the configured node extracted from the endpoint uri.
+     *
+     * @param message the message that may contain a ZOOKEEPER_NODE header.
+     * @return the node property or null if the header was null
+     */
+    public static String getNodeFromMessage(Message message, String defaultNode) {
+        return getZookeeperProperty(message, ZooKeeperMessage.ZOOKEEPER_NODE, defaultNode, String.class);
+    }
+
+    public static Integer getVersionFromMessage(Message message) {
+        return getZookeeperProperty(message, ZooKeeperMessage.ZOOKEEPER_NODE_VERSION, -1, Integer.class);
+    }
+
+    public static byte[] getPayloadFromExchange(Exchange exchange) {
+        return ExchangeHelper.convertToType(exchange, byte[].class, exchange.getIn().getBody());
+    }
+
+    @SuppressWarnings("unchecked")
+    public static List<ACL> getAclListFromMessage(Message in) {
+        return getZookeeperProperty(in, ZooKeeperMessage.ZOOKEEPER_ACL, Ids.OPEN_ACL_UNSAFE, List.class);
+    }
+
+    public static <T> T getZookeeperProperty(Message m, String propertyName, T defaultValue, Class<? extends T> type) {
+        T value = m.getHeader(propertyName, type);
+        if (value == null) {
+            value = defaultValue;
+        }
+        return value;
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZookeeperProducer.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZookeeperProducer.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZookeeperProducer.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZookeeperProducer.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,195 @@
+/**
+ * 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.zookeeper;
+
+import static java.lang.String.format;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Message;
+import org.apache.camel.component.zookeeper.operations.CreateOperation;
+import org.apache.camel.component.zookeeper.operations.GetChildrenOperation;
+import org.apache.camel.component.zookeeper.operations.OperationResult;
+import org.apache.camel.component.zookeeper.operations.SetDataOperation;
+import org.apache.camel.impl.DefaultProducer;
+import org.apache.camel.util.ExchangeHelper;
+
+import org.apache.zookeeper.AsyncCallback.StatCallback;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+import static org.apache.camel.component.zookeeper.ZooKeeperUtils.getAclListFromMessage;
+import static org.apache.camel.component.zookeeper.ZooKeeperUtils.getCreateMode;
+import static org.apache.camel.component.zookeeper.ZooKeeperUtils.getCreateModeFromString;
+import static org.apache.camel.component.zookeeper.ZooKeeperUtils.getNodeFromMessage;
+import static org.apache.camel.component.zookeeper.ZooKeeperUtils.getPayloadFromExchange;
+import static org.apache.camel.component.zookeeper.ZooKeeperUtils.getVersionFromMessage;
+
+
+/**
+ * <code>ZookeeperProducer</code> attempts to set the content of nodes in the
+ * {@link ZooKeeper} cluster with the payloads of the of the exchanges it
+ * receives.
+ *
+ * @version $
+ */
+@SuppressWarnings("rawtypes")
+public class ZookeeperProducer extends DefaultProducer {
+
+    private ZooKeeperConfiguration configuration;
+
+    private ZooKeeperConnectionManager zkm;
+
+    public ZookeeperProducer(ZooKeeperEndpoint endpoint) {
+        super(endpoint);
+        this.configuration = endpoint.getConfiguration();
+        this.zkm = endpoint.getConnectionManager();
+    }
+
+    public void process(Exchange exchange) throws Exception {
+
+        ZooKeeper connection = zkm.getConnection();
+        ProductionContext context = new ProductionContext(connection, exchange);
+
+        if (ExchangeHelper.isOutCapable(exchange)) {
+            if (log.isDebugEnabled()) {
+                log.debug(format("Storing data to znode '%s', waiting for confirmation", context.node));
+            }
+
+            OperationResult result = synchronouslySetData(context);
+            if (configuration.listChildren()) {
+                result = listChildren(context);
+            }
+
+            updateExchangeWithResult(context, result);
+        } else {
+            asynchronouslySetDataOnNode(connection, context);
+        }
+    }
+
+    private void asynchronouslySetDataOnNode(ZooKeeper connection, ProductionContext context) {
+        if (log.isDebugEnabled()) {
+            log.debug(format("Storing data to node '%s', not waiting for confirmation", context.node));
+        }
+        connection.setData(context.node, context.payload, context.version, new AsyncSetDataCallback(), context);
+    }
+
+    private void updateExchangeWithResult(ProductionContext context, OperationResult result) {
+        ZooKeeperMessage out = new ZooKeeperMessage(context.node, result.getStatistics(), context.in.getHeaders());
+        if (result.isOk()) {
+            out.setBody(result.getResult());
+        } else {
+            context.exchange.setException(result.getException());
+        }
+        
+        context.exchange.setOut(out);
+    }
+
+    private OperationResult listChildren(ProductionContext context) throws Exception {
+        return new GetChildrenOperation(context.connection, configuration.getPath()).get();
+    }
+
+    /** Simple container to avoid passing all these around as parameters */
+    private class ProductionContext {
+        ZooKeeper connection;
+        Exchange exchange;
+        Message in;
+        byte[] payload;
+        int version;
+        String node;
+
+        public ProductionContext(ZooKeeper connection, Exchange exchange) {
+            this.connection = connection;
+            this.exchange = exchange;
+            this.in = exchange.getIn();
+            this.node = getNodeFromMessage(in, configuration.getPath());
+            this.version = getVersionFromMessage(in);
+            this.payload = getPayloadFromExchange(exchange);
+        }
+    }
+
+    private class AsyncSetDataCallback implements StatCallback {
+
+        public void processResult(int rc, String node, Object ctx, Stat statistics) {
+            if (Code.NONODE.equals(Code.get(rc))) {
+                if (configuration.shouldCreate()) {
+                    log.warn(format("Node '%s' did not exist, creating it...", node));
+                    ProductionContext context = (ProductionContext)ctx;
+                    OperationResult<String> result = null;
+                    try {
+                        result = createNode(context);
+                    } catch (Exception e) {
+                        log.error(format("Error trying to create node '%s'", node), e);
+                    }
+
+                    if (result == null || !result.isOk()) {
+                        log.error(format("Error creating node '%s'", node), result.getException());
+                    }
+                }
+            } else {
+                logStoreComplete(node, statistics);
+            }
+        }
+    }
+
+    private OperationResult<String> createNode(ProductionContext ctx) throws Exception {
+        CreateOperation create = new CreateOperation(ctx.connection, ctx.node);
+        create.setPermissions(getAclListFromMessage(ctx.exchange.getIn()));
+        
+        CreateMode mode = null;
+        String modeString = configuration.getCreateMode();
+        if (modeString != null) {
+            try {
+                mode = getCreateModeFromString(modeString, CreateMode.EPHEMERAL);
+            } catch (Exception e) { }
+        } else {
+            mode = getCreateMode(ctx.exchange.getIn(), CreateMode.EPHEMERAL);
+        }
+        create.setCreateMode(mode == null ? CreateMode.EPHEMERAL : mode);
+        create.setData(ctx.payload);
+        return create.get();
+    }
+
+    /**
+     * Tries to set the data first and if a no node error is received then an
+     * attempt will be made to create it instead.
+     */
+    private OperationResult synchronouslySetData(ProductionContext ctx) throws Exception {
+
+        SetDataOperation setData = new SetDataOperation(ctx.connection, ctx.node, ctx.payload);
+        setData.setVersion(ctx.version);
+
+        OperationResult result = setData.get();
+
+        if (!result.isOk() && configuration.shouldCreate() && result.failedDueTo(Code.NONODE)) {
+            log.warn(format("Node '%s' did not exist, creating it.", ctx.node));
+            result = createNode(ctx);
+        }
+        return result;
+    }
+
+    private void logStoreComplete(String path, Stat statistics) {
+        if (log.isDebugEnabled()) {
+            if (log.isTraceEnabled()) {
+                log.trace(format("Stored data to node '%s', and receive statistics %s", path, statistics));
+            } else {
+                log.debug(format("Stored data to node '%s'", path));
+            }
+        }
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/AnyOfOperations.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/AnyOfOperations.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/AnyOfOperations.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/AnyOfOperations.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,75 @@
+/**
+ * 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.zookeeper.operations;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * <code>AnyOfOperations</code> is a composite operation of one or more sub
+ * operation, executing each in turn until any one succeeds. If any execute
+ * successfully, this operation succeeds; if the sub operations are all executed
+ * without success it fails.
+ * <p>
+ * It is mostly used for test and watch scenarios where a node is tested for
+ * existence, data or children, falling back to a corresponding watch operation
+ * if the test operation fails.
+ */
+@SuppressWarnings("rawtypes")
+public class AnyOfOperations extends ZooKeeperOperation {
+
+    private ZooKeeperOperation[] keeperOperations;
+
+    public AnyOfOperations(String node, ZooKeeperOperation... keeperOperations) {
+        super(null, node, false);
+        this.keeperOperations = keeperOperations;
+    }
+
+    @Override
+    public OperationResult get() throws InterruptedException, ExecutionException {
+        for (ZooKeeperOperation op : keeperOperations) {
+            try {
+                OperationResult result = op.get();
+                if (result.isOk()) {
+                    return result;
+                }
+            } catch (Exception e) {
+            }
+        }
+        throw new ExecutionException("All operations exhausted without one result", null);
+    }
+
+    @Override
+    public OperationResult get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+        return get();
+    }
+
+    @Override
+    public OperationResult getResult() {
+        return null;
+    }
+
+    @Override
+    public ZooKeeperOperation createCopy() throws Exception {
+        ZooKeeperOperation[] copy = new ZooKeeperOperation[keeperOperations.length];
+        for (int x = 0; x < keeperOperations.length; x++) {
+            copy[x] = keeperOperations[x].createCopy();
+        }
+        return new AnyOfOperations(node, copy);
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ChildrenChangedOperation.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,64 @@
+/**
+ * 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.zookeeper.operations;
+
+import java.util.List;
+
+import org.apache.zookeeper.AsyncCallback.Children2Callback;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * <code>ChildrenChangedOperation</code> is an watch driven operation.
+ * It will wait for an watched event indicating that the children associated with a
+ * given node have been modified before retrieving the changed list.
+ */
+public class ChildrenChangedOperation extends FutureEventDrivenOperation<List<String>> {
+
+    private boolean getChangedListing;
+
+    public ChildrenChangedOperation(ZooKeeper connection, String znode) {
+        this(connection, znode, true);
+    }
+
+    public ChildrenChangedOperation(ZooKeeper connection, String znode, boolean getChangedListing) {
+        super(connection, znode, EventType.NodeChildrenChanged);
+        this.getChangedListing = getChangedListing;
+    }
+
+    @Override
+    protected void installWatch() {
+        connection.getChildren(getNode(), this, new Children2Callback() {
+            public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
+            }
+        }, null);
+    }
+
+    @Override
+    public OperationResult<List<String>> getResult() {
+        return getChangedListing ? new GetChildrenOperation(connection, node).getResult() : null;
+    }
+
+    @Override
+    @SuppressWarnings("rawtypes")
+    public ZooKeeperOperation createCopy() throws Exception {
+        ChildrenChangedOperation copy = (ChildrenChangedOperation) super.createCopy();
+        copy.getChangedListing = getChangedListing;
+        return copy;
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/CreateOperation.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/CreateOperation.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/CreateOperation.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/CreateOperation.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,75 @@
+/**
+ * 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.zookeeper.operations;
+
+import java.util.List;
+
+import static java.lang.String.format;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * <code>CreateOperation</code> is a basic Zookeeper operation used to create
+ * and set the data contained in a given node
+ */
+public class CreateOperation extends ZooKeeperOperation<String> {
+
+    private static final List<ACL> DEFAULT_PERMISSIONS = Ids.OPEN_ACL_UNSAFE;
+
+    private static final CreateMode DEFAULT_MODE = CreateMode.EPHEMERAL;
+
+    private byte[] data;
+
+    private List<ACL> permissions = DEFAULT_PERMISSIONS;
+
+    private CreateMode createMode = DEFAULT_MODE;
+
+    public CreateOperation(ZooKeeper connection, String node) {
+        super(connection, node);
+    }
+
+    @Override
+    public OperationResult<String> getResult() {
+        try {
+            String created = connection.create(node, data, permissions, createMode);
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(format("Created node '%s' using mode '%s'", created, createMode));
+            }
+            // for consistency with other operations return an empty stats set.
+            return new OperationResult<String>(created, new Stat());
+        } catch (Exception e) {
+            return new OperationResult<String>(e);
+        }
+    }
+
+    public void setData(byte[] data) {
+        this.data = data;
+    }
+
+    public void setPermissions(List<ACL> permissions) {
+        this.permissions = permissions;
+    }
+
+    public void setCreateMode(CreateMode createMode) {
+        this.createMode = createMode;
+    }
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/DataChangedOperation.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,57 @@
+/**
+ * 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.zookeeper.operations;
+
+import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * <code>DataChangedOperation</code> is an watch driven operation. It will wait
+ * for an watched event indicating that the data contained in a given
+ * node has changed before optionally retrieving the changed data.
+ */
+@SuppressWarnings("rawtypes")
+public class DataChangedOperation extends FutureEventDrivenOperation<byte[]> {
+
+    protected static final Class[] CONSTRUCTOR_ARGS = {ZooKeeper.class, String.class, boolean.class};
+    
+    private boolean getChangedData;
+
+    public DataChangedOperation(ZooKeeper connection, String znode, boolean getChangedData) {
+        super(connection, znode, EventType.NodeDataChanged, EventType.NodeDeleted);
+        this.getChangedData = getChangedData;
+    }
+
+    @Override
+    protected void installWatch() {
+        connection.getData(getNode(), this, new DataCallback() {
+            public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
+            }
+        }, null);
+    }
+
+    public OperationResult<byte[]> getResult() {
+        return getChangedData ? new GetDataOperation(connection, getNode()).getResult() : null;
+    }
+
+    @Override
+    public ZooKeeperOperation createCopy() throws Exception {
+        return getClass().getConstructor(CONSTRUCTOR_ARGS).newInstance(new Object[] {connection, node, getChangedData});
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistenceChangedOperation.java Wed Aug 24 22:31:37 2011
@@ -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.zookeeper.operations;
+
+import org.apache.zookeeper.AsyncCallback.StatCallback;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * <code>DataChangedOperation</code> is an watch driven operation. It will wait
+ * for an watched event indicating that a given node has been created or
+ * deleted.
+ */
+public class ExistenceChangedOperation extends FutureEventDrivenOperation<String> {
+
+    public ExistenceChangedOperation(ZooKeeper connection, String znode) {
+        super(connection, znode, EventType.NodeCreated, EventType.NodeDeleted);
+    }
+
+    @Override
+    protected void installWatch() {
+        connection.exists(getNode(), this, new StatCallback() {
+            public void processResult(int rc, String path, Object ctx, Stat stat) {
+            }
+        }, null);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Installed exists watch");
+        }
+    }
+
+    @Override
+    public OperationResult<String> getResult() {
+        try {
+            String path = getNode();
+            Stat statistics = connection.exists(path, true);
+            return new OperationResult<String>(path, statistics);
+        } catch (Exception e) {
+            return new OperationResult<String>(e);
+        }
+    }
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistsOperation.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistsOperation.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistsOperation.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/ExistsOperation.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,63 @@
+/**
+ * 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.zookeeper.operations;
+
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * <code>ExistsOperation</code> is a basic ZooKeeper operation used to test the
+ * existence of a given node.
+ */
+public class ExistsOperation extends ZooKeeperOperation<String> {
+
+    private boolean mustExist;
+
+    public ExistsOperation(ZooKeeper connection, String node) {
+        this(connection, node, true);
+    }
+
+    public ExistsOperation(ZooKeeper connection, String node, boolean mustExist) {
+        super(connection, node);
+        this.mustExist = mustExist;
+    }
+
+    @Override
+    public OperationResult<String> getResult() {
+        try {
+            Stat statistics = connection.exists(node, true);
+            boolean ok = isOk(statistics);
+            if (LOG.isTraceEnabled()) {
+                LOG.trace(ok ? "node exists" : "node does not exist");
+            }
+            return new OperationResult<String>(node, statistics, ok);
+        } catch (Exception e) {
+            return new OperationResult<String>(e);
+        }
+    }
+
+    private boolean isOk(Stat statistics) {
+        boolean ok = false;
+        if (mustExist) {
+            ok = statistics != null;
+        } else {
+            ok = statistics == null;
+        }
+        return ok;
+    }
+
+}

Added: camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java
URL: http://svn.apache.org/viewvc/camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java?rev=1161306&view=auto
==============================================================================
--- camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java (added)
+++ camel/trunk/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/operations/FutureEventDrivenOperation.java Wed Aug 24 22:31:37 2011
@@ -0,0 +1,107 @@
+/**
+ * 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.zookeeper.operations;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import static java.lang.String.format;
+
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooKeeper;
+
+/**
+ * <code>FutureEventDrivenOperation</code> uses ZooKeepers {@link Watcher}
+ * mechanism to await specific ZooKeeper events. Typically this is used to await
+ * changes to a particular node before retrieving the change.
+ */
+public abstract class FutureEventDrivenOperation<ResultType> extends ZooKeeperOperation<ResultType> implements Watcher {
+
+    private EventType[] awaitedTypes;
+
+    private CountDownLatch waitForAnyWatchedType = new CountDownLatch(1);
+
+    private WatchedEvent event;
+
+    public FutureEventDrivenOperation(ZooKeeper connection, String node, EventType... awaitedTypes) {
+        super(connection, node);
+        this.awaitedTypes = awaitedTypes;
+    }
+
+    public void process(WatchedEvent event) {
+        this.event = event;
+        EventType received = event.getType();
+        if (LOG.isDebugEnabled()) {
+            LOG.debug(format("Recieved event of type %s for node '%s'", received, event.getPath()));
+        }
+
+        for (EventType watched : awaitedTypes) {
+            if (watched.equals(received)) {
+                result = getResult();
+                waitForAnyWatchedType.countDown();
+            }
+        }
+
+        if (LOG.isTraceEnabled() && waitForAnyWatchedType.getCount() > 0) {
+
+            StringBuilder b = new StringBuilder();
+            for (EventType type : awaitedTypes) {
+                b.append(type).append(", ");
+            }
+            if (b.length() > 0) {
+                b.setLength(b.length() - 2);
+            }
+            LOG.trace(String.format("Recieved event of type %s did not match any watched types %s", received, awaitedTypes));
+        }
+    }
+
+    public OperationResult<ResultType> get() throws InterruptedException, ExecutionException {
+        installWatch();
+        waitingThreads.add(Thread.currentThread());
+        waitForAnyWatchedType.await();
+        return result;
+    }
+
+    public OperationResult<ResultType> get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
+        installWatch();
+        waitingThreads.add(Thread.currentThread());
+        waitForAnyWatchedType.await(timeout, unit);
+        return result;
+    }
+
+    /**
+     * Install the watcher to receive {@link WatchedEvent}s. It should use the
+     * appropriate asynchronous ZooKeeper call to do this so as not to block the
+     * route from starting. Once one of the watched for types of event is
+     * received a call is made to getResult, which can use the appropriate
+     * synchronous call to retrieve the actual data.
+     */
+    protected abstract void installWatch();
+
+    public WatchedEvent getEvent() {
+        return event;
+    }
+
+    public EventType[] getWatchedForTypes() {
+        return awaitedTypes;
+    }
+
+}