proper shutdown of executor in AbstractChannelAdapter and EndpointExecutor

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

@@ -23,8 +23,11 @@ import java.util.concurrent.TimeUnit;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 import org.springframework.util.Assert;
 
 /**
@@ -32,7 +35,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class EndpointExecutor {
+public class EndpointExecutor implements Lifecycle {
 
 	private Log logger = LogFactory.getLog(this.getClass());
 
@@ -40,6 +43,14 @@ public class EndpointExecutor {
 
 	private ThreadPoolExecutor threadPoolExecutor;
 
+	private int corePoolSize;
+
+	private int maxPoolSize;
+
+	private volatile boolean running;
+
+	private Object lifecycleMonitor = new Object();
+
 	private int successiveErrorCount;
 
 	private int successiveErrorThreshold = -1;
@@ -49,13 +60,51 @@ public class EndpointExecutor {
 	private int totalErrorThreshold = -1;
 
 
-	public EndpointExecutor(MessageEndpoint endpoint, int corePoolSize, int maximumPoolSize) {
+	public EndpointExecutor(MessageEndpoint endpoint, int corePoolSize, int maxPoolSize) {
 		Assert.notNull(endpoint, "'endpoint' must not be null");
+		Assert.isTrue(corePoolSize > 0, "'corePoolSize' must be at least 1");
+		Assert.isTrue(maxPoolSize > 0, "'maxPoolSize' must be at least 1");
+		Assert.isTrue(maxPoolSize >= corePoolSize, "'corePoolSize' cannot exceed 'maxPoolSize'");
 		this.endpoint = endpoint;
-		this.threadPoolExecutor = new EndpointThreadPoolExecutor(corePoolSize, maximumPoolSize);
+		this.corePoolSize = corePoolSize;
+		this.maxPoolSize = maxPoolSize;
+	}
+
+	public void setCorePoolSize(int corePoolSize) {
+		this.corePoolSize = corePoolSize;
+	}
+
+	public void setMaxPoolSize(int maxPoolSize) {
+		this.maxPoolSize = maxPoolSize;
+	}
+
+	public boolean isRunning() {
+		return this.running;
+	}
+
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!this.running) {
+				this.threadPoolExecutor = new EndpointThreadPoolExecutor(this.corePoolSize, this.maxPoolSize);
+			}
+			this.running = true;
+		}
+	}
+
+	public void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (this.isRunning()) {
+				this.threadPoolExecutor.shutdown();
+				this.threadPoolExecutor = null;
+			}
+			this.running = false;
+		}
 	}
 
 	public void executeTask(Message<?> message) {
+		if (threadPoolExecutor == null) {
+			throw new MessageHandlingException("executor is not running");
+		}
 		this.threadPoolExecutor.execute(new EndpointTask(this.endpoint, message));
 	}
 
@@ -119,6 +168,9 @@ public class EndpointExecutor {
 
 		public EndpointThreadPoolExecutor(int corePoolSize, int maximumPoolSize) {
 			super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>());
+			CustomizableThreadFactory threadFactory = new CustomizableThreadFactory();
+			threadFactory.setThreadNamePrefix("endpoint-executor-");
+			this.setThreadFactory(threadFactory);
 		}
 
 		@Override

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

@@ -185,6 +185,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 					channelName + "' endpoint='" + endpointName + "'");
 		}
 		if (this.isRunning()) {
+			endpointExecutor.start();
 			scheduleDispatcherTask(dispatcherTask);
 			if (this.logger.isInfoEnabled()) {
 				logger.info("scheduled dispatcher task: channel='" +
@@ -236,6 +237,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		synchronized (this.lifecycleMonitor) {
 			if (!this.isRunning()) {
 				this.running = true;
+				for (EndpointExecutor executor : endpointExecutors.values()) {
+					executor.start();
+				}
 				for (DispatcherTask task : this.dispatcherTasks) {
 					scheduleDispatcherTask(task);
 				}
@@ -247,6 +251,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
 		synchronized (this.lifecycleMonitor) {
 			if (this.isRunning()) {
 				this.running = false;
+				for (EndpointExecutor executor : endpointExecutors.values()) {
+					executor.stop();
+				}
 				this.dispatcherExecutor.shutdownNow();
 			}
 		}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractChannelAdapter.java

@@ -26,6 +26,7 @@ import java.util.concurrent.TimeoutException;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.MessageHandlingException;
@@ -142,16 +143,17 @@ public abstract class AbstractChannelAdapter implements MessageChannel, Initiali
 			throw new MessageHandlingException("adapter not initialized");
 		}
 		ExecutorService executor = Executors.newSingleThreadExecutor();
-		Future<Message> result = executor.submit(new Callable<Message>() {
-			public Message call() throws Exception {
-				return receive();
-			}
-		});
 		try {
+			Future<Message> result = executor.submit(new Callable<Message>() {
+				public Message call() throws Exception {
+					return receive();
+				}
+			});
 			result.get(timeout, TimeUnit.MILLISECONDS);
 			if (result.isDone()) {
 				return result.get();
 			}
+			result.cancel(true);
 		}
 		catch (InterruptedException e) {
 			Thread.currentThread().interrupt();
@@ -163,7 +165,9 @@ public abstract class AbstractChannelAdapter implements MessageChannel, Initiali
 		catch (ExecutionException e) {
 			throw new MessageHandlingException("Exception occurred in message source", e);
 		}
-		result.cancel(true);
+		finally {
+			executor.shutdown();
+		}
 		return null;
 	}