Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Now encapsulating EndpointTask within the EndpointExecutor. Preparing for overhaul of consumer + dispatcher functionality to support passive and active sources.

Browse Source
This commit is contained in:
Mark Fisher
2007-12-21 18:22:20 +00:00
parent ae9f7273d7
commit 184a91e3e3
3 changed files with 82 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

99
spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java
View File

@@ -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<Runnable>());
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<Runnable>());
}
@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;
}
}
}

33
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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;
}
}
}
}