Author: davsclaus
Date: Fri Mar 12 15:27:40 2010
New Revision: 922296
URL: http://svn.apache.org/viewvc?rev=922296&view=rev
Log:
CAMEL-1588: Added ShutdownableService to have both stop/shutdown state. So you
can do different work.
Added:
camel/trunk/camel-core/src/main/java/org/apache/camel/ShutdownableService.java
- copied, changed from r922137,
camel/trunk/camel-core/src/main/java/org/apache/camel/SuspendableService.java
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateShutdownThreadPoolTest.java
- copied, changed from r922137,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSimpleSizeTest.java
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/CamelContext.java
camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.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/RouteService.java
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ServiceSupport.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ServiceHelper.java
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/ServiceSupportTest.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=922296&r1=922295&r2=922296&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 Fri
Mar 12 15:27:40 2010
@@ -67,6 +67,13 @@ public interface CamelContext extends Se
*/
String getVersion();
+ /**
+ * Get the status of this context
+ *
+ * @return the status
+ */
+ ServiceStatus getStatus();
+
// Service Methods
//-----------------------------------------------------------------------
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
--- camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
(original)
+++ camel/trunk/camel-core/src/main/java/org/apache/camel/ServiceStatus.java
Fri Mar 12 15:27:40 2010
@@ -19,7 +19,7 @@ package org.apache.camel;
import java.io.Serializable;
/**
- * Reresents the status of a {...@link Service} instance
+ * Represents the status of a {...@link Service} instance
*
* @version $Revision$
*/
Copied:
camel/trunk/camel-core/src/main/java/org/apache/camel/ShutdownableService.java
(from r922137,
camel/trunk/camel-core/src/main/java/org/apache/camel/SuspendableService.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/ShutdownableService.java?p2=camel/trunk/camel-core/src/main/java/org/apache/camel/ShutdownableService.java&p1=camel/trunk/camel-core/src/main/java/org/apache/camel/SuspendableService.java&r1=922137&r2=922296&rev=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/SuspendableService.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/ShutdownableService.java
Fri Mar 12 15:27:40 2010
@@ -17,32 +17,22 @@
package org.apache.camel;
/**
- * A {...@link Service} which is also capable of suspending and resuming.
+ * A {...@link org.apache.camel.Service} which is also capable of shutting
down.
* <p/>
- * This is useable for services which needs more fine grained control at
runtime supporting suspension.
- * Other services may select to mimic suspending by just stopping the service.
+ * This is useable for services which needs more fine grained control when
{...@link CamelContext} is shutting down.
+ * Then its possible to have different behavior when stopping or shutting down.
* <p/>
- * For example this is use by the JmsConsumer which suspends the Spring JMS
listener instead of stopping
- * the consumer totally.
+ * For example to shutdown thread pools on shutdown and <b>not</b> on stop.
*
* @version $Revision$
*/
-public interface SuspendableService extends Service {
+public interface ShutdownableService extends Service {
/**
- * Suspends the service.
- */
- void suspend();
-
- /**
- * Resumes the service.
- */
- void resume();
-
- /**
- * Tests whether the service is suspended or not.
+ * Shutdown the service, which means it cannot be started again.
*
- * @return <tt>true</tt> if suspended
+ * @throws Exception is thrown if shutting down failed
*/
- boolean isSuspended();
-}
+ void shutdown() throws Exception;
+
+}
\ No newline at end of file
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/component/seda/SedaConsumer.java
Fri Mar 12 15:27:40 2010
@@ -189,6 +189,7 @@ public class SedaConsumer extends Servic
protected void doStop() throws Exception {
endpoint.onStopped(this);
+ // must shutdown executor on stop to avoid overhead of having them
running
executor.shutdownNow();
executor = null;
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=922296&r1=922295&r2=922296&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
Fri Mar 12 15:27:40 2010
@@ -1172,34 +1172,34 @@ public class DefaultCamelContext extends
}
protected synchronized void doStop() throws Exception {
- LOG.info("Apache Camel " + getVersion() + " (CamelContext:" +
getName() + ") is stopping");
+ LOG.info("Apache Camel " + getVersion() + " (CamelContext:" +
getName() + ") is shutting down");
EventHelper.notifyCamelContextStopping(this);
// stop route inputs in the same order as they was started so we stop
the very first inputs first
shutdownStrategy.shutdown(this, getRouteStartupOrder());
getRouteStartupOrder().clear();
- stopServices(routeServices.values());
+ shutdownServices(routeServices.values());
// do not clear route services as we can start Camel again and get the
route back as before
// the stop order is important
- stopServices(servicesToClose);
+ shutdownServices(servicesToClose);
servicesToClose.clear();
- stopServices(endpoints.values());
+ shutdownServices(endpoints.values());
endpoints.clear();
- stopServices(components.values());
+ shutdownServices(components.values());
components.clear();
// special shutdown of a shared producer service pool as it should
only be shutdown by camel context
if (producerServicePool instanceof SharedProducerServicePool) {
((SharedProducerServicePool) producerServicePool).shutdown(this);
} else {
- stopServices(producerServicePool);
+ shutdownServices(producerServicePool);
}
- stopServices(inflightRepository);
+ shutdownServices(inflightRepository);
try {
for (LifecycleStrategy strategy : lifecycleStrategies) {
@@ -1212,10 +1212,10 @@ public class DefaultCamelContext extends
// must notify that we are stopped before stopping the management
strategy
EventHelper.notifyCamelContextStopped(this);
- // stop management as the last one
- stopServices(getManagementStrategy());
+ // shutdown management as the last one
+ shutdownServices(getManagementStrategy());
- LOG.info("Apache Camel " + getVersion() + " (CamelContext:" +
getName() + ") stopped");
+ LOG.info("Apache Camel " + getVersion() + " (CamelContext:" +
getName() + ") shutdown");
}
private void stopServices(Object service) throws Exception {
@@ -1229,12 +1229,23 @@ public class DefaultCamelContext extends
}
}
- private void stopServices(Collection<?> services) throws Exception {
+ private void shutdownServices(Object service) throws Exception {
+ // allow us to do custom work before delegating to service helper
+ try {
+ ServiceHelper.stopAndShutdownService(service);
+ } catch (Exception e) {
+ LOG.warn("Error occurred while shutting down service: " + service
+ ". This exception will be ignored.");
+ // fire event
+ EventHelper.notifyServiceStopFailure(this, service, e);
+ }
+ }
+
+ private void shutdownServices(Collection<?> services) throws Exception {
// reverse stopping by default
- stopServices(services, true);
+ shutdownServices(services, true);
}
- private void stopServices(Collection<?> services, boolean reverse) throws
Exception {
+ private void shutdownServices(Collection<?> services, boolean reverse)
throws Exception {
Collection<Object> list = CastUtils.cast(services);
if (reverse) {
ArrayList<Object> reverseList = new ArrayList<Object>(services);
@@ -1243,7 +1254,7 @@ public class DefaultCamelContext extends
}
for (Object service : list) {
- stopServices(service);
+ shutdownServices(service);
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteService.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/RouteService.java
Fri Mar 12 15:27:40 2010
@@ -165,6 +165,9 @@ public class RouteService extends Servic
strategy.onRoutesRemove(routes);
}
+ // if we are stopping CamelContext then we are shutting down
+ boolean isShutdownCamelContext = camelContext.isStopping();
+
for (Route route : routes) {
if (LOG.isTraceEnabled()) {
LOG.trace("Stopping route: " + route);
@@ -177,10 +180,14 @@ public class RouteService extends Servic
for (Service service : services) {
doGetChildServices(list, service);
}
- stopChildService(route, list);
+ stopChildService(route, list, isShutdownCamelContext);
// stop the route itself
- ServiceHelper.stopService(route);
+ if (isShutdownCamelContext) {
+ ServiceHelper.stopAndShutdownService(route);
+ } else {
+ ServiceHelper.stopService(route);
+ }
// fire event
EventHelper.notifyRouteStopped(camelContext, route);
@@ -204,12 +211,16 @@ public class RouteService extends Servic
}
}
- protected void stopChildService(Route route, List<Service> services)
throws Exception {
+ protected void stopChildService(Route route, List<Service> services,
boolean shutdown) throws Exception {
for (Service service : services) {
for (LifecycleStrategy strategy :
camelContext.getLifecycleStrategies()) {
strategy.onServiceRemove(camelContext, service, route);
}
- ServiceHelper.stopService(service);
+ if (shutdown) {
+ ServiceHelper.stopAndShutdownService(service);
+ } else {
+ ServiceHelper.stopService(service);
+ }
removeChildService(service);
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ServiceSupport.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ServiceSupport.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ServiceSupport.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/impl/ServiceSupport.java
Fri Mar 12 15:27:40 2010
@@ -22,6 +22,7 @@ import java.util.concurrent.atomic.Atomi
import org.apache.camel.Service;
import org.apache.camel.ServiceStatus;
+import org.apache.camel.ShutdownableService;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.ServiceHelper;
@@ -31,12 +32,14 @@ import org.apache.camel.util.ServiceHelp
*
* @version $Revision$
*/
-public abstract class ServiceSupport implements Service {
+public abstract class ServiceSupport implements Service, ShutdownableService {
private final AtomicBoolean started = new AtomicBoolean(false);
private final AtomicBoolean starting = new AtomicBoolean(false);
private final AtomicBoolean stopping = new AtomicBoolean(false);
private final AtomicBoolean stopped = new AtomicBoolean(false);
+ private final AtomicBoolean shuttingdown = new AtomicBoolean(false);
+ private final AtomicBoolean shutdown = new AtomicBoolean(false);
private Collection<Object> childServices;
private String version;
@@ -101,6 +104,26 @@ public abstract class ServiceSupport imp
}
}
+ public void shutdown() throws Exception {
+ // ensure we are stopped first
+ stop();
+
+ if (shuttingdown.compareAndSet(false, true)) {
+ try {
+ try {
+ doShutdown();
+ } finally {
+ if (childServices != null) {
+ ServiceHelper.stopAndShutdownService(childServices);
+ }
+ }
+ } finally {
+ shutdown.set(true);
+ shuttingdown.set(false);
+ }
+ }
+ }
+
/**
* Returns the current status
*/
@@ -165,6 +188,10 @@ public abstract class ServiceSupport imp
protected abstract void doStop() throws Exception;
+ protected void doShutdown() throws Exception {
+ // noop
+ }
+
@SuppressWarnings("unchecked")
protected void addChildService(Object childService) {
synchronized (this) {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/MulticastProcessor.java
Fri Mar 12 15:27:40 2010
@@ -355,19 +355,23 @@ public class MulticastProcessor extends
return new ProcessorExchangePair(processor, prepared, exchange);
}
+ protected void doStart() throws Exception {
+ ServiceHelper.startServices(processors);
+ }
+
protected void doStop() throws Exception {
+ ServiceHelper.stopServices(processors);
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ // only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdown();
- executorService.awaitTermination(0, TimeUnit.SECONDS);
+ executorService.shutdownNow();
executorService = null;
}
- ServiceHelper.stopServices(processors);
}
- protected void doStart() throws Exception {
- ServiceHelper.startServices(processors);
- }
-
private static void setToEndpoint(Exchange exchange, Processor processor) {
if (processor instanceof Producer) {
Producer producer = (Producer) processor;
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/OnCompletionProcessor.java
Fri Mar 12 15:27:40 2010
@@ -63,12 +63,17 @@ public class OnCompletionProcessor exten
@Override
protected void doStop() throws Exception {
+ ServiceHelper.stopService(processor);
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ // only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdown();
+ executorService.shutdownNow();
// must null it so we can restart
executorService = null;
}
- ServiceHelper.stopService(processor);
}
public void process(Exchange exchange) throws Exception {
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/SendAsyncProcessor.java
Fri Mar 12 15:27:40 2010
@@ -275,6 +275,7 @@ public class SendAsyncProcessor extends
protected void doStop() throws Exception {
super.doStop();
+ // must shutdown thread pools on stop as we are consumers
if (producerExecutorService != null) {
producerExecutorService.shutdownNow();
producerExecutorService = null;
@@ -284,7 +285,6 @@ public class SendAsyncProcessor extends
executorService = null;
}
completedTasks.clear();
-
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/ThreadsProcessor.java
Fri Mar 12 15:27:40 2010
@@ -23,6 +23,7 @@ import java.util.concurrent.Future;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
+import org.apache.camel.ShutdownableService;
import org.apache.camel.WaitForTaskToComplete;
import org.apache.camel.util.ExchangeHelper;
import org.apache.camel.util.ObjectHelper;
@@ -108,11 +109,12 @@ public class ThreadsProcessor extends De
return executorService;
}
- protected void doStop() throws Exception {
- super.doStop();
+ @Override
+ protected void doShutdown() throws Exception {
+ super.doShutdown();
+ // only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdown();
- // must null it so we can restart
+ executorService.shutdownNow();
executorService = null;
}
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/WireTapProcessor.java
Fri Mar 12 15:27:40 2010
@@ -53,18 +53,14 @@ public class WireTapProcessor extends Se
}
@Override
- protected void doStart() throws Exception {
- super.doStart();
- }
-
- @Override
- protected void doStop() throws Exception {
+ protected void doShutdown() throws Exception {
+ super.doShutdown();
+ // only shutdown thread pool on shutdown
if (executorService != null) {
- executorService.shutdown();
+ executorService.shutdownNow();
// must null it so we can restart
executorService = null;
}
- super.doStop();
}
@Override
@@ -81,7 +77,7 @@ public class WireTapProcessor extends Se
getProducerCache(exchange).doInProducer(destination, exchange,
pattern, new ProducerCallback<Exchange>() {
public Exchange doInProducer(Producer producer, Exchange exchange,
ExchangePattern pattern) throws Exception {
Exchange wireTapExchange = configureExchange(exchange,
pattern);
- procesWireTap(producer, wireTapExchange);
+ processWireTap(producer, wireTapExchange);
return wireTapExchange;
}
});
@@ -92,10 +88,10 @@ public class WireTapProcessor extends Se
*
* @param exchange the exchange to wire tap
*/
- protected void procesWireTap(final Producer producer, final Exchange
exchange) {
+ protected void processWireTap(final Producer producer, final Exchange
exchange) {
// use submit instead of execute to force it to use a new thread,
execute might
// decide to use current thread, so we must submit a new task
- // as we dont care for the response we dont hold the future object and
wait for the result
+ // as we don't care for the response we dont hold the future object
and wait for the result
getExecutorService().submit(new Callable<Exchange>() {
public Exchange call() throws Exception {
if (LOG.isDebugEnabled()) {
@@ -153,7 +149,7 @@ public class WireTapProcessor extends Se
}
public ExecutorService getExecutorService() {
- if (executorService == null || executorService.isShutdown()) {
+ if (executorService == null) {
executorService = createExecutorService();
}
return executorService;
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/processor/aggregate/AggregateProcessor.java
Fri Mar 12 15:27:40 2010
@@ -495,11 +495,6 @@ public class AggregateProcessor extends
protected void doStop() throws Exception {
ServiceHelper.stopService(timeoutMap);
- if (executorService != null) {
- executorService.shutdown();
- executorService = null;
- }
-
ServiceHelper.stopService(aggregationRepository);
if (closedCorrelationKeys != null) {
@@ -507,4 +502,13 @@ public class AggregateProcessor extends
}
}
+ @Override
+ protected void doShutdown() throws Exception {
+ // only shutdown thread pool when we are shutting down
+ if (executorService != null) {
+ executorService.shutdownNow();
+ executorService = null;
+ }
+ }
+
}
Modified:
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ServiceHelper.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/main/java/org/apache/camel/util/ServiceHelper.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ServiceHelper.java
(original)
+++
camel/trunk/camel-core/src/main/java/org/apache/camel/util/ServiceHelper.java
Fri Mar 12 15:27:40 2010
@@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.List;
import org.apache.camel.Service;
+import org.apache.camel.ShutdownableService;
import org.apache.camel.SuspendableService;
import org.apache.camel.impl.ServiceSupport;
import org.apache.commons.logging.Log;
@@ -40,6 +41,9 @@ public final class ServiceHelper {
private ServiceHelper() {
}
+ /**
+ * Starts all of the given services
+ */
public static void startService(Object value) throws Exception {
if (value instanceof Service) {
Service service = (Service)value;
@@ -84,6 +88,9 @@ public final class ServiceHelper {
stopServices(list);
}
+ /**
+ * Stops all of the given services, throwing the first exception caught
+ */
public static void stopService(Object value) throws Exception {
if (value instanceof Service) {
Service service = (Service)value;
@@ -110,7 +117,72 @@ public final class ServiceHelper {
}
service.stop();
} catch (Exception e) {
- LOG.debug("Caught exception shutting down: " + e, e);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Caught exception stopping service: " +
service, e);
+ }
+ if (firstException == null) {
+ firstException = e;
+ }
+ }
+ }
+ }
+ if (firstException != null) {
+ throw firstException;
+ }
+ }
+
+ /**
+ * Stops and shutdowns all of the given services, throwing the first
exception caught
+ */
+ public static void stopAndShutdownServices(Object... services) throws
Exception {
+ List<Object> list = Arrays.asList(services);
+ stopAndShutdownServices(list);
+ }
+
+ /**
+ * Stops and shutdowns all of the given services, throwing the first
exception caught
+ */
+ public static void stopAndShutdownService(Object value) throws Exception {
+ if (value instanceof Service) {
+ // must stop it first
+ stopService(value);
+ }
+
+ // then try to shutdown
+ if (value instanceof ShutdownableService) {
+ ShutdownableService service = (ShutdownableService)value;
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Shutting down service " + value);
+ }
+ service.shutdown();
+ } else if (value instanceof Collection) {
+ stopAndShutdownServices((Collection<?>)value);
+ }
+ }
+
+ /**
+ * Stops and shutdowns all of the given services, throwing the first
exception caught
+ */
+ public static void stopAndShutdownServices(Collection<?> services) throws
Exception {
+ Exception firstException = null;
+
+ for (Object value : services) {
+
+ // must stop it first
+ stopService(value);
+
+ // then try to shutdown
+ if (value instanceof ShutdownableService) {
+ ShutdownableService service = (ShutdownableService)value;
+ try {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Shutting down service: " + service);
+ }
+ service.shutdown();
+ } catch (Exception e) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Caught exception shutting down service: " +
service, e);
+ }
if (firstException == null) {
firstException = e;
}
Modified:
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/ServiceSupportTest.java
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/impl/ServiceSupportTest.java?rev=922296&r1=922295&r2=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/ServiceSupportTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/impl/ServiceSupportTest.java
Fri Mar 12 15:27:40 2010
@@ -48,4 +48,48 @@ public class ServiceSupportTest extends
assertEquals(false, service.isStarted());
assertEquals(false, service.isStarting());
}
+
+ private class MyShutdownService extends ServiceSupport {
+
+ private boolean shutdown;
+
+ @Override
+ protected void doStart() throws Exception {
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ shutdown = true;
+ }
+
+ public boolean isShutdown() {
+ return shutdown;
+ }
+ }
+
+ public void testServiceSupportShutdown() throws Exception {
+ MyShutdownService service = new MyShutdownService();
+ service.start();
+
+ assertEquals(true, service.isStarted());
+ assertEquals(false, service.isStarting());
+ assertEquals(false, service.isStopped());
+ assertEquals(false, service.isStopping());
+ assertEquals(false, service.isShutdown());
+
+ service.shutdown();
+
+ assertEquals(true, service.isStopped());
+ assertEquals(false, service.isStopping());
+ assertEquals(false, service.isStarted());
+ assertEquals(false, service.isStarting());
+
+ assertEquals(true, service.isShutdown());
+ }
+
+
}
Copied:
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateShutdownThreadPoolTest.java
(from r922137,
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSimpleSizeTest.java)
URL:
http://svn.apache.org/viewvc/camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateShutdownThreadPoolTest.java?p2=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateShutdownThreadPoolTest.java&p1=camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSimpleSizeTest.java&r1=922137&r2=922296&rev=922296&view=diff
==============================================================================
---
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateSimpleSizeTest.java
(original)
+++
camel/trunk/camel-core/src/test/java/org/apache/camel/processor/aggregator/AggregateShutdownThreadPoolTest.java
Fri Mar 12 15:27:40 2010
@@ -16,23 +16,78 @@
*/
package org.apache.camel.processor.aggregator;
+import java.util.concurrent.ExecutorService;
+
import org.apache.camel.ContextTestSupport;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.processor.BodyInAggregatingStrategy;
+import org.apache.camel.util.concurrent.ExecutorServiceHelper;
/**
* @version $Revision$
*/
-public class AggregateSimpleSizeTest extends ContextTestSupport {
+public class AggregateShutdownThreadPoolTest extends ContextTestSupport {
+
+ private ExecutorService myPool =
ExecutorServiceHelper.newCachedThreadPool("myPool", true);
- public void testAggregateSimpleSize() throws Exception {
+ public void testAggregateShutdownDefaultThreadPoolTest() throws Exception {
getMockEndpoint("mock:aggregated").expectedBodiesReceived("A+B+C");
- template.sendBodyAndHeader("direct:start", "A", "id", 123);
- template.sendBodyAndHeader("direct:start", "B", "id", 123);
- template.sendBodyAndHeader("direct:start", "C", "id", 123);
+ template.sendBodyAndHeader("direct:foo", "A", "id", 123);
+ template.sendBodyAndHeader("direct:foo", "B", "id", 123);
+ template.sendBodyAndHeader("direct:foo", "C", "id", 123);
+
+ assertMockEndpointsSatisfied();
+
+ context.stopRoute("foo");
+
+ resetMocks();
+
+ context.startRoute("foo");
+
+ getMockEndpoint("mock:aggregated").expectedBodiesReceived("D+E+F");
+
+ template.sendBodyAndHeader("direct:foo", "D", "id", 123);
+ template.sendBodyAndHeader("direct:foo", "E", "id", 123);
+ template.sendBodyAndHeader("direct:foo", "F", "id", 123);
assertMockEndpointsSatisfied();
+
+ context.stop();
+ }
+
+ public void testAggregateShutdownCustomThreadPoolTest() throws Exception {
+ assertEquals(false, myPool.isShutdown());
+
+ getMockEndpoint("mock:aggregated").expectedBodiesReceived("A+B+C");
+
+ template.sendBodyAndHeader("direct:bar", "A", "id", 123);
+ template.sendBodyAndHeader("direct:bar", "B", "id", 123);
+ template.sendBodyAndHeader("direct:bar", "C", "id", 123);
+
+ assertMockEndpointsSatisfied();
+ assertEquals(false, myPool.isShutdown());
+
+ context.stopRoute("bar");
+ assertEquals(false, myPool.isShutdown());
+
+ resetMocks();
+
+ context.startRoute("bar");
+ assertEquals(false, myPool.isShutdown());
+
+ getMockEndpoint("mock:aggregated").expectedBodiesReceived("D+E+F");
+
+ template.sendBodyAndHeader("direct:bar", "D", "id", 123);
+ template.sendBodyAndHeader("direct:bar", "E", "id", 123);
+ template.sendBodyAndHeader("direct:bar", "F", "id", 123);
+
+ assertMockEndpointsSatisfied();
+ assertEquals(false, myPool.isShutdown());
+
+ context.stop();
+ // now it should be shutdown when CamelContext is stopped/shutdown
+ assertEquals(true, myPool.isShutdown());
}
@Override
@@ -40,15 +95,13 @@ public class AggregateSimpleSizeTest ext
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- // START SNIPPET: e1
- from("direct:start")
- // aggregate all exchanges correlated by the id header.
- // Aggregate them using the BodyInAggregatingStrategy
strategy which
- // and after 3 messages has been aggregated then complete
the aggregation
- // and send it to mock:aggregated
+ from("direct:foo").routeId("foo")
.aggregate(header("id"), new
BodyInAggregatingStrategy()).completionSize(3)
.to("mock:aggregated");
- // END SNIPPET: e1
+
+ from("direct:bar").routeId("bar")
+ .aggregate(header("id"), new
BodyInAggregatingStrategy()).executorService(myPool).completionSize(3)
+ .to("mock:aggregated");
}
};
}
