You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2016/10/28 08:53:04 UTC
[1/3] camel git commit: Added Content Based Router docs to Gitbook
Repository: camel
Updated Branches:
refs/heads/master 7e2db36e7 -> 4cf3a03f3
Added Content Based Router docs to Gitbook
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/495f0810
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/495f0810
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/495f0810
Branch: refs/heads/master
Commit: 495f0810816df4b2c04132ebeaf2ec991db7cae4
Parents: 7e2db36
Author: Andrea Cosentino <an...@gmail.com>
Authored: Fri Oct 28 10:45:35 2016 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Fri Oct 28 10:47:07 2016 +0200
----------------------------------------------------------------------
.../src/main/docs/content-based-router.adoc | 79 ++++++++++++++++++++
docs/user-manual/en/SUMMARY.md | 2 +
2 files changed, 81 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/495f0810/camel-core/src/main/docs/content-based-router.adoc
----------------------------------------------------------------------
diff --git a/camel-core/src/main/docs/content-based-router.adoc b/camel-core/src/main/docs/content-based-router.adoc
new file mode 100644
index 0000000..809b332
--- /dev/null
+++ b/camel-core/src/main/docs/content-based-router.adoc
@@ -0,0 +1,79 @@
+[[ContentBasedRouter-ContentBasedRouter]]
+Content Based Router
+^^^^^^^^^^^^^^^^^^^^
+
+The
+http://www.enterpriseintegrationpatterns.com/ContentBasedRouter.html[Content
+Based Router] from the link:enterprise-integration-patterns.html[EIP
+patterns] allows you to route messages to the correct destination based
+on the contents of the message exchanges.
+
+image:http://www.enterpriseintegrationpatterns.com/img/ContentBasedRouter.gif[image]
+
+The following example shows how to route a request from an input
+*seda:a* endpoint to either *seda:b*, *seda:c* or *seda:d* depending on
+the evaluation of various link:predicate.html[Predicate] expressions
+
+*Using the link:fluent-builders.html[Fluent Builders]*
+
+[source,java]
+--------------------------------------------------------
+RouteBuilder builder = new RouteBuilder() {
+ public void configure() {
+ errorHandler(deadLetterChannel("mock:error"));
+
+ from("direct:a")
+ .choice()
+ .when(header("foo").isEqualTo("bar"))
+ .to("direct:b")
+ .when(header("foo").isEqualTo("cheese"))
+ .to("direct:c")
+ .otherwise()
+ .to("direct:d");
+ }
+};
+--------------------------------------------------------
+
+TIP: See link:why-can-i-not-use-when-or-otherwise-in-a-java-camel-route.html[Why
+can I not use when or otherwise in a Java Camel route] if you have
+problems with the Java DSL, accepting using `when` or `otherwise`.
+
+
+*Using the link:spring-xml-extensions.html[Spring XML Extensions]*
+
+[source,xml]
+-------------------------------------------------------------------------------------------
+<camelContext errorHandlerRef="errorHandler" xmlns="http://camel.apache.org/schema/spring">
+ <route>
+ <from uri="direct:a"/>
+ <choice>
+ <when>
+ <xpath>$foo = 'bar'</xpath>
+ <to uri="direct:b"/>
+ </when>
+ <when>
+ <xpath>$foo = 'cheese'</xpath>
+ <to uri="direct:c"/>
+ </when>
+ <otherwise>
+ <to uri="direct:d"/>
+ </otherwise>
+ </choice>
+ </route>
+</camelContext>
+-------------------------------------------------------------------------------------------
+
+For further examples of this pattern in use you could look at the
+http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/ChoiceTest.java?view=markup[junit
+test case]
+
+[[ContentBasedRouter-UsingThisPattern]]
+Using This Pattern
+++++++++++++++++++
+
+If you would like to use this EIP Pattern then please read the
+link:getting-started.html[Getting Started], you may also find the
+link:architecture.html[Architecture] useful particularly the description
+of link:endpoint.html[Endpoint] and link:uris.html[URIs]. Then you could
+try out some of the link:examples.html[Examples] first before trying
+this pattern out.
http://git-wip-us.apache.org/repos/asf/camel/blob/495f0810/docs/user-manual/en/SUMMARY.md
----------------------------------------------------------------------
diff --git a/docs/user-manual/en/SUMMARY.md b/docs/user-manual/en/SUMMARY.md
index 2233ac5..26af756 100644
--- a/docs/user-manual/en/SUMMARY.md
+++ b/docs/user-manual/en/SUMMARY.md
@@ -92,6 +92,8 @@
* [Event Message](event-message.adoc)
* [Return Address](return-address.adoc)
* [Request Reply](request-reply.adoc)
+ * Message Routing
+ * [Content Based Router](content-based-router.adoc)
* Community
* FAQ
[3/3] camel git commit: CAMEL-10292 - camel-zookeeper fixed styling
Posted by ac...@apache.org.
CAMEL-10292 - camel-zookeeper fixed styling
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/4cf3a03f
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/4cf3a03f
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/4cf3a03f
Branch: refs/heads/master
Commit: 4cf3a03f3ae7e742dba9f4dd398df91ca696a656
Parents: 34ea6c9
Author: Paolo Antinori <pa...@redhat.com>
Authored: Fri Oct 28 10:20:19 2016 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Fri Oct 28 10:49:27 2016 +0200
----------------------------------------------------------------------
.../component/zookeeper/ZooKeeperProducer.java | 2 +-
.../zookeeper/policy/CuratorLeaderElection.java | 54 ++++++++++----------
.../policy/CuratorLeaderRoutePolicy.java | 18 ++++---
.../FailoverCuratorLeaderRoutePolicyTest.java | 16 +++---
4 files changed, 45 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/4cf3a03f/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
index 9c01632..37b236e 100644
--- a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
@@ -64,7 +64,7 @@ public class ZooKeeperProducer extends DefaultProducer {
public void process(Exchange exchange) throws Exception {
- if(connection == null){
+ if (connection == null) {
connection = this.zkm.getConnection();
}
ProductionContext context = new ProductionContext(connection, exchange);
http://git-wip-us.apache.org/repos/asf/camel/blob/4cf3a03f/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
index b0cf654..2296213 100644
--- a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
@@ -16,12 +16,20 @@
*/
package org.apache.camel.component.zookeeper.policy;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
import org.apache.camel.CamelContext;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.StatefulService;
import org.apache.camel.impl.JavaUuidGenerator;
import org.apache.camel.spi.UuidGenerator;
-import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
@@ -30,14 +38,6 @@ import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
/**
* <code>CuratorLeaderElection</code> uses the leader election capabilities of a
@@ -62,12 +62,11 @@ public class CuratorLeaderElection {
private final String candidateName;
private final Lock lock = new ReentrantLock();
private final CountDownLatch electionComplete = new CountDownLatch(1);
+ private final List<ElectionWatcher> watchers = new ArrayList<ElectionWatcher>();
private AtomicBoolean masterNode = new AtomicBoolean(false);
private volatile boolean isCandidateCreated;
private int enabledCount = 1;
private UuidGenerator uuidGenerator = new JavaUuidGenerator();
- private final List<ElectionWatcher> watchers = new ArrayList<ElectionWatcher>();
-
private LeaderSelector leaderSelector;
private CuratorFramework client;
@@ -81,9 +80,9 @@ public class CuratorLeaderElection {
this.uri = uri;
this.candidateName = createCandidateName();
- String connectionString = uri.substring( 1 + uri.indexOf(':')).split("/")[0];
+ String connectionString = uri.substring(1 + uri.indexOf(':')).split("/")[0];
String protocol = uri.substring(0, uri.indexOf(':'));
- String path = uri.replace(protocol + ":" + connectionString, "" );
+ String path = uri.replace(protocol + ":" + connectionString, "");
client = CuratorFrameworkFactory.newClient(connectionString, new ExponentialBackoffRetry(1000, 3));
client.start();
@@ -91,10 +90,19 @@ public class CuratorLeaderElection {
leaderSelector.start();
}
- public void shutdownClients(){
+ // stolen from org/apache/camel/processor/CamelInternalProcessor
+ public static boolean isCamelStopping(CamelContext context) {
+ if (context instanceof StatefulService) {
+ StatefulService ss = (StatefulService) context;
+ return ss.isStopping() || ss.isStopped();
+ }
+ return false;
+ }
+
+ public void shutdownClients() {
try {
leaderSelector.close();
- } finally{
+ } finally {
client.close();
}
}
@@ -120,7 +128,6 @@ public class CuratorLeaderElection {
return candidateName;
}
-
private void notifyElectionWatchers() {
for (ElectionWatcher watcher : watchers) {
try {
@@ -135,16 +142,7 @@ public class CuratorLeaderElection {
return watchers.add(e);
}
- // stolen from org/apache/camel/processor/CamelInternalProcessor
- public static boolean isCamelStopping(CamelContext context) {
- if (context instanceof StatefulService) {
- StatefulService ss = (StatefulService) context;
- return ss.isStopping() || ss.isStopped();
- }
- return false;
- }
-
- class CamelLeaderElectionListener extends LeaderSelectorListenerAdapter{
+ class CamelLeaderElectionListener extends LeaderSelectorListenerAdapter {
@Override
public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
@@ -153,10 +151,10 @@ public class CuratorLeaderElection {
notifyElectionWatchers();
// this is supposed to never return as long as it wants to keep its own leader status
- while(!isCamelStopping(camelContext)){
+ while (!isCamelStopping(camelContext)) {
try {
Thread.sleep(5000);
- } catch (InterruptedException e){
+ } catch (InterruptedException e) {
Thread.interrupted();
break;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/4cf3a03f/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
index 6033989..7d27eb7 100644
--- a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
@@ -5,9 +5,9 @@
* 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ * 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.
@@ -16,17 +16,19 @@
*/
package org.apache.camel.component.zookeeper.policy;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
import org.apache.camel.NonManagedService;
import org.apache.camel.Route;
import org.apache.camel.support.RoutePolicySupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.Set;
-import java.util.concurrent.CopyOnWriteArraySet;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+
/**
* <code>CuratorLeaderRoutePolicy</code> uses Apache Curator LeaderElection receipe to implement the behavior of having
http://git-wip-us.apache.org/repos/asf/camel/blob/4cf3a03f/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
index 4cc9742..9392f79 100644
--- a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
+++ b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
@@ -5,9 +5,9 @@
* 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
+ *
+ * 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.
@@ -16,6 +16,8 @@
*/
package org.apache.camel.component.zookeeper.policy;
+import java.util.concurrent.TimeUnit;
+
import org.apache.camel.CamelContext;
import org.apache.camel.ExchangePattern;
import org.apache.camel.ProducerTemplate;
@@ -28,14 +30,14 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.concurrent.TimeUnit;
import static org.hamcrest.CoreMatchers.is;
public class FailoverCuratorLeaderRoutePolicyTest extends ZooKeeperTestSupport {
- private static final Logger LOG = LoggerFactory.getLogger(FailoverCuratorLeaderRoutePolicyTest.class);
- static final String ZNODE = "/curatorleader";
+ public static final String ZNODE = "/curatorleader";
public static final String BASE_ZNODE = "/someapp";
+ private static final Logger LOG = LoggerFactory.getLogger(FailoverCuratorLeaderRoutePolicyTest.class);
+
protected CamelContext createCamelContext() throws Exception {
disableJMX();
@@ -138,6 +140,4 @@ public class FailoverCuratorLeaderRoutePolicyTest extends ZooKeeperTestSupport {
from("vm:" + routename).routePolicy(policy).id(routename).to("mock:controlled");
}
}
-
- ;
}
[2/3] camel git commit: CAMEL-10292 - camel-zookeeper added new
RoutePolicy
Posted by ac...@apache.org.
CAMEL-10292 - camel-zookeeper added new RoutePolicy
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/34ea6c98
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/34ea6c98
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/34ea6c98
Branch: refs/heads/master
Commit: 34ea6c98e37cbcd65a725f1e34d0f27ec0e56dd3
Parents: 495f081
Author: Paolo Antinori <pa...@redhat.com>
Authored: Thu Oct 27 17:36:36 2016 +0200
Committer: Andrea Cosentino <an...@gmail.com>
Committed: Fri Oct 28 10:49:27 2016 +0200
----------------------------------------------------------------------
components/camel-zookeeper/pom.xml | 15 ++
.../component/zookeeper/ZooKeeperProducer.java | 3 +
.../zookeeper/policy/CuratorLeaderElection.java | 171 +++++++++++++++++++
.../policy/CuratorLeaderRoutePolicy.java | 167 ++++++++++++++++++
.../FailoverCuratorLeaderRoutePolicyTest.java | 143 ++++++++++++++++
.../src/test/resources/log4j2.properties | 10 +-
parent/pom.xml | 1 +
7 files changed, 509 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/34ea6c98/components/camel-zookeeper/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/pom.xml b/components/camel-zookeeper/pom.xml
index 689c064..2021d2c 100644
--- a/components/camel-zookeeper/pom.xml
+++ b/components/camel-zookeeper/pom.xml
@@ -44,6 +44,7 @@
<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>
@@ -66,6 +67,20 @@
<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>
+
+
+
<!-- test dependencies -->
<dependency>
<groupId>org.apache.camel</groupId>
http://git-wip-us.apache.org/repos/asf/camel/blob/34ea6c98/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
index 449a138..9c01632 100644
--- a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/ZooKeeperProducer.java
@@ -64,6 +64,9 @@ public class ZooKeeperProducer extends DefaultProducer {
public void process(Exchange exchange) throws Exception {
+ if(connection == null){
+ connection = this.zkm.getConnection();
+ }
ProductionContext context = new ProductionContext(connection, exchange);
String operation = exchange.getIn().getHeader(ZooKeeperMessage.ZOOKEEPER_OPERATION, String.class);
http://git-wip-us.apache.org/repos/asf/camel/blob/34ea6c98/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
new file mode 100644
index 0000000..b0cf654
--- /dev/null
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderElection.java
@@ -0,0 +1,171 @@
+/**
+ * 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.policy;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.StatefulService;
+import org.apache.camel.impl.JavaUuidGenerator;
+import org.apache.camel.spi.UuidGenerator;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.recipes.leader.LeaderSelector;
+import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * <code>CuratorLeaderElection</code> uses the leader election capabilities of a
+ * ZooKeeper cluster to control which nodes are enabled. It is typically used in
+ * fail-over scenarios controlling identical instances of an application across
+ * a cluster of Camel based servers. <p> The election is configured with a single
+ * server that should be marked as master.
+ * <p> All instances of the election must also be configured with the same path on the ZooKeeper
+ * cluster where the election will be carried out. It is good practice for this
+ * to indicate the application e.g. <tt>/someapplication/someroute/</tt> note
+ * that these nodes should exist before using the election. <p> See <a
+ * href="http://hadoop.apache.org/zookeeper/docs/current/recipes.html#sc_leaderElection">
+ * for more on how Leader election</a> is archived with ZooKeeper.
+ */
+public class CuratorLeaderElection {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CuratorLeaderElection.class);
+ private final ProducerTemplate producerTemplate;
+ private final CamelContext camelContext;
+ private final String uri;
+
+ private final String candidateName;
+ private final Lock lock = new ReentrantLock();
+ private final CountDownLatch electionComplete = new CountDownLatch(1);
+ private AtomicBoolean masterNode = new AtomicBoolean(false);
+ private volatile boolean isCandidateCreated;
+ private int enabledCount = 1;
+ private UuidGenerator uuidGenerator = new JavaUuidGenerator();
+ private final List<ElectionWatcher> watchers = new ArrayList<ElectionWatcher>();
+
+ private LeaderSelector leaderSelector;
+ private CuratorFramework client;
+
+ public CuratorLeaderElection(CamelContext camelContext, String uri) {
+ this(camelContext.createProducerTemplate(), camelContext, uri);
+ }
+
+ public CuratorLeaderElection(ProducerTemplate producerTemplate, CamelContext camelContext, String uri) {
+ this.camelContext = camelContext;
+ this.producerTemplate = producerTemplate;
+ this.uri = uri;
+ this.candidateName = createCandidateName();
+
+ String connectionString = uri.substring( 1 + uri.indexOf(':')).split("/")[0];
+ String protocol = uri.substring(0, uri.indexOf(':'));
+ String path = uri.replace(protocol + ":" + connectionString, "" );
+ client = CuratorFrameworkFactory.newClient(connectionString, new ExponentialBackoffRetry(1000, 3));
+ client.start();
+
+ leaderSelector = new LeaderSelector(client, path, new CamelLeaderElectionListener());
+ leaderSelector.start();
+ }
+
+ public void shutdownClients(){
+ try {
+ leaderSelector.close();
+ } finally{
+ client.close();
+ }
+ }
+
+ public boolean isMaster() {
+ return masterNode.get();
+ }
+
+ private String createCandidateName() {
+ StringBuilder builder = new StringBuilder();
+ try {
+ /* UUID would be enough, also using hostname for human readability */
+ builder.append(InetAddress.getLocalHost().getCanonicalHostName());
+ } catch (UnknownHostException ex) {
+ LOG.warn("Failed to get the local hostname.", ex);
+ builder.append("unknown-host");
+ }
+ builder.append("-").append(uuidGenerator.generateUuid());
+ return builder.toString();
+ }
+
+ public String getCandidateName() {
+ return candidateName;
+ }
+
+
+ private void notifyElectionWatchers() {
+ for (ElectionWatcher watcher : watchers) {
+ try {
+ watcher.electionResultChanged();
+ } catch (Exception e) {
+ LOG.warn("Election watcher " + watcher + " of type " + watcher.getClass() + " threw an exception.", e);
+ }
+ }
+ }
+
+ public boolean addElectionWatcher(ElectionWatcher e) {
+ return watchers.add(e);
+ }
+
+ // stolen from org/apache/camel/processor/CamelInternalProcessor
+ public static boolean isCamelStopping(CamelContext context) {
+ if (context instanceof StatefulService) {
+ StatefulService ss = (StatefulService) context;
+ return ss.isStopping() || ss.isStopped();
+ }
+ return false;
+ }
+
+ class CamelLeaderElectionListener extends LeaderSelectorListenerAdapter{
+
+ @Override
+ public void takeLeadership(CuratorFramework curatorFramework) throws Exception {
+ masterNode.set(true);
+ LOG.info("{} is now leader", getCandidateName());
+ notifyElectionWatchers();
+
+ // this is supposed to never return as long as it wants to keep its own leader status
+ while(!isCamelStopping(camelContext)){
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e){
+ Thread.interrupted();
+ break;
+ }
+ }
+ masterNode.set(false);
+ LOG.info("{} has given up its own leadership", getCandidateName());
+ notifyElectionWatchers();
+ }
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/camel/blob/34ea6c98/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
new file mode 100644
index 0000000..6033989
--- /dev/null
+++ b/components/camel-zookeeper/src/main/java/org/apache/camel/component/zookeeper/policy/CuratorLeaderRoutePolicy.java
@@ -0,0 +1,167 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.policy;
+
+import org.apache.camel.NonManagedService;
+import org.apache.camel.Route;
+import org.apache.camel.support.RoutePolicySupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * <code>CuratorLeaderRoutePolicy</code> uses Apache Curator LeaderElection receipe to implement the behavior of having
+ * at max 1 instance of a route, controlled by a specific policy, running. It is typically used in
+ * fail-over scenarios controlling identical instances of a route across a
+ * cluster of Camel based servers.
+ * <p>
+ * The policy affects the normal startup lifecycle of CamelContext and Routes, automatically set autoStart property of
+ * routes controlled by this policy to false.
+ * After Curator receipe identiffies the current Policy instance as the Leader between a set of clients that are
+ * competing for the role, it will start the route, and only at that moment the route will start its business.
+ * This specific behavior is designed to avoid scenarios where such a policy would kick in only after a route had
+ * already been started, with the risk, for consumers for example, that some source event might have already been
+ * consumed.
+ * <p>
+ * All instances of the policy must also be configured with the same path on the
+ * ZooKeeper cluster where the election will be carried out. It is good practice
+ * for this to indicate the application e.g. <tt>/someapplication/someroute/</tt> note
+ * that these nodes should exist before using the policy.
+ * <p>
+ * See <a href="http://hadoop.apache.org/zookeeper/docs/current/recipes.html#sc_leaderElection">
+ * for more on how Leader election</a> is archived with ZooKeeper.
+ */
+public class CuratorLeaderRoutePolicy extends RoutePolicySupport implements ElectionWatcher, NonManagedService {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CuratorLeaderRoutePolicy.class);
+ private final String uri;
+ private final Lock lock = new ReentrantLock();
+ private final Set<Route> suspendedRoutes = new CopyOnWriteArraySet<Route>();
+ private final AtomicBoolean shouldProcessExchanges = new AtomicBoolean();
+ private volatile boolean shouldStopRoute = true;
+
+ private final Lock electionLock = new ReentrantLock();
+ private CuratorLeaderElection election;
+
+ public CuratorLeaderRoutePolicy(String uri) {
+ this.uri = uri;
+ }
+
+ public CuratorLeaderRoutePolicy(CuratorLeaderElection election) {
+ this.election = election;
+ this.uri = null;
+ }
+
+ @Override
+ public void onInit(Route route) {
+ ensureElectionIsCreated(route);
+ LOG.info("Route managed by {}. Setting route {} AutoStartup flag to false.", this.getClass(), route.getId());
+ route.getRouteContext().getRoute().setAutoStartup("false");
+ ensureElectionIsCreated(route);
+
+ if (election.isMaster()) {
+ if (shouldStopRoute) {
+ startManagedRoute(route);
+ }
+ } else {
+ if (shouldStopRoute) {
+ stopManagedRoute(route);
+ }
+ }
+ }
+
+ private void ensureElectionIsCreated(Route route) {
+ if (election == null) {
+ electionLock.lock();
+ try {
+ if (election == null) { // re-test
+ election = new CuratorLeaderElection(route.getRouteContext().getCamelContext(), uri);
+ election.addElectionWatcher(this);
+ }
+ } finally {
+ electionLock.unlock();
+ }
+ }
+ }
+
+ private void startManagedRoute(Route route) {
+ try {
+ lock.lock();
+ if (suspendedRoutes.contains(route)) {
+ startRoute(route);
+ suspendedRoutes.remove(route);
+ }
+ } catch (Exception e) {
+ handleException(e);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void stopManagedRoute(Route route) {
+ try {
+ lock.lock();
+ // check that we should still suspend once the lock is acquired
+ if (!suspendedRoutes.contains(route) && !shouldProcessExchanges.get()) {
+ stopRoute(route);
+ suspendedRoutes.add(route);
+ }
+ } catch (Exception e) {
+ handleException(e);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void electionResultChanged() {
+ if (election.isMaster()) {
+ startAllStoppedRoutes();
+ }
+ }
+
+ private void startAllStoppedRoutes() {
+ try {
+ lock.lock();
+ if (!suspendedRoutes.isEmpty()) {
+ if (log.isDebugEnabled()) {
+ log.debug("{} route(s) have been stopped previously by policy, restarting.", suspendedRoutes.size());
+ }
+ for (Route suspended : suspendedRoutes) {
+ log.debug("Starting route {}.", suspended.getId());
+ startRoute(suspended);
+ }
+ suspendedRoutes.clear();
+ }
+
+ } catch (Exception e) {
+ handleException(e);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ election.shutdownClients();
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/34ea6c98/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
new file mode 100644
index 0000000..4cc9742
--- /dev/null
+++ b/components/camel-zookeeper/src/test/java/org/apache/camel/component/zookeeper/policy/FailoverCuratorLeaderRoutePolicyTest.java
@@ -0,0 +1,143 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.policy;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ExchangePattern;
+import org.apache.camel.ProducerTemplate;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.zookeeper.ZooKeeperTestSupport;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.hamcrest.CoreMatchers.is;
+
+public class FailoverCuratorLeaderRoutePolicyTest extends ZooKeeperTestSupport {
+ private static final Logger LOG = LoggerFactory.getLogger(FailoverCuratorLeaderRoutePolicyTest.class);
+ static final String ZNODE = "/curatorleader";
+ public static final String BASE_ZNODE = "/someapp";
+
+ protected CamelContext createCamelContext() throws Exception {
+ disableJMX();
+ return super.createCamelContext();
+ }
+
+ @Test
+ public void masterSlaveScenarioContolledByPolicy() throws Exception {
+ ZookeeperPolicyEnforcedContext master = createEnforcedContext("master");
+ ZookeeperPolicyEnforcedContext slave = createEnforcedContext("slave");
+ Thread.sleep(5000);
+ // Send messages to the master and the slave.
+ // The route is enabled in the master and gets through, but that sent to
+ // the slave context is rejected.
+ master.sendMessageToEnforcedRoute("message for master", 1);
+ slave.sendMessageToEnforcedRoute("message for slave", 0);
+
+ // trigger failover by killing the master... then assert that the slave
+ // can now receive messages.
+ master.shutdown();
+ slave.sendMessageToEnforcedRoute("second message for slave", 1);
+ slave.shutdown();
+ }
+
+
+ @Test
+ public void ensureRoutesDoNotStartBeforeElection() throws Exception {
+ DefaultCamelContext context = new DefaultCamelContext();
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ CuratorLeaderRoutePolicy policy = new CuratorLeaderRoutePolicy("zookeeper:localhost:" + getServerPort() + BASE_ZNODE + ZNODE + 2);
+ from("timer://foo?fixedRate=true&period=5").routePolicy(policy).id("single_route").autoStartup(true).to("mock:controlled");
+ }
+ });
+ context.start();
+ // this check verifies that a route marked as autostartable is not started automatically. It will be the policy responsibility to eventually start it.
+ assertThat(context.getRouteStatus("single_route").isStarted(), is(false));
+ assertThat(context.getRouteStatus("single_route").isStarting(), is(false));
+
+ context.shutdown();
+ }
+
+ private static class ZookeeperPolicyEnforcedContext {
+ private CamelContext controlledContext;
+ private ProducerTemplate template;
+ private MockEndpoint mock;
+ private String routename;
+
+ ZookeeperPolicyEnforcedContext(String name) throws Exception {
+ controlledContext = new DefaultCamelContext();
+ routename = name;
+ template = controlledContext.createProducerTemplate();
+ mock = controlledContext.getEndpoint("mock:controlled", MockEndpoint.class);
+ controlledContext.addRoutes(new FailoverRoute(name));
+ controlledContext.start();
+ }
+
+ public void sendMessageToEnforcedRoute(String message, int expected) throws InterruptedException {
+ mock.expectedMessageCount(expected);
+ try {
+ template.sendBody("vm:" + routename, ExchangePattern.InOut, message);
+ } catch (Exception e) {
+ if (expected > 0) {
+ LOG.error(e.getMessage(), e);
+ fail("Expected messages...");
+ }
+ }
+ mock.await(2, TimeUnit.SECONDS);
+ mock.assertIsSatisfied(2000);
+ }
+
+ public void shutdown() throws Exception {
+ LogFactory.getLog(getClass()).debug("stopping");
+ controlledContext.stop();
+ LogFactory.getLog(getClass()).debug("stopped");
+ }
+ }
+
+ private ZookeeperPolicyEnforcedContext createEnforcedContext(String name) throws Exception, InterruptedException {
+ ZookeeperPolicyEnforcedContext context = new ZookeeperPolicyEnforcedContext(name);
+ delay(1000);
+ return context;
+ }
+
+ public static class FailoverRoute extends RouteBuilder {
+
+ private String routename;
+
+ public FailoverRoute(String routename) {
+ // need names as if we use the same direct ep name in two contexts
+ // in the same vm shutting down one context shuts the endpoint for
+ // both.
+ this.routename = routename;
+ }
+
+ public void configure() throws Exception {
+ CuratorLeaderRoutePolicy policy = new CuratorLeaderRoutePolicy("zookeeper:localhost:" + getServerPort() + BASE_ZNODE + ZNODE);
+ from("vm:" + routename).routePolicy(policy).id(routename).to("mock:controlled");
+ }
+ }
+
+ ;
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/34ea6c98/components/camel-zookeeper/src/test/resources/log4j2.properties
----------------------------------------------------------------------
diff --git a/components/camel-zookeeper/src/test/resources/log4j2.properties b/components/camel-zookeeper/src/test/resources/log4j2.properties
index c1843bb..81447cc 100644
--- a/components/camel-zookeeper/src/test/resources/log4j2.properties
+++ b/components/camel-zookeeper/src/test/resources/log4j2.properties
@@ -25,8 +25,16 @@ appender.out.name = out
appender.out.layout.type = PatternLayout
appender.out.layout.pattern = [%t] %c{1} %-5p %m%n
logger.zookeeper.name = org.apache.zookeeper
-logger.zookeeper.level = WARN
+logger.zookeeper.level = INFO
+logger.camel-zookeeper.name = org.apache.camel.component.zookeeper
+logger.camel-zookeeper.level = INFO
+logger.camel-support.name = org.apache.camel.support
+logger.camel-support.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/34ea6c98/parent/pom.xml
----------------------------------------------------------------------
diff --git a/parent/pom.xml b/parent/pom.xml
index bd3f1bf..34d9d8b 100644
--- a/parent/pom.xml
+++ b/parent/pom.xml
@@ -147,6 +147,7 @@
<consul-client-version>0.12.7</consul-client-version>
<consul-client-bundle-version>0.12.7_1</consul-client-bundle-version>
<cobertura-maven-plugin-version>2.7</cobertura-maven-plugin-version>
+ <curator-version>2.11.0</curator-version>
<cxf-version>3.1.7</cxf-version>
<cxf-version-range>[3.0,4.0)</cxf-version-range>
<cxf-xjc-plugin-version>3.0.5</cxf-xjc-plugin-version>