diff --git a/spring-integration-core/.classpath b/spring-integration-core/.classpath
index 009e67c417..9181e43c30 100644
--- a/spring-integration-core/.classpath
+++ b/spring-integration-core/.classpath
@@ -10,8 +10,8 @@
-
-
+
+
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java
index 087cf0e360..f9e978c91b 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java
@@ -23,17 +23,23 @@ import java.util.concurrent.TimeUnit;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.springframework.integration.bus.MessageBus.EndpointTask;
+import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
/**
- * A subclass of {@link ThreadPoolExecutor} with configurable error thresholds.
+ * Encapsulates a {@link ThreadPoolExecutor} with configurable error thresholds.
*
* @author Mark Fisher
*/
-public class EndpointExecutor extends ThreadPoolExecutor {
+public class EndpointExecutor {
private Log logger = LogFactory.getLog(this.getClass());
+ private MessageEndpoint endpoint;
+
+ private ThreadPoolExecutor threadPoolExecutor;
+
private int successiveErrorCount;
private int successiveErrorThreshold = -1;
@@ -43,8 +49,14 @@ public class EndpointExecutor extends ThreadPoolExecutor {
private int totalErrorThreshold = -1;
- public EndpointExecutor(int corePoolSize, int maximumPoolSize) {
- super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue());
+ public EndpointExecutor(MessageEndpoint endpoint, int corePoolSize, int maximumPoolSize) {
+ Assert.notNull(endpoint, "'endpoint' must not be null");
+ this.endpoint = endpoint;
+ this.threadPoolExecutor = new EndpointThreadPoolExecutor(corePoolSize, maximumPoolSize);
+ }
+
+ public void executeTask(Message message) {
+ this.threadPoolExecutor.execute(new EndpointTask(this.endpoint, message));
}
/**
@@ -65,25 +77,70 @@ public class EndpointExecutor extends ThreadPoolExecutor {
this.totalErrorThreshold = totalErrorThreshold;
}
- @Override
- protected void afterExecute(Runnable r, Throwable t) {
- EndpointTask task = (EndpointTask) r;
- if (task.getError() != null) {
- if (logger.isWarnEnabled()) {
- logger.warn("Exception occurred in endpoint execution", task.getError());
+ public int getActiveCount() {
+ return this.threadPoolExecutor.getActiveCount();
+ }
+
+ public boolean isShutdown() {
+ return this.threadPoolExecutor.isShutdown();
+ }
+
+
+ private static class EndpointTask implements Runnable {
+
+ private MessageEndpoint endpoint;
+
+ private Message message;
+
+ private Throwable error;
+
+
+ EndpointTask(MessageEndpoint endpoint, Message message) {
+ this.endpoint = endpoint;
+ this.message = message;
+ }
+
+ public Throwable getError() {
+ return this.error;
+ }
+
+ public void run() {
+ try {
+ this.endpoint.messageReceived(this.message);
}
- this.successiveErrorCount++;
- this.totalErrorCount++;
- if ((successiveErrorThreshold >= 0 && successiveErrorCount > successiveErrorThreshold)
- || (totalErrorThreshold >= 0 && totalErrorCount > totalErrorThreshold)) {
- if (logger.isInfoEnabled()) {
- logger.info("error threshold exceeded, shutting down now");
- }
- this.shutdownNow();
+ catch (Throwable t) {
+ this.error = t;
}
}
- else {
- successiveErrorCount = 0;
+ }
+
+
+ private class EndpointThreadPoolExecutor extends ThreadPoolExecutor {
+
+ public EndpointThreadPoolExecutor(int corePoolSize, int maximumPoolSize) {
+ super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue());
+ }
+
+ @Override
+ protected void afterExecute(Runnable r, Throwable t) {
+ EndpointTask task = (EndpointTask) r;
+ if (task.getError() != null) {
+ if (logger.isWarnEnabled()) {
+ logger.warn("Exception occurred in endpoint execution", task.getError());
+ }
+ successiveErrorCount++;
+ totalErrorCount++;
+ if ((successiveErrorThreshold >= 0 && successiveErrorCount > successiveErrorThreshold)
+ || (totalErrorThreshold >= 0 && totalErrorCount > totalErrorThreshold)) {
+ if (logger.isInfoEnabled()) {
+ logger.info("error threshold exceeded, shutting down now");
+ }
+ this.shutdownNow();
+ }
+ }
+ else {
+ successiveErrorCount = 0;
+ }
}
}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
index 156dc36c5a..9489b8e1cf 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
@@ -176,7 +176,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
logger.info("activated subscription to channel '" + channelName +
"' for endpoint '" + endpointName + "'");
}
- EndpointExecutor endpointExecutor = new EndpointExecutor(policy.getConcurrency(), policy.getMaxConcurrency());
+ EndpointExecutor endpointExecutor = new EndpointExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency());
endpointExecutors.put(endpoint, endpointExecutor);
DispatcherTask dispatcherTask = new DispatcherTask(channel, endpoint, policy);
this.dispatcherTasks.add(dispatcherTask);
@@ -298,7 +298,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
int attempts = 0;
while (!taskSubmitted) {
try {
- executor.execute(new EndpointTask(this.endpoint, message));
+ executor.executeTask(message);
taskSubmitted = true;
}
catch (RejectedExecutionException rex) {
@@ -337,33 +337,4 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
}
}
-
- public static class EndpointTask implements Runnable {
-
- private MessageEndpoint endpoint;
-
- private Message message;
-
- private Throwable error;
-
-
- EndpointTask(MessageEndpoint endpoint, Message message) {
- this.endpoint = endpoint;
- this.message = message;
- }
-
- public Throwable getError() {
- return this.error;
- }
-
- public void run() {
- try {
- this.endpoint.messageReceived(this.message);
- }
- catch (Throwable t) {
- this.error = t;
- }
- }
- }
-
}