camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lburgazz...@apache.org
Subject camel git commit: CAMEL-9886 : Create an etcd based RoutePolicy
Date Wed, 27 Apr 2016 15:09:16 GMT
Repository: camel
Updated Branches:
  refs/heads/master 1502688d2 -> bba76eff1


CAMEL-9886 : Create an etcd based RoutePolicy


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

Branch: refs/heads/master
Commit: bba76eff18dc8653de9f60d1fcfca90c7028bb31
Parents: 1502688
Author: lburgazzoli <lburgazzoli@gmail.com>
Authored: Wed Apr 27 17:06:24 2016 +0200
Committer: lburgazzoli <lburgazzoli@gmail.com>
Committed: Wed Apr 27 17:07:03 2016 +0200

----------------------------------------------------------------------
 .../apache/camel/component/etcd/EtcdHelper.java |  35 ++
 .../camel/component/etcd/EtcdWatchConsumer.java |  12 +-
 .../component/etcd/policy/EtcdRoutePolicy.java  | 330 +++++++++++++++++++
 .../etcd/policy/EtcdRoutePolicyMain.java        |  43 +++
 .../src/test/resources/log4j.properties         |   4 +-
 5 files changed, 412 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/bba76eff/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdHelper.java
b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdHelper.java
new file mode 100644
index 0000000..c86e1a0
--- /dev/null
+++ b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdHelper.java
@@ -0,0 +1,35 @@
+/**
+ * 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.etcd;
+
+import mousio.etcd4j.responses.EtcdErrorCode;
+import mousio.etcd4j.responses.EtcdException;
+
+public final class EtcdHelper  {
+    private static final String OUTDATED_EVENT_MSG = "requested index is outdated and cleared";
+
+    private EtcdHelper() {
+    }
+
+    public static boolean isOutdatedIndexException(EtcdException exception) {
+        if (exception.isErrorCode(EtcdErrorCode.EventIndexCleared) && exception.etcdMessage
!= null) {
+            return exception.etcdMessage.toLowerCase().contains(OUTDATED_EVENT_MSG);
+        }
+
+        return false;
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/bba76eff/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
index 9f85a0a..2a197b3 100644
--- a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
+++ b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/EtcdWatchConsumer.java
@@ -22,7 +22,6 @@ import java.util.concurrent.atomic.AtomicLong;
 
 import mousio.client.promises.ResponsePromise;
 import mousio.etcd4j.requests.EtcdKeyGetRequest;
-import mousio.etcd4j.responses.EtcdErrorCode;
 import mousio.etcd4j.responses.EtcdException;
 import mousio.etcd4j.responses.EtcdKeysResponse;
 import org.apache.camel.Exchange;
@@ -32,7 +31,6 @@ import org.slf4j.LoggerFactory;
 
 public class EtcdWatchConsumer extends AbstractEtcdConsumer implements ResponsePromise.IsSimplePromiseResponseHandler<EtcdKeysResponse>
{
     private static final Logger LOGGER = LoggerFactory.getLogger(EtcdWatchConsumer.class);
-    private static final String OUTDATED_EVENT_MSG = "requested index is outdated and cleared";
 
     private final EtcdWatchEndpoint endpoint;
     private final EtcdConfiguration configuration;
@@ -81,7 +79,7 @@ public class EtcdWatchConsumer extends AbstractEtcdConsumer implements ResponseP
             // }
             //
             // So we set the index to the one returned by the exception + 1
-            if (isOutdatedIndexException(exception)) {
+            if (EtcdHelper.isOutdatedIndexException(exception)) {
                 LOGGER.debug("Outdated index, key: {}, cause={}", getPath(), exception.etcdCause);
 
                 // We set the index to the one returned by the exception + 1.
@@ -154,12 +152,4 @@ public class EtcdWatchConsumer extends AbstractEtcdConsumer implements
ResponseP
 
         request.send().addListener(this);
     }
-
-    private boolean isOutdatedIndexException(EtcdException exception) {
-        if (exception.isErrorCode(EtcdErrorCode.EventIndexCleared) && exception.etcdMessage
!= null) {
-            return exception.etcdMessage.toLowerCase().contains(OUTDATED_EVENT_MSG);
-        }
-
-        return false;
-    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/bba76eff/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicy.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicy.java
b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicy.java
new file mode 100644
index 0000000..8c0cc12
--- /dev/null
+++ b/components/camel-etcd/src/main/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicy.java
@@ -0,0 +1,330 @@
+/**
+ * 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.etcd.policy;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import mousio.client.promises.ResponsePromise;
+import mousio.etcd4j.EtcdClient;
+import mousio.etcd4j.responses.EtcdErrorCode;
+import mousio.etcd4j.responses.EtcdException;
+import mousio.etcd4j.responses.EtcdKeysResponse;
+import org.apache.camel.Exchange;
+import org.apache.camel.NonManagedService;
+import org.apache.camel.Route;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.component.etcd.EtcdHelper;
+import org.apache.camel.support.RoutePolicySupport;
+import org.apache.camel.util.ObjectHelper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EtcdRoutePolicy extends RoutePolicySupport implements ResponsePromise.IsSimplePromiseResponseHandler<EtcdKeysResponse>,
NonManagedService {
+    private static final Logger LOGGER = LoggerFactory.getLogger(EtcdRoutePolicy.class);
+
+    private final Object lock;
+    private final EtcdClient client;
+    private final boolean managedClient;
+    private final AtomicBoolean leader;
+    private final Set<Route> suspendedRoutes;
+    private final AtomicLong index;
+
+    private String serviceName;
+    private String servicePath;
+    private int ttl;
+    private int watchTimeout;
+    private boolean shouldStopConsumer;
+
+    public EtcdRoutePolicy() {
+        this(new EtcdClient(), true);
+    }
+
+    public EtcdRoutePolicy(EtcdClient client) {
+        this(client, false);
+    }
+
+    public EtcdRoutePolicy(EtcdClient client, boolean managedClient) {
+        this.client = client;
+        this.managedClient = managedClient;
+        this.suspendedRoutes =  new HashSet<>();
+        this.leader = new AtomicBoolean(false);
+        this.lock = new Object();
+        this.index = new AtomicLong(0);
+        this.serviceName = null;
+        this.servicePath = null;
+        this.ttl = 60;
+        this.watchTimeout = ttl / 3;
+        this.shouldStopConsumer = true;
+    }
+
+    @Override
+    public void onExchangeBegin(Route route, Exchange exchange)  {
+        if (leader.get()) {
+            if (shouldStopConsumer) {
+                startConsumer(route);
+            }
+        } else {
+            if (shouldStopConsumer) {
+                stopConsumer(route);
+            }
+
+            exchange.setException(new IllegalStateException(
+                "Etcd based route policy prohibits processing exchanges, stopping route and
failing the exchange")
+            );
+        }
+    }
+
+    @Override
+    public void onStop(Route route) {
+        synchronized (lock) {
+            suspendedRoutes.remove(route);
+        }
+    }
+
+    @Override
+    public synchronized void onSuspend(Route route) {
+        synchronized (lock) {
+            suspendedRoutes.remove(route);
+        }
+    }
+
+    @Override
+    protected void doStart() throws Exception {
+        setLeader(tryTakeLeadership());
+        watch();
+
+        super.doStart();
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        if (managedClient) {
+            client.close();
+        }
+
+        super.doStop();
+    }
+
+    // *************************************************************************
+    //
+    // *************************************************************************
+
+    protected void setLeader(boolean isLeader) {
+        if (isLeader && leader.compareAndSet(false, isLeader)) {
+            LOGGER.info("Leadership taken (path={}, name={})", servicePath, serviceName);
+            startAllStoppedConsumers();
+        } else {
+            if(!leader.getAndSet(isLeader) && isLeader) {
+                LOGGER.info("Leadership lost (path={}, name={})", servicePath, serviceName);
+            }
+        }
+    }
+
+    private void startConsumer(Route route) {
+        synchronized (lock) {
+            try {
+                if (suspendedRoutes.contains(route)) {
+                    startConsumer(route.getConsumer());
+                    suspendedRoutes.remove(route);
+                }
+            } catch (Exception e) {
+                handleException(e);
+            }
+        }
+    }
+
+    private void stopConsumer(Route route) {
+        synchronized (lock) {
+            try {
+                if (!suspendedRoutes.contains(route)) {
+                    LOGGER.debug("Stopping consumer for {} ({})", route.getId(), route.getConsumer());
+                    stopConsumer(route.getConsumer());
+                    suspendedRoutes.add(route);
+                }
+            } catch (Exception e) {
+                handleException(e);
+            }
+        }
+    }
+
+    private void startAllStoppedConsumers() {
+        synchronized (lock) {
+            try {
+                for (Route route : suspendedRoutes) {
+                    LOGGER.debug("Starting consumer for {} ({})", route.getId(), route.getConsumer());
+                    startConsumer(route.getConsumer());
+                }
+
+                suspendedRoutes.clear();
+            } catch (Exception e) {
+                handleException(e);
+            }
+        }
+    }
+
+    // *************************************************************************
+    // Getter/Setters
+    // *************************************************************************
+
+    public EtcdClient getClient() {
+        return client;
+    }
+
+    public String getServiceName() {
+        return serviceName;
+    }
+
+    public void setServiceName(String serviceName) {
+        this.serviceName = serviceName;
+    }
+
+    public String getServicePath() {
+        return servicePath;
+    }
+
+    public void setServicePath(String servicePath) {
+        this.servicePath = servicePath;
+    }
+
+    public int getTtl() {
+        return ttl;
+    }
+
+    public void setTtl(int ttl) {
+        this.ttl = ttl;
+    }
+
+    public int getWatchTimeout() {
+        return watchTimeout;
+    }
+
+    public void setWatchTimeout(int watchTimeout) {
+        this.watchTimeout = watchTimeout;
+    }
+
+    public boolean isShouldStopConsumer() {
+        return shouldStopConsumer;
+    }
+
+    public void setShouldStopConsumer(boolean shouldStopConsumer) {
+        this.shouldStopConsumer = shouldStopConsumer;
+    }
+
+    // *************************************************************************
+    // Watch
+    // *************************************************************************
+
+    @Override
+    public void onResponse(ResponsePromise<EtcdKeysResponse> promise) {
+        if (!isRunAllowed()) {
+            return;
+        }
+
+        Throwable throwable = promise.getException();
+        if (throwable != null && throwable instanceof EtcdException) {
+            EtcdException exception = (EtcdException) throwable;
+            if (EtcdHelper.isOutdatedIndexException(exception)) {
+                LOGGER.debug("Outdated index, key={}, cause={}", servicePath, exception.etcdCause);
+                index.set(exception.index + 1);
+                throwable = null;
+            }
+        } else {
+            try {
+                EtcdKeysResponse response = promise.get();
+                setIndex(response);
+
+                if (response.node.value == null) {
+                    setLeader(tryTakeLeadership());
+                } else if (!ObjectHelper.equal(serviceName, response.node.value) &&
leader.get()) {
+                    // Looks like I've lost leadership
+                    setLeader(false);
+                }
+            } catch (TimeoutException e) {
+                LOGGER.debug("Timeout watching for {}", servicePath);
+                throwable = null;
+            } catch (Exception e1) {
+                throwable = e1;
+            }
+        }
+
+        if (throwable == null) {
+            try {
+                watch();
+            } catch (Exception e) {
+                throw new RuntimeCamelException(e);
+            }
+        } else {
+            throw new RuntimeCamelException(throwable);
+        }
+    }
+
+    private void watch() throws Exception {
+        if (isRunAllowed()) {
+            if (leader.get()) {
+                setIndex(client.refresh(servicePath, ttl)
+                    .send()
+                    .get()
+                );
+            }
+
+            LOGGER.debug("Watch (path={}, isLeader={}, index={})", servicePath, leader.get(),
index.get());
+
+            client.get(servicePath)
+                .waitForChange(index.get())
+                .timeout(ttl / 3, TimeUnit.SECONDS)
+                .send()
+                .addListener(this);
+        }
+    }
+
+    private boolean tryTakeLeadership() throws Exception {
+        boolean result = false;
+
+        try {
+            EtcdKeysResponse response = getClient()
+                .put(servicePath, serviceName)
+                .prevExist(false)
+                .ttl(ttl)
+                .send()
+                .get();
+
+            result = ObjectHelper.equal(serviceName, response.node.value);
+            setIndex(response);
+        } catch (EtcdException e) {
+            if (!e.isErrorCode(EtcdErrorCode.NodeExist)) {
+                throw e;
+            }
+        }
+
+        return result;
+    }
+
+    private void setIndex(EtcdKeysResponse response) {
+        if (response != null && response.node != null) {
+            index.set(response.node.modifiedIndex + 1);
+            LOGGER.debug("Index received={}, next={}", response.node.modifiedIndex, index.get());
+        } else {
+            index.set(response.etcdIndex + 1);
+            LOGGER.debug("Index received={}, next={}", response.node.modifiedIndex, index.get());
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/bba76eff/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicyMain.java
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicyMain.java
b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicyMain.java
new file mode 100644
index 0000000..149f485
--- /dev/null
+++ b/components/camel-etcd/src/test/java/org/apache/camel/component/etcd/policy/EtcdRoutePolicyMain.java
@@ -0,0 +1,43 @@
+/**
+ * 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.etcd.policy;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.main.Main;
+
+public class EtcdRoutePolicyMain {
+    public static void main(final String[] args) throws Exception {
+        Main main = new Main();
+        main.addRouteBuilder(new RouteBuilder() {
+            public void configure() {
+                EtcdRoutePolicy policy = new EtcdRoutePolicy();
+                policy.setServicePath("/camel/services/leader");
+                policy.setServiceName(args[1]);
+                policy.setTtl(15);
+
+                fromF("file:///tmp/camel?delete=true")
+                    .routeId(args[1])
+                    .routePolicy(policy)
+                    .setHeader("EtcdRouteID", constant(args[1]))
+                    .setHeader("EtcdServiceName", constant(args[0]))
+                    .to("log:org.apache.camel.component.etcd?level=INFO&showAll=true");
+            }
+        });
+
+        main.run();
+    }
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/bba76eff/components/camel-etcd/src/test/resources/log4j.properties
----------------------------------------------------------------------
diff --git a/components/camel-etcd/src/test/resources/log4j.properties b/components/camel-etcd/src/test/resources/log4j.properties
index a915605..67bab96 100644
--- a/components/camel-etcd/src/test/resources/log4j.properties
+++ b/components/camel-etcd/src/test/resources/log4j.properties
@@ -4,7 +4,9 @@
 log4j.rootLogger=INFO, file
 
 # uncomment the following line to turn on Camel debugging
-#log4j.logger.org.apache.camel=DEBUG
+log4j.logger.org.apache.camel=INFO
+log4j.logger.org.apache.camel.component.etcd=DEBUG
+log4j.logger.mousio.etcd4j=INFO
 
 # CONSOLE appender not used by default
 log4j.appender.out=org.apache.log4j.ConsoleAppender


Mime
View raw message