You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2017/01/20 09:36:12 UTC
[1/5] camel git commit: Add unit tests for
ThrottlingExceptionRoutePolicy and include in LC strategy
Repository: camel
Updated Branches:
refs/heads/master a4f82a2c9 -> 0243960c5
Add unit tests for ThrottlingExceptionRoutePolicy and include in LC
strategy
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/5351baa5
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/5351baa5
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/5351baa5
Branch: refs/heads/master
Commit: 5351baa5b2957a75e74432490ad29641e6f169f7
Parents: 69b716b
Author: CodeSmell <mb...@gmail.com>
Authored: Wed Jan 18 13:59:23 2017 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jan 20 09:44:37 2017 +0100
----------------------------------------------------------------------
...agedThrottlingExceptionRoutePolicyMBean.java | 41 ++---
.../impl/ThrottlingExceptionRoutePolicy.java | 4 +-
.../DefaultManagementLifecycleStrategy.java | 4 +
.../ManagedThrottlingExceptionRoutePolicy.java | 32 ++--
.../management/ManagedRoutePolicyTest.java | 77 ---------
...nagedThrottlingExceptionRoutePolicyTest.java | 169 +++++++++++++++++++
...anagedThrottlingInflightRoutePolicyTest.java | 77 +++++++++
7 files changed, 293 insertions(+), 111 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
index 86e394c..a83b690 100644
--- a/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
+++ b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
@@ -17,39 +17,40 @@
package org.apache.camel.api.management.mbean;
import org.apache.camel.api.management.ManagedAttribute;
+import org.apache.camel.api.management.ManagedOperation;
public interface ManagedThrottlingExceptionRoutePolicyMBean extends ManagedServiceMBean {
- @ManagedAttribute(description = "how long to wait before moving open circuit to half open")
- long getHalfOpenAfter();
+ @ManagedAttribute(description = "How long to wait before moving open circuit to half open")
+ Long getHalfOpenAfter();
- @ManagedAttribute(description = "how long to wait before moving open circuit to half open")
- void setHalfOpenAfter(long milliseconds);
+ @ManagedAttribute(description = "How long to wait before moving open circuit to half open")
+ void setHalfOpenAfter(Long milliseconds);
- @ManagedAttribute(description = "the range of time that failures should occur within")
- long getFailureWindow();
+ @ManagedAttribute(description = "The range of time that failures should occur within")
+ Long getFailureWindow();
- @ManagedAttribute(description = "the range of time that failures should occur within")
- void setFailureWindow(long milliseconds);
+ @ManagedAttribute(description = "The range of time that failures should occur within")
+ void setFailureWindow(Long milliseconds);
- @ManagedAttribute(description = "number of failures before opening circuit")
- int getFailureThreshold();
+ @ManagedAttribute(description = "Number of failures before opening circuit")
+ Integer getFailureThreshold();
- @ManagedAttribute(description = "number of failures before opening circuit")
- void setFailureThreshold(int numberOfFailures);
+ @ManagedAttribute(description = "Number of failures before opening circuit")
+ void setFailureThreshold(Integer numberOfFailures);
- @ManagedAttribute(description = "State")
+ @ManagedOperation(description = "The current state of the circuit")
String currentState();
@ManagedAttribute(description = "The half open handler registered (if any)")
- String hasHalfOpenHandler();
+ String getHalfOpenHandlerName();
- @ManagedAttribute(description = "the number of failures caught")
- int currentFailures();
+ @ManagedAttribute(description = "The number of failures caught")
+ Integer getCurrentFailures();
- @ManagedAttribute(description = "number of ms since the last failure was recorded")
- long getLastFailure();
+ @ManagedAttribute(description = "Number of ms since the last failure was recorded")
+ Long getLastFailure();
- @ManagedAttribute(description = "number ms since the circuit was opened")
- long getOpenAt();
+ @ManagedAttribute(description = "Number ms since the circuit was opened")
+ Long getOpenAt();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java b/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
index 34b755a..e23e1bc 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
@@ -257,9 +257,9 @@ public class ThrottlingExceptionRoutePolicy extends RoutePolicySupport implement
int num = state.get();
String routeState = stateAsString(num);
if (failures.get() > 0) {
- return String.format("*** State %s, failures %d, last failure %d ms ago", routeState, failures.get(), System.currentTimeMillis() - lastFailure);
+ return String.format("State %s, failures %d, last failure %d ms ago", routeState, failures.get(), System.currentTimeMillis() - lastFailure);
} else {
- return String.format("*** State %s, failures %d", routeState, failures.get());
+ return String.format("State %s, failures %d", routeState, failures.get());
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java b/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
index 6a2b58d..7c3784e 100644
--- a/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
+++ b/camel-core/src/main/java/org/apache/camel/management/DefaultManagementLifecycleStrategy.java
@@ -52,6 +52,7 @@ import org.apache.camel.impl.DefaultEndpointRegistry;
import org.apache.camel.impl.DefaultTransformerRegistry;
import org.apache.camel.impl.EventDrivenConsumerRoute;
import org.apache.camel.impl.ProducerCache;
+import org.apache.camel.impl.ThrottlingExceptionRoutePolicy;
import org.apache.camel.impl.ThrottlingInflightRoutePolicy;
import org.apache.camel.management.mbean.ManagedAsyncProcessorAwaitManager;
import org.apache.camel.management.mbean.ManagedBacklogDebugger;
@@ -67,6 +68,7 @@ import org.apache.camel.management.mbean.ManagedRoute;
import org.apache.camel.management.mbean.ManagedRuntimeEndpointRegistry;
import org.apache.camel.management.mbean.ManagedService;
import org.apache.camel.management.mbean.ManagedStreamCachingStrategy;
+import org.apache.camel.management.mbean.ManagedThrottlingExceptionRoutePolicy;
import org.apache.camel.management.mbean.ManagedThrottlingInflightRoutePolicy;
import org.apache.camel.management.mbean.ManagedTracer;
import org.apache.camel.management.mbean.ManagedTransformerRegistry;
@@ -469,6 +471,8 @@ public class DefaultManagementLifecycleStrategy extends ServiceSupport implement
return getManagedObjectForProcessor(context, (Processor) service, route);
} else if (service instanceof ThrottlingInflightRoutePolicy) {
answer = new ManagedThrottlingInflightRoutePolicy(context, (ThrottlingInflightRoutePolicy) service);
+ } else if (service instanceof ThrottlingExceptionRoutePolicy) {
+ answer = new ManagedThrottlingExceptionRoutePolicy(context, (ThrottlingExceptionRoutePolicy) service);
} else if (service instanceof ConsumerCache) {
answer = new ManagedConsumerCache(context, (ConsumerCache) service);
} else if (service instanceof ProducerCache) {
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
index bd2af1f..8030eae 100644
--- a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
+++ b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
@@ -37,32 +37,32 @@ public class ManagedThrottlingExceptionRoutePolicy extends ManagedService implem
}
@Override
- public long getHalfOpenAfter() {
+ public Long getHalfOpenAfter() {
return getPolicy().getHalfOpenAfter();
}
@Override
- public void setHalfOpenAfter(long milliseconds) {
+ public void setHalfOpenAfter(Long milliseconds) {
getPolicy().setHalfOpenAfter(milliseconds);
}
@Override
- public long getFailureWindow() {
+ public Long getFailureWindow() {
return getPolicy().getFailureWindow();
}
@Override
- public void setFailureWindow(long milliseconds) {
+ public void setFailureWindow(Long milliseconds) {
getPolicy().setFailureWindow(milliseconds);
}
@Override
- public int getFailureThreshold() {
+ public Integer getFailureThreshold() {
return getPolicy().getFailureThreshold();
}
@Override
- public void setFailureThreshold(int numberOfFailures) {
+ public void setFailureThreshold(Integer numberOfFailures) {
getPolicy().setFailureThreshold(numberOfFailures);
}
@@ -72,7 +72,7 @@ public class ManagedThrottlingExceptionRoutePolicy extends ManagedService implem
}
@Override
- public String hasHalfOpenHandler() {
+ public String getHalfOpenHandlerName() {
ThrottlingExceptionHalfOpenHandler obj = getPolicy().getHalfOpenHandler();
if (obj != null) {
return obj.getClass().getSimpleName();
@@ -82,18 +82,26 @@ public class ManagedThrottlingExceptionRoutePolicy extends ManagedService implem
}
@Override
- public int currentFailures() {
+ public Integer getCurrentFailures() {
return getPolicy().getFailures();
}
@Override
- public long getLastFailure() {
- return System.currentTimeMillis() - getPolicy().getLastFailure();
+ public Long getLastFailure() {
+ if (getPolicy().getLastFailure() == 0) {
+ return 0L;
+ } else {
+ return System.currentTimeMillis() - getPolicy().getLastFailure();
+ }
}
@Override
- public long getOpenAt() {
- return System.currentTimeMillis() - getPolicy().getOpenedAt();
+ public Long getOpenAt() {
+ if (getPolicy().getOpenedAt() == 0) {
+ return 0L;
+ } else {
+ return System.currentTimeMillis() - getPolicy().getOpenedAt();
+ }
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/test/java/org/apache/camel/management/ManagedRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedRoutePolicyTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedRoutePolicyTest.java
deleted file mode 100644
index 62e1987..0000000
--- a/camel-core/src/test/java/org/apache/camel/management/ManagedRoutePolicyTest.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/**
- * 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.management;
-
-import java.util.Set;
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
-
-import org.apache.camel.ServiceStatus;
-import org.apache.camel.builder.RouteBuilder;
-import org.apache.camel.impl.ThrottlingInflightRoutePolicy;
-
-/**
- * @version
- */
-public class ManagedRoutePolicyTest extends ManagementTestSupport {
-
- public void testRoutes() throws Exception {
- // JMX tests dont work well on AIX CI servers (hangs them)
- if (isPlatform("aix")) {
- return;
- }
-
- MBeanServer mbeanServer = getMBeanServer();
-
- Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=routes,*"), null);
- assertEquals(1, set.size());
-
- ObjectName on = set.iterator().next();
-
- boolean registered = mbeanServer.isRegistered(on);
- assertEquals("Should be registered", true, registered);
-
- String uri = (String) mbeanServer.getAttribute(on, "EndpointUri");
- // the route has this starting endpoint uri
- assertEquals("direct://start", uri);
-
- Integer val = (Integer) mbeanServer.getAttribute(on, "InflightExchanges");
- // the route has no inflight exchanges
- assertEquals(0, val.intValue());
-
- // should be started
- String state = (String) mbeanServer.getAttribute(on, "State");
- assertEquals("Should be started", ServiceStatus.Started.name(), state);
-
- // should have route policy
- String policy = (String) mbeanServer.getAttribute(on, "RoutePolicyList");
- assertNotNull(policy);
- assertTrue("Should be a throttling, was: " + policy, policy.startsWith("ThrottlingInflightRoutePolicy"));
- }
-
- @Override
- protected RouteBuilder createRouteBuilder() throws Exception {
- return new RouteBuilder() {
- @Override
- public void configure() throws Exception {
- from("direct:start").routePolicy(new ThrottlingInflightRoutePolicy())
- .to("log:foo").to("mock:result");
- }
- };
- }
-
-}
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
new file mode 100644
index 0000000..0d41938
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
@@ -0,0 +1,169 @@
+/**
+ * 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.management;
+
+import java.util.Set;
+
+import javax.management.JMX;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.api.management.mbean.ManagedThrottlingExceptionRoutePolicyMBean;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.ThrottlingExceptionHalfOpenHandler;
+import org.apache.camel.impl.ThrottlingExceptionRoutePolicy;
+import org.junit.Test;
+
+public class ManagedThrottlingExceptionRoutePolicyTest extends ManagementTestSupport {
+
+ @Test
+ public void testRoutes() throws Exception {
+ // JMX tests dont work well on AIX CI servers (hangs them)
+ if (isPlatform("aix")) {
+ return;
+ }
+
+ MBeanServer mbeanServer = getMBeanServer();
+
+ // get the Camel route
+ Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=routes,*"), null);
+ assertEquals(1, set.size());
+ ObjectName on = set.iterator().next();
+ boolean registered = mbeanServer.isRegistered(on);
+ assertEquals("Should be registered", true, registered);
+
+ // check the starting endpoint uri
+ String uri = (String) mbeanServer.getAttribute(on, "EndpointUri");
+ assertEquals("direct://start", uri);
+
+ // should be started
+ String state = (String) mbeanServer.getAttribute(on, "State");
+ assertEquals(ServiceStatus.Started.name(), state);
+
+ // should have ThrottlingExceptionRoutePolicy route policy
+ String policy = (String) mbeanServer.getAttribute(on, "RoutePolicyList");
+ assertNotNull(policy);
+ assertTrue(policy.startsWith("ThrottlingExceptionRoutePolicy"));
+
+ // get the RoutePolicy
+ String mbeanName = String.format("org.apache.camel:context=camel-1,name=%s,type=services", policy);
+ set = mbeanServer.queryNames(new ObjectName(mbeanName), null);
+ assertEquals(1, set.size());
+ on = set.iterator().next();
+ assertTrue(mbeanServer.isRegistered(on));
+
+ // the route has no failures
+ String myType = (String) mbeanServer.getAttribute(on, "ServiceType");
+ assertEquals("ThrottlingExceptionRoutePolicy", myType);
+
+ ManagedThrottlingExceptionRoutePolicyMBean proxy = JMX.newMBeanProxy(mbeanServer, on, ManagedThrottlingExceptionRoutePolicyMBean.class);
+ assertNotNull(proxy);
+
+ // state should be closed w/ no failures
+ String myState = proxy.currentState();
+ assertEquals("State closed, failures 0", myState);
+
+ // the route has no failures
+ Integer val = proxy.getCurrentFailures();
+ assertEquals(0, val.intValue());
+
+ // the route has no failures
+ Long lastFail = proxy.getLastFailure();
+ assertEquals(0L, lastFail.longValue());
+
+ // the route is closed
+ Long openAt = proxy.getOpenAt();
+ assertEquals(0L, openAt.longValue());
+
+ // the route has a handler
+ String handlerClass = proxy.getHalfOpenHandlerName();
+ assertEquals("DummyHandler", handlerClass);
+
+ // values set during construction of class
+ Integer threshold = proxy.getFailureThreshold();
+ assertEquals(10, threshold.intValue());
+
+ Long window = proxy.getFailureWindow();
+ assertEquals(1000L, window.longValue());
+
+ Long halfOpenAfter = proxy.getHalfOpenAfter();
+ assertEquals(5000L, halfOpenAfter.longValue());
+
+ // change value
+ proxy.setHalfOpenAfter(10000L);
+ halfOpenAfter = proxy.getHalfOpenAfter();
+ assertEquals(10000L, halfOpenAfter.longValue());
+
+ try {
+ getMockEndpoint("mock:result").expectedMessageCount(0);
+ template.sendBody("direct:start", "Hello World");
+ assertMockEndpointsSatisfied();
+ } catch (Exception e) {
+ // expected
+ }
+
+ // state should be closed w/ no failures
+ myState = proxy.currentState();
+ assertTrue(myState.contains("State closed, failures 1, last failure"));
+
+ // the route has no failures
+ val = proxy.getCurrentFailures();
+ assertEquals(1, val.intValue());
+
+ // the route has no failures
+ lastFail = proxy.getLastFailure();
+ assertTrue(lastFail.longValue() > 0);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ ThrottlingExceptionRoutePolicy policy = new ThrottlingExceptionRoutePolicy(10, 1000, 5000, null);
+ policy.setHalfOpenHandler(new DummyHandler());
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start").routeId("testRoute")
+ .routePolicy(policy)
+ .to("log:foo")
+ .process(new BoomProcess())
+ .to("mock:result");
+ }
+ };
+ }
+
+ class BoomProcess implements Processor {
+
+ @Override
+ public void process(Exchange exchange) throws Exception {
+ throw new RuntimeException("boom!");
+ }
+
+ }
+
+ class DummyHandler implements ThrottlingExceptionHalfOpenHandler {
+
+ @Override
+ public boolean isReadyToBeClosed() {
+ return false;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/5351baa5/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingInflightRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingInflightRoutePolicyTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingInflightRoutePolicyTest.java
new file mode 100644
index 0000000..bab3a3c
--- /dev/null
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingInflightRoutePolicyTest.java
@@ -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.management;
+
+import java.util.Set;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.ThrottlingInflightRoutePolicy;
+
+/**
+ * @version
+ */
+public class ManagedThrottlingInflightRoutePolicyTest extends ManagementTestSupport {
+
+ public void testRoutes() throws Exception {
+ // JMX tests dont work well on AIX CI servers (hangs them)
+ if (isPlatform("aix")) {
+ return;
+ }
+
+ MBeanServer mbeanServer = getMBeanServer();
+
+ Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=routes,*"), null);
+ assertEquals(1, set.size());
+
+ ObjectName on = set.iterator().next();
+
+ boolean registered = mbeanServer.isRegistered(on);
+ assertEquals("Should be registered", true, registered);
+
+ String uri = (String) mbeanServer.getAttribute(on, "EndpointUri");
+ // the route has this starting endpoint uri
+ assertEquals("direct://start", uri);
+
+ Integer val = (Integer) mbeanServer.getAttribute(on, "InflightExchanges");
+ // the route has no inflight exchanges
+ assertEquals(0, val.intValue());
+
+ // should be started
+ String state = (String) mbeanServer.getAttribute(on, "State");
+ assertEquals("Should be started", ServiceStatus.Started.name(), state);
+
+ // should have route policy
+ String policy = (String) mbeanServer.getAttribute(on, "RoutePolicyList");
+ assertNotNull(policy);
+ assertTrue("Should be a throttling, was: " + policy, policy.startsWith("ThrottlingInflightRoutePolicy"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("direct:start").routePolicy(new ThrottlingInflightRoutePolicy())
+ .to("log:foo").to("mock:result");
+ }
+ };
+ }
+
+}
[5/5] camel git commit: Fixed test. This closes #1404
Posted by da...@apache.org.
Fixed test. This closes #1404
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0243960c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0243960c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0243960c
Branch: refs/heads/master
Commit: 0243960c5c6f5bfb0e140e52c19f980a4229f82d
Parents: 3232712
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Jan 20 10:36:03 2017 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jan 20 10:36:03 2017 +0100
----------------------------------------------------------------------
.../management/ManagedThrottlingExceptionRoutePolicyTest.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0243960c/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
index 0d41938..eab7635 100644
--- a/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
+++ b/camel-core/src/test/java/org/apache/camel/management/ManagedThrottlingExceptionRoutePolicyTest.java
@@ -123,11 +123,11 @@ public class ManagedThrottlingExceptionRoutePolicyTest extends ManagementTestSu
myState = proxy.currentState();
assertTrue(myState.contains("State closed, failures 1, last failure"));
- // the route has no failures
+ // the route has 1 failure
val = proxy.getCurrentFailures();
assertEquals(1, val.intValue());
- // the route has no failures
+ // the route has 1 failure X mills ago
lastFail = proxy.getLastFailure();
assertTrue(lastFail.longValue() > 0);
}
@@ -153,6 +153,8 @@ public class ManagedThrottlingExceptionRoutePolicyTest extends ManagementTestSu
@Override
public void process(Exchange exchange) throws Exception {
+ // need to sleep a little to cause last failure to be slow
+ Thread.sleep(50);
throw new RuntimeException("boom!");
}
[3/5] camel git commit: replace changes lost in merge
Posted by da...@apache.org.
replace changes lost in merge
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/dd227460
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/dd227460
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/dd227460
Branch: refs/heads/master
Commit: dd22746086e769aa85eb3019b2c9de88beeabff5
Parents: 5351baa
Author: CodeSmell <mb...@gmail.com>
Authored: Thu Jan 19 17:13:02 2017 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jan 20 09:45:52 2017 +0100
----------------------------------------------------------------------
.../camel/impl/ThrottlingExceptionRoutePolicy.java | 12 ++++--------
1 file changed, 4 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/dd227460/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java b/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
index e23e1bc..aac5ee2 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
@@ -231,7 +231,10 @@ public class ThrottlingExceptionRoutePolicy extends RoutePolicySupport implement
try {
lock.lock();
startConsumer(route.getConsumer());
- this.reset();
+ failures.set(0);
+ lastFailure = 0;
+ openedAt = 0;
+ state.set(STATE_CLOSED);
logState();
} catch (Exception e) {
handleException(e);
@@ -240,13 +243,6 @@ public class ThrottlingExceptionRoutePolicy extends RoutePolicySupport implement
}
}
- private void reset() {
- failures.set(0);
- lastFailure = 0;
- openedAt = 0;
- state.set(STATE_CLOSED);
- }
-
private void logState() {
if (LOG.isDebugEnabled()) {
LOG.debug(dumpState());
[2/5] camel git commit: CAMEL-10719 add JMX management to
ThrottlingExceptionRoutePolicy
Posted by da...@apache.org.
CAMEL-10719 add JMX management to ThrottlingExceptionRoutePolicy
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/69b716bb
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/69b716bb
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/69b716bb
Branch: refs/heads/master
Commit: 69b716bbb38ff3754b72038526f44e74429ddb6d
Parents: a4f82a2
Author: CodeSmell <mb...@gmail.com>
Authored: Tue Jan 17 16:22:27 2017 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jan 20 09:44:37 2017 +0100
----------------------------------------------------------------------
...agedThrottlingExceptionRoutePolicyMBean.java | 55 +++++++++++
.../impl/ThrottlingExceptionRoutePolicy.java | 42 ++++++++-
.../ManagedThrottlingExceptionRoutePolicy.java | 99 ++++++++++++++++++++
3 files changed, 193 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/69b716bb/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
new file mode 100644
index 0000000..86e394c
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/api/management/mbean/ManagedThrottlingExceptionRoutePolicyMBean.java
@@ -0,0 +1,55 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.api.management.mbean;
+
+import org.apache.camel.api.management.ManagedAttribute;
+
+public interface ManagedThrottlingExceptionRoutePolicyMBean extends ManagedServiceMBean {
+
+ @ManagedAttribute(description = "how long to wait before moving open circuit to half open")
+ long getHalfOpenAfter();
+
+ @ManagedAttribute(description = "how long to wait before moving open circuit to half open")
+ void setHalfOpenAfter(long milliseconds);
+
+ @ManagedAttribute(description = "the range of time that failures should occur within")
+ long getFailureWindow();
+
+ @ManagedAttribute(description = "the range of time that failures should occur within")
+ void setFailureWindow(long milliseconds);
+
+ @ManagedAttribute(description = "number of failures before opening circuit")
+ int getFailureThreshold();
+
+ @ManagedAttribute(description = "number of failures before opening circuit")
+ void setFailureThreshold(int numberOfFailures);
+
+ @ManagedAttribute(description = "State")
+ String currentState();
+
+ @ManagedAttribute(description = "The half open handler registered (if any)")
+ String hasHalfOpenHandler();
+
+ @ManagedAttribute(description = "the number of failures caught")
+ int currentFailures();
+
+ @ManagedAttribute(description = "number of ms since the last failure was recorded")
+ long getLastFailure();
+
+ @ManagedAttribute(description = "number ms since the circuit was opened")
+ long getOpenAt();
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/69b716bb/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java b/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
index aaa4eca..34b755a 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/ThrottlingExceptionRoutePolicy.java
@@ -255,11 +255,11 @@ public class ThrottlingExceptionRoutePolicy extends RoutePolicySupport implement
public String dumpState() {
int num = state.get();
- String state = stateAsString(num);
+ String routeState = stateAsString(num);
if (failures.get() > 0) {
- return String.format("State %s, failures %d, last failure %d ms ago", state, failures.get(), System.currentTimeMillis() - lastFailure);
+ return String.format("*** State %s, failures %d, last failure %d ms ago", routeState, failures.get(), System.currentTimeMillis() - lastFailure);
} else {
- return String.format("State %s, failures %d", state, failures.get());
+ return String.format("*** State %s, failures %d", routeState, failures.get());
}
}
@@ -295,4 +295,40 @@ public class ThrottlingExceptionRoutePolicy extends RoutePolicySupport implement
this.halfOpenHandler = halfOpenHandler;
}
+ public int getFailureThreshold() {
+ return failureThreshold;
+ }
+
+ public void setFailureThreshold(int failureThreshold) {
+ this.failureThreshold = failureThreshold;
+ }
+
+ public long getFailureWindow() {
+ return failureWindow;
+ }
+
+ public void setFailureWindow(long failureWindow) {
+ this.failureWindow = failureWindow;
+ }
+
+ public long getHalfOpenAfter() {
+ return halfOpenAfter;
+ }
+
+ public void setHalfOpenAfter(long halfOpenAfter) {
+ this.halfOpenAfter = halfOpenAfter;
+ }
+
+ public int getFailures() {
+ return failures.get();
+ }
+
+ public long getLastFailure() {
+ return lastFailure;
+ }
+
+ public long getOpenedAt() {
+ return openedAt;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/camel/blob/69b716bb/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
new file mode 100644
index 0000000..bd2af1f
--- /dev/null
+++ b/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedThrottlingExceptionRoutePolicy.java
@@ -0,0 +1,99 @@
+/**
+ * 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.management.mbean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.api.management.ManagedResource;
+import org.apache.camel.api.management.mbean.ManagedThrottlingExceptionRoutePolicyMBean;
+import org.apache.camel.impl.ThrottlingExceptionHalfOpenHandler;
+import org.apache.camel.impl.ThrottlingExceptionRoutePolicy;
+
+@ManagedResource(description = "Managed ThrottlingExceptionRoutePolicy")
+public class ManagedThrottlingExceptionRoutePolicy extends ManagedService implements ManagedThrottlingExceptionRoutePolicyMBean {
+
+ private final ThrottlingExceptionRoutePolicy policy;
+
+ public ManagedThrottlingExceptionRoutePolicy(CamelContext context, ThrottlingExceptionRoutePolicy policy) {
+ super(context, policy);
+ this.policy = policy;
+ }
+
+ public ThrottlingExceptionRoutePolicy getPolicy() {
+ return policy;
+ }
+
+ @Override
+ public long getHalfOpenAfter() {
+ return getPolicy().getHalfOpenAfter();
+ }
+
+ @Override
+ public void setHalfOpenAfter(long milliseconds) {
+ getPolicy().setHalfOpenAfter(milliseconds);
+ }
+
+ @Override
+ public long getFailureWindow() {
+ return getPolicy().getFailureWindow();
+ }
+
+ @Override
+ public void setFailureWindow(long milliseconds) {
+ getPolicy().setFailureWindow(milliseconds);
+ }
+
+ @Override
+ public int getFailureThreshold() {
+ return getPolicy().getFailureThreshold();
+ }
+
+ @Override
+ public void setFailureThreshold(int numberOfFailures) {
+ getPolicy().setFailureThreshold(numberOfFailures);
+ }
+
+ @Override
+ public String currentState() {
+ return getPolicy().dumpState();
+ }
+
+ @Override
+ public String hasHalfOpenHandler() {
+ ThrottlingExceptionHalfOpenHandler obj = getPolicy().getHalfOpenHandler();
+ if (obj != null) {
+ return obj.getClass().getSimpleName();
+ } else {
+ return "";
+ }
+ }
+
+ @Override
+ public int currentFailures() {
+ return getPolicy().getFailures();
+ }
+
+ @Override
+ public long getLastFailure() {
+ return System.currentTimeMillis() - getPolicy().getLastFailure();
+ }
+
+ @Override
+ public long getOpenAt() {
+ return System.currentTimeMillis() - getPolicy().getOpenedAt();
+ }
+
+}
[4/5] camel git commit: Regen
Posted by da...@apache.org.
Regen
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/32327122
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/32327122
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/32327122
Branch: refs/heads/master
Commit: 32327122264d43e8f9d8704542c7b6114e8b14a3
Parents: dd22746
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Jan 20 10:24:53 2017 +0100
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jan 20 10:24:53 2017 +0100
----------------------------------------------------------------------
.../camel-amqp/src/main/docs/amqp-component.adoc | 8 +++++---
.../springboot/AMQPComponentConfiguration.java | 17 +++++++++++++++++
2 files changed, 22 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/32327122/components/camel-amqp/src/main/docs/amqp-component.adoc
----------------------------------------------------------------------
diff --git a/components/camel-amqp/src/main/docs/amqp-component.adoc b/components/camel-amqp/src/main/docs/amqp-component.adoc
index 644c5ca..97bd209 100644
--- a/components/camel-amqp/src/main/docs/amqp-component.adoc
+++ b/components/camel-amqp/src/main/docs/amqp-component.adoc
@@ -36,7 +36,7 @@ link:../../../../camel-jms/src/main/docs/readme.html[JMS] component after the de
// component options: START
-The AMQP component supports 74 options which are listed below.
+The AMQP component supports 75 options which are listed below.
@@ -117,6 +117,7 @@ The AMQP component supports 74 options which are listed below.
| messageCreatedStrategy | advanced | | MessageCreatedStrategy | To use the given MessageCreatedStrategy which are invoked when Camel creates new instances of javax.jms.Message objects when Camel is sending a JMS message.
| waitForProvisionCorrelationToBeUpdatedCounter | advanced | 50 | int | Number of times to wait for provisional correlation id to be updated to the actual correlation id when doing request/reply over JMS and when the option useMessageIDAsCorrelationID is enabled.
| waitForProvisionCorrelationToBeUpdatedThreadSleepingTime | advanced | 100 | long | Interval in millis to sleep each time while waiting for provisional correlation id to be updated.
+| correlationProperty | producer (advanced) | | String | Use this JMS property to correlate messages in InOut exchange pattern (request-reply) instead of JMSCorrelationID property. This allows you to exchange messages with systems that do not correlate messages using JMSCorrelationID JMS property. If used JMSCorrelationID will not be used or set by Camel. The value of here named property will be generated if not supplied in the header of the message under the same name.
| headerFilterStrategy | filter | | HeaderFilterStrategy | To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter header to and from Camel message.
|=======================================================================
{% endraw %}
@@ -129,7 +130,7 @@ The AMQP component supports 74 options which are listed below.
// endpoint options: START
-The AMQP component supports 86 endpoint options which are listed below:
+The AMQP component supports 87 endpoint options which are listed below:
{% raw %}
[width="100%",cols="2,1,1m,1m,5",options="header"]
@@ -178,6 +179,7 @@ The AMQP component supports 86 endpoint options which are listed below:
| timeToLive | producer | -1 | long | When sending messages specifies the time-to-live of the message (in milliseconds).
| allowNullBody | producer (advanced) | true | boolean | Whether to allow sending messages with no body. If this option is false and the message body is null then an JMSException is thrown.
| alwaysCopyMessage | producer (advanced) | false | boolean | If true Camel will always make a JMS message copy of the message when it is passed to the producer for sending. Copying the message is needed in some situations such as when a replyToDestinationSelectorName is set (incidentally Camel will set the alwaysCopyMessage option to true if a replyToDestinationSelectorName is set)
+| correlationProperty | producer (advanced) | | String | When using InOut exchange pattern use this JMS property instead of JMSCorrelationID JMS property to correlate messages. If set messages will be correlated solely on the value of this property JMSCorrelationID property will be ignored and not set by Camel.
| disableTimeToLive | producer (advanced) | false | boolean | Use this option to force disabling time to live. For example when you do request/reply over JMS then Camel will by default use the requestTimeout value as time to live on the message being sent. The problem is that the sender and receiver systems have to have their clocks synchronized so they are in sync. This is not always so easy to archive. So you can use disableTimeToLive=true to not set a time to live value on the sent message. Then the message will not expire on the receiver system. See below in section About time to live for more details.
| forceSendOriginalMessage | producer (advanced) | false | boolean | When using mapJmsMessage=false Camel will create a new JMS message to send to a new JMS destination if you touch the headers (get or set) during the route. Set this option to true to force Camel to send the original JMS message that was received.
| includeSentJMSMessageID | producer (advanced) | false | boolean | Only applicable when sending to JMS destination using InOnly (eg fire and forget). Enabling this option will enrich the Camel Exchange with the actual JMSMessageID that was used by the JMS client when the message was sent to the JMS destination.
@@ -356,4 +358,4 @@ prefix, so you don't have to configure it explicitly.
* link:configuring-camel.html[Configuring Camel]
* link:component.html[Component]
* link:endpoint.html[Endpoint]
-* link:getting-started.html[Getting Started]
\ No newline at end of file
+* link:getting-started.html[Getting Started]
http://git-wip-us.apache.org/repos/asf/camel/blob/32327122/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
----------------------------------------------------------------------
diff --git a/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java b/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
index b1d49c6..a1d2929 100644
--- a/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
+++ b/platforms/spring-boot/components-starter/camel-amqp-starter/src/main/java/org/apache/camel/component/amqp/springboot/AMQPComponentConfiguration.java
@@ -554,6 +554,15 @@ public class AMQPComponentConfiguration {
*/
private Long waitForProvisionCorrelationToBeUpdatedThreadSleepingTime = 100L;
/**
+ * Use this JMS property to correlate messages in InOut exchange pattern
+ * (request-reply) instead of JMSCorrelationID property. This allows you to
+ * exchange messages with systems that do not correlate messages using
+ * JMSCorrelationID JMS property. If used JMSCorrelationID will not be used
+ * or set by Camel. The value of here named property will be generated if
+ * not supplied in the header of the message under the same name.
+ */
+ private String correlationProperty;
+ /**
* To use a custom org.apache.camel.spi.HeaderFilterStrategy to filter
* header to and from Camel message.
*/
@@ -1160,6 +1169,14 @@ public class AMQPComponentConfiguration {
this.waitForProvisionCorrelationToBeUpdatedThreadSleepingTime = waitForProvisionCorrelationToBeUpdatedThreadSleepingTime;
}
+ public String getCorrelationProperty() {
+ return correlationProperty;
+ }
+
+ public void setCorrelationProperty(String correlationProperty) {
+ this.correlationProperty = correlationProperty;
+ }
+
public HeaderFilterStrategy getHeaderFilterStrategy() {
return headerFilterStrategy;
}