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 2011/01/16 09:36:54 UTC
svn commit: r1059513 - in /camel/trunk/camel-core/src:
main/java/org/apache/camel/ main/java/org/apache/camel/impl/
main/java/org/apache/camel/management/mbean/
main/java/org/apache/camel/spi/ test/java/org/apache/camel/impl/
test/java/org/apache/camel...
Author: davsclaus
Date: Sun Jan 16 08:36:54 2011
New Revision: 1059513
URL: http://svn.apache.org/viewvc?rev=1059513&view=rev
Log:
CAMEL-3240: Added shutdown route with timeout and abort. Thanks to Ben for patch.
Added:
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/StopRouteAbortAfterTimeoutTest.java
camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedRouteStopWithAbortAfterTimeoutTest.java
Removed:
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/ShutdownRouteGracefulGiveUpTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultShutdownStrategy.java
camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
camel/trunk/camel-core/src/main/java/org/apache/camel/spi/ShutdownStrategy.java
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java?rev=1059513&r1=1059512&r2=1059513&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java Sun Jan 16 08:36:54 2011
@@ -428,17 +428,18 @@ public interface CamelContext extends Su
void stopRoute(String routeId, long timeout, TimeUnit timeUnit) throws Exception;
/**
- * Stops the given route using {@link org.apache.camel.spi.ShutdownStrategy} with a specified timeout.
+ * Stops the given route using {@link org.apache.camel.spi.ShutdownStrategy} with a specified timeout
+ * and optional abortAfterTimeout mode.
*
* @param routeId the route id
* @param timeout timeout
* @param timeUnit the unit to use
- * @param giveUp should give up after timeout
+ * @param abortAfterTimeout should abort shutdown after timeout
* @return <tt>true</tt> if the route is stopped before the timeout
* @throws Exception is thrown if the route could not be stopped for whatever reason
* @see #suspendRoute(String, long, java.util.concurrent.TimeUnit)
*/
- boolean stopRoute(String routeId, long timeout, TimeUnit timeUnit, boolean giveUp) throws Exception;
+ boolean stopRoute(String routeId, long timeout, TimeUnit timeUnit, boolean abortAfterTimeout) throws Exception;
/**
* Shutdown and <b>removes</b> the given route using {@link org.apache.camel.spi.ShutdownStrategy}.
@@ -967,4 +968,4 @@ public interface CamelContext extends Su
*/
void setLazyLoadTypeConverters(Boolean lazyLoadTypeConverters);
-}
\ No newline at end of file
+}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java?rev=1059513&r1=1059512&r2=1059513&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultCamelContext.java Sun Jan 16 08:36:54 2011
@@ -727,18 +727,19 @@ public class DefaultCamelContext extends
}
}
- public synchronized boolean stopRoute(String routeId, long timeout, TimeUnit timeUnit, boolean giveUp) throws Exception {
+ public synchronized boolean stopRoute(String routeId, long timeout, TimeUnit timeUnit, boolean abortAfterTimeout) throws Exception {
RouteService routeService = routeServices.get(routeId);
if (routeService != null) {
- List<RouteStartupOrder> routes = new ArrayList<RouteStartupOrder>(1);
- RouteStartupOrder order = new DefaultRouteStartupOrder(1, routeService.getRoutes().iterator().next(), routeService);
- routes.add(order);
+ RouteStartupOrder route = new DefaultRouteStartupOrder(1, routeService.getRoutes().iterator().next(), routeService);
- boolean completed = getShutdownStrategy().shutdown(this, routes, timeout, timeUnit, giveUp);
+ boolean completed = getShutdownStrategy().shutdown(this, route, timeout, timeUnit, abortAfterTimeout);
if (completed) {
// must stop route service as well
routeService.setRemovingRoutes(false);
stopRouteService(routeService);
+ } else {
+ // shutdown was aborted, make sure route is re-started properly
+ startRouteService(routeService, false);
}
return completed;
}
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultShutdownStrategy.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultShutdownStrategy.java?rev=1059513&r1=1059512&r2=1059513&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultShutdownStrategy.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/impl/DefaultShutdownStrategy.java Sun Jan 16 08:36:54 2011
@@ -92,15 +92,17 @@ public class DefaultShutdownStrategy ext
doShutdown(context, routes, timeout, timeUnit, false, false);
}
- public boolean shutdown(CamelContext context, List<RouteStartupOrder> routes, long timeout, TimeUnit timeUnit, boolean giveUp) throws Exception {
- return doShutdown(context, routes, timeout, timeUnit, false, giveUp);
+ public boolean shutdown(CamelContext context, RouteStartupOrder route, long timeout, TimeUnit timeUnit, boolean abortAfterTimeout) throws Exception {
+ List<RouteStartupOrder> routes = new ArrayList<RouteStartupOrder>(1);
+ routes.add(route);
+ return doShutdown(context, routes, timeout, timeUnit, false, abortAfterTimeout);
}
public void suspend(CamelContext context, List<RouteStartupOrder> routes, long timeout, TimeUnit timeUnit) throws Exception {
doShutdown(context, routes, timeout, timeUnit, true, false);
}
- protected boolean doShutdown(CamelContext context, List<RouteStartupOrder> routes, long timeout, TimeUnit timeUnit, boolean suspendOnly, boolean giveUp) throws Exception {
+ protected boolean doShutdown(CamelContext context, List<RouteStartupOrder> routes, long timeout, TimeUnit timeUnit, boolean suspendOnly, boolean abortAfterTimeout) throws Exception {
StopWatch watch = new StopWatch();
// at first sort according to route startup order
@@ -121,7 +123,7 @@ public class DefaultShutdownStrategy ext
}
// use another thread to perform the shutdowns so we can support timeout
- Future future = getExecutorService().submit(new ShutdownTask(context, routesOrdered, suspendOnly));
+ Future future = getExecutorService().submit(new ShutdownTask(context, routesOrdered, suspendOnly, abortAfterTimeout));
try {
if (timeout > 0) {
future.get(timeout, timeUnit);
@@ -132,9 +134,9 @@ public class DefaultShutdownStrategy ext
// timeout then cancel the task
future.cancel(true);
- //if set, stop processing and return false to indicate that the shutdown is giving up
- if (giveUp) {
- LOG.warn("Timeout occurred. Giving up now.");
+ // if set, stop processing and return false to indicate that the shutdown is aborting
+ if (abortAfterTimeout) {
+ LOG.warn("Timeout occurred. Aborting the shutdown now.");
return false;
} else {
if (shutdownNowOnTimeout) {
@@ -331,11 +333,13 @@ public class DefaultShutdownStrategy ext
private final CamelContext context;
private final List<RouteStartupOrder> routes;
private final boolean suspendOnly;
+ private final boolean abortAfterTimeout;
- public ShutdownTask(CamelContext context, List<RouteStartupOrder> routes, boolean suspendOnly) {
+ public ShutdownTask(CamelContext context, List<RouteStartupOrder> routes, boolean suspendOnly, boolean abortAfterTimeout) {
this.context = context;
this.routes = routes;
this.suspendOnly = suspendOnly;
+ this.abortAfterTimeout = abortAfterTimeout;
}
public void run() {
@@ -434,9 +438,15 @@ public class DefaultShutdownStrategy ext
+ (TimeUnit.SECONDS.convert(getTimeout(), getTimeUnit()) - (loopCount++ * loopDelaySeconds)) + " seconds.");
Thread.sleep(loopDelaySeconds * 1000);
} catch (InterruptedException e) {
- LOG.warn("Interrupted while waiting during graceful shutdown, will force shutdown now.");
- Thread.currentThread().interrupt();
- break;
+ if (abortAfterTimeout) {
+ LOG.warn("Interrupted while waiting during graceful shutdown, will abort.");
+ //Thread.currentThread().interrupt();
+ return;
+ } else {
+ LOG.warn("Interrupted while waiting during graceful shutdown, will force shutdown now.");
+ Thread.currentThread().interrupt();
+ break;
+ }
}
} else {
done = true;
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java?rev=1059513&r1=1059512&r2=1059513&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/management/mbean/ManagedRoute.java Sun Jan 16 08:36:54 2011
@@ -143,6 +143,14 @@ public class ManagedRoute extends Manage
context.stopRoute(getRouteId(), timeout, TimeUnit.SECONDS);
}
+ @ManagedOperation(description = "Stop route, abort stop after timeout (in seconds)")
+ public boolean stop(Long timeout, Boolean abortAfterTimeout) throws Exception {
+ if (!context.getStatus().isStarted()) {
+ throw new IllegalArgumentException("CamelContext is not started");
+ }
+ return context.stopRoute(getRouteId(), timeout, TimeUnit.SECONDS, abortAfterTimeout);
+ }
+
@ManagedOperation(description = "Shutdown and remove route")
@Deprecated
public void shutdown() throws Exception {
Modified: camel/trunk/camel-core/src/main/java/org/apache/camel/spi/ShutdownStrategy.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/spi/ShutdownStrategy.java?rev=1059513&r1=1059512&r2=1059513&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/spi/ShutdownStrategy.java (original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/spi/ShutdownStrategy.java Sun Jan 16 08:36:54 2011
@@ -70,17 +70,17 @@ public interface ShutdownStrategy extend
void shutdown(CamelContext context, List<RouteStartupOrder> routes, long timeout, TimeUnit timeUnit) throws Exception;
/**
- * Shutdown the routes using a specified timeout instead of the default timeout values and supports "give up" mode
+ * Shutdown the route using a specified timeout instead of the default timeout values and supports abortAfterTimeout mode
*
* @param context the camel context
- * @param routes the routes, ordered by the order they was started
+ * @param route the route
* @param timeout timeout
* @param timeUnit the unit to use
- * @param giveUp should give up after timeout
+ * @param abortAfterTimeout should abort shutdown after timeout
* @return <tt>true</tt> if the route is stopped before the timeout
- * @throws Exception is thrown if error shutting down the consumers, however its preferred to avoid this
+ * @throws Exception is thrown if error shutting down the consumer, however its preferred to avoid this
*/
- boolean shutdown(CamelContext context, List<RouteStartupOrder> routes, long timeout, TimeUnit timeUnit, boolean giveUp) throws Exception;
+ boolean shutdown(CamelContext context, RouteStartupOrder route, long timeout, TimeUnit timeUnit, boolean abortAfterTimeout) throws Exception;
/**
* Suspends the routes using a specified timeout instead of the default timeout values
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/impl/StopRouteAbortAfterTimeoutTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/impl/StopRouteAbortAfterTimeoutTest.java?rev=1059513&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/impl/StopRouteAbortAfterTimeoutTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/impl/StopRouteAbortAfterTimeoutTest.java Sun Jan 16 08:36:54 2011
@@ -0,0 +1,90 @@
+/**
+ * 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.impl;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+
+public class StopRouteAbortAfterTimeoutTest extends ContextTestSupport {
+
+ public void testStopRouteWithAbortAfterTimeoutTrue() throws Exception {
+
+ MockEndpoint mockEP = getMockEndpoint("mock:result");
+ mockEP.setExpectedMessageCount(10);
+
+ //send some message through the route
+ for (int i = 0; i < 5; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ // stop route with a 2s timeout and abortAfterTimeout=true (should abort after 2s)
+ boolean stopRouteResponse = context.stopRoute("start", 2, TimeUnit.SECONDS, true);
+
+ // confirm that route is still running
+ assertFalse("stopRoute response should be False", stopRouteResponse);
+ assertEquals("route should still be started", true, context.getRouteStatus("start").isStarted());
+
+ //send some more messages through the route
+ for (int i = 5; i < 10; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ mockEP.assertIsSatisfied();
+ }
+
+ public void testStopRouteWithAbortAfterTimeoutFalse() throws Exception {
+
+ MockEndpoint mockEP = getMockEndpoint("mock:result");
+
+ //send some message through the route
+ for (int i = 0; i < 5; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ // stop route with a 2s timeout and abortAfterTimeout=false (normal timeout behavior)
+ boolean stopRouteResponse = context.stopRoute("start", 2, TimeUnit.SECONDS, false);
+
+ // the route should have been forced stopped
+ assertTrue("stopRoute response should be True", stopRouteResponse);
+ assertEquals("route should be stopped", true, context.getRouteStatus("start").isStopped());
+
+ // send some more messages through the route
+ for (int i = 5; i < 10; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ Thread.sleep(3000);
+
+ assertTrue("Should not have received more than 5 messages", mockEP.getExchanges().size() <= 5);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ // shutdown this test faster
+ context.getShutdownStrategy().setTimeout(3);
+
+ from("seda:start").routeId("start").delay(500).to("mock:result");
+ }
+ };
+ }
+}
Added: camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedRouteStopWithAbortAfterTimeoutTest.java
URL: http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedRouteStopWithAbortAfterTimeoutTest.java?rev=1059513&view=auto
==============================================================================
--- camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedRouteStopWithAbortAfterTimeoutTest.java (added)
+++ camel/trunk/camel-core/src/test/java/org/apache/camel/management/ManagedRouteStopWithAbortAfterTimeoutTest.java Sun Jan 16 08:36:54 2011
@@ -0,0 +1,127 @@
+/**
+ * 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.component.mock.MockEndpoint;
+
+/**
+ * @version $Revision: 1058591 $
+ */
+public class ManagedRouteStopWithAbortAfterTimeoutTest extends ManagementTestSupport {
+
+ public void testStopRouteWithAbortAfterTimeoutTrue() throws Exception {
+
+ MockEndpoint mockEP = getMockEndpoint("mock:result");
+ mockEP.setExpectedMessageCount(10);
+
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = getRouteObjectName(mbeanServer);
+
+ // confirm that route has started
+ String state = (String) mbeanServer.getAttribute(on, "State");
+ assertEquals("route should be started", ServiceStatus.Started.name(), state);
+
+ //send some message through the route
+ for (int i = 0; i < 5; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ // stop route with a 2s timeout and abortAfterTimeout=true (should abort after 2s)
+ Long timeout = new Long(2);
+ Boolean abortAfterTimeout = Boolean.TRUE;
+ Object[] params = {timeout, abortAfterTimeout};
+ String[] sig = {"java.lang.Long", "java.lang.Boolean"};
+ Boolean stopRouteResponse = (Boolean) mbeanServer.invoke(on, "stop", params, sig);
+
+ // confirm that route is still running
+ state = (String) mbeanServer.getAttribute(on, "State");
+ assertFalse("stopRoute response should be False", stopRouteResponse);
+ assertEquals("route should still be started", ServiceStatus.Started.name(), state);
+
+ //send some more messages through the route
+ for (int i = 5; i < 10; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ mockEP.assertIsSatisfied();
+ }
+
+ public void testStopRouteWithAbortAfterTimeoutFalse() throws Exception {
+
+ MockEndpoint mockEP = getMockEndpoint("mock:result");
+
+ MBeanServer mbeanServer = getMBeanServer();
+ ObjectName on = getRouteObjectName(mbeanServer);
+
+ // confirm that route has started
+ String state = (String) mbeanServer.getAttribute(on, "State");
+ assertEquals("route should be started", ServiceStatus.Started.name(), state);
+
+ //send some message through the route
+ for (int i = 0; i < 5; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ // stop route with a 2s timeout and abortAfterTimeout=false (normal timeout behavior)
+ Long timeout = new Long(2);
+ Boolean abortAfterTimeout = Boolean.FALSE;
+ Object[] params = {timeout, abortAfterTimeout};
+ String[] sig = {"java.lang.Long", "java.lang.Boolean"};
+ Boolean stopRouteResponse = (Boolean) mbeanServer.invoke(on, "stop", params, sig);
+
+ // confirm that route is stopped
+ state = (String) mbeanServer.getAttribute(on, "State");
+ assertTrue("stopRoute response should be True", stopRouteResponse);
+ assertEquals("route should be stopped", ServiceStatus.Stopped.name(), state);
+
+ // send some more messages through the route
+ for (int i = 5; i < 10; i++) {
+ template.sendBody("seda:start", "message-" + i);
+ }
+
+ Thread.sleep(3000);
+
+ assertTrue("Should not have received more than 5 messages", mockEP.getExchanges().size() <= 5);
+ }
+
+ static ObjectName getRouteObjectName(MBeanServer mbeanServer) throws Exception {
+ Set<ObjectName> set = mbeanServer.queryNames(new ObjectName("*:type=routes,*"), null);
+ assertEquals(1, set.size());
+ return set.iterator().next();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ // shutdown this test faster
+ context.getShutdownStrategy().setTimeout(3);
+
+ from("seda:start").delay(500).to("mock:result");
+ }
+ };
+ }
+
+}