INT-2166: Add SecurityContext Propagation

JIRA: https://jira.spring.io/browse/INT-2166

* Introduce `ThreadStatePropagationChannelInterceptor` based on the `ExecutorChannelInterceptor`
* Add `SecurityContextPropagationChannelInterceptor`,`SecurityContextCleanupChannelInterceptor`
* Introduce `AbstractExecutorChannel` to utilize `ExecutorChannelInterceptor` logic
* Introduce `MessageHandlingTaskDecorator` to avoid package tangle from `dispatcher` and `channel`
* Introduce `SecurityContextCleanupAdvice` for those cases when we don't get deal with `MessageChannel`s already, but want to have proper way to cleanup `SecurityContext`
* Make `GlobalChannelInterceptorProcessor` as `SmartInitializingSingleton` to avoid `phase` conflicts.
* Fix `MessagingAnnotationPostProcessor` to use `beanFactory.initializeBean(endpoint, endpointBeanName);` instead of manual `start()` invocation bypassing the `phase` logic, hence having a bug, when endpoints have been started very early
* Optimise `AbstractPollableChannel` to use `size` field from `ChannelInterceptorList` instead of `size()` from `Collection<?>`
* Fix `AnnotatedEndpointActivationTests` extracting separate component for annotation configuration instead of using test class directly. This caused very late Messaging Annotations process on that class
* Fix typo in the `spring-integration-jdbc-4.2.xsd`
* Remove some `SOUT`s throughout the project

TODO Docs

PR Comments:

* Remove redundant `AbstractExecutorChannel#executorInterceptors` and make logic based on the `super.interceptors`
 * Fix wrong imports order
 * JavaDocs for `ThreadStatePropagationChannelInterceptor`
 * Docs for `SecurityContext` propagation

INT-3593: Fix FTP PartialSuccess Tests

JIRA: https://jira.spring.io/browse/INT-3593

Sort the files for the MPUT tests.

INT-2166: Add SecurityContext Propagation

JIRA: https://jira.spring.io/browse/INT-2166

* Introduce `ThreadStatePropagationChannelInterceptor` based on the `ExecutorChannelInterceptor`
* Add `SecurityContextPropagationChannelInterceptor`,`SecurityContextCleanupChannelInterceptor`
* Introduce `AbstractExecutorChannel` to utilize `ExecutorChannelInterceptor` logic
* Introduce `MessageHandlingTaskDecorator` to avoid package tangle from `dispatcher` and `channel`
* Introduce `SecurityContextCleanupAdvice` for those cases when we don't get deal with `MessageChannel`s already, but want to have proper way to cleanup `SecurityContext`
* Make `GlobalChannelInterceptorProcessor` as `SmartInitializingSingleton` to avoid `phase` conflicts.
* Fix `MessagingAnnotationPostProcessor` to use `beanFactory.initializeBean(endpoint, endpointBeanName);` instead of manual `start()` invocation bypassing the `phase` logic, hence having a bug, when endpoints have been started very early
* Optimise `AbstractPollableChannel` to use `size` field from `ChannelInterceptorList` instead of `size()` from `Collection<?>`
* Fix `AnnotatedEndpointActivationTests` extracting separate component for annotation configuration instead of using test class directly. This caused very late Messaging Annotations process on that class
* Fix typo in the `spring-integration-jdbc-4.2.xsd`
* Remove some `SOUT`s throughout the project

TODO Docs

PR Comments:

* Remove redundant `AbstractExecutorChannel#executorInterceptors` and make logic based on the `super.interceptors`
 * Fix wrong imports order
 * JavaDocs for `ThreadStatePropagationChannelInterceptor`
 * Docs for `SecurityContext` propagation

Doc Polishing

Address PR comments

Address PR comments

* Extract `ExecutorChannelInterceptor` logic in the `PollingConsumer`
to have an ability to invoke `afterMessageHandled()` on the TaskScheduler's Thread
for example for the `SecurityContext` clean up
* Get rid of all that redundant "clean up" stuff
* Docs polishing

Fix `NPE` in the `PollingConsumer`

Introduce `ExecutorChannelInterceptorAware` to avoid iterators on each message

Polishing; Docs, Sonar

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java

@@ -0,0 +1,209 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ *  Licensed 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.springframework.integration.channel;
+
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.Executor;
+
+import org.springframework.integration.dispatcher.AbstractDispatcher;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
+import org.springframework.messaging.support.MessageHandlingRunnable;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * The {@link AbstractSubscribableChannel} base implementation for those inheritors
+ * which logic may be based on the {@link Executor}.
+ * <p>
+ * Utilizes common operations for the {@link AbstractDispatcher}.
+ * <p>
+ * Implements the {@link ExecutorChannelInterceptor}s logic when the message handling
+ * is handed to the {@link Executor#execute(Runnable)}.
+ *
+ * @author Artem Bilan
+ * @see ExecutorChannel
+ * @see PublishSubscribeChannel
+ * @since 4.2
+ */
+public abstract class AbstractExecutorChannel extends AbstractSubscribableChannel
+		implements ExecutorChannelInterceptorAware {
+
+	protected volatile Executor executor;
+
+	protected volatile AbstractDispatcher dispatcher;
+
+	protected volatile Integer maxSubscribers;
+
+	protected volatile int executorInterceptorsSize;
+
+	public AbstractExecutorChannel(Executor executor) {
+		this.executor = executor;
+	}
+
+	/**
+	 * Specify the maximum number of subscribers supported by the
+	 * channel's dispatcher.
+	 *
+	 * @param maxSubscribers The maximum number of subscribers allowed.
+	 */
+	public void setMaxSubscribers(int maxSubscribers) {
+		this.maxSubscribers = maxSubscribers;
+		this.dispatcher.setMaxSubscribers(maxSubscribers);
+	}
+
+	@Override
+	public void setInterceptors(List<ChannelInterceptor> interceptors) {
+		super.setInterceptors(interceptors);
+		for (ChannelInterceptor interceptor : interceptors) {
+			if (interceptor instanceof ExecutorChannelInterceptor) {
+				this.executorInterceptorsSize++;
+			}
+		}
+	}
+
+	@Override
+	public void addInterceptor(ChannelInterceptor interceptor) {
+		super.addInterceptor(interceptor);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize++;
+		}
+	}
+
+	@Override
+	public void addInterceptor(int index, ChannelInterceptor interceptor) {
+		super.addInterceptor(index, interceptor);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize++;
+		}
+	}
+
+	@Override
+	public boolean removeInterceptor(ChannelInterceptor interceptor) {
+		boolean removed = super.removeInterceptor(interceptor);
+		if (removed && interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize--;
+		}
+		return removed;
+	}
+
+	@Override
+	public ChannelInterceptor removeInterceptor(int index) {
+		ChannelInterceptor interceptor = super.removeInterceptor(index);
+		if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize--;
+		}
+		return interceptor;
+	}
+
+	@Override
+	public boolean hasExecutorInterceptors() {
+		return this.executorInterceptorsSize > 0;
+	}
+
+	protected class MessageHandlingTask implements Runnable {
+
+		private final MessageHandlingRunnable delegate;
+
+		public MessageHandlingTask(MessageHandlingRunnable task) {
+			this.delegate = task;
+		}
+
+		@Override
+		public void run() {
+			Message<?> message = this.delegate.getMessage();
+			MessageHandler messageHandler = this.delegate.getMessageHandler();
+			Assert.notNull(messageHandler, "'messageHandler' must not be null");
+			Deque<ExecutorChannelInterceptor> interceptorStack = null;
+			try {
+				if (executorInterceptorsSize > 0) {
+					interceptorStack = new ArrayDeque<ExecutorChannelInterceptor>();
+					message = applyBeforeHandle(message, interceptorStack);
+					if (message == null) {
+						return;
+					}
+				}
+				messageHandler.handleMessage(message);
+				if (!CollectionUtils.isEmpty(interceptorStack)) {
+					triggerAfterMessageHandled(message, null, interceptorStack);
+				}
+			}
+			catch (Exception ex) {
+				if (!CollectionUtils.isEmpty(interceptorStack)) {
+					triggerAfterMessageHandled(message, ex, interceptorStack);
+				}
+				if (ex instanceof MessagingException) {
+					throw (MessagingException) ex;
+				}
+				String description = "Failed to handle " + message + " to " + this + " in " + messageHandler;
+				throw new MessageDeliveryException(message, description, ex);
+			}
+			catch (Error ex) {//NOSONAR - ok, we re-throw below
+				if (!CollectionUtils.isEmpty(interceptorStack)) {
+					String description = "Failed to handle " + message + " to " + this + " in " + messageHandler;
+					triggerAfterMessageHandled(message, new MessageDeliveryException(message, description, ex),
+							interceptorStack);
+				}
+				throw ex;
+			}
+		}
+
+		private Message<?> applyBeforeHandle(Message<?> message, Deque<ExecutorChannelInterceptor> interceptorStack) {
+			for (ChannelInterceptor interceptor : AbstractExecutorChannel.this.interceptors.interceptors) {
+				if (interceptor instanceof ExecutorChannelInterceptor) {
+					ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor;
+					message = executorInterceptor.beforeHandle(message, AbstractExecutorChannel.this,
+							this.delegate.getMessageHandler());
+					if (message == null) {
+						if (isLoggingEnabled() && logger.isDebugEnabled()) {
+							logger.debug(executorInterceptor.getClass().getSimpleName()
+									+ " returned null from beforeHandle, i.e. precluding the send.");
+						}
+						triggerAfterMessageHandled(null, null, interceptorStack);
+						return null;
+					}
+					interceptorStack.add(executorInterceptor);
+				}
+			}
+			return message;
+		}
+
+		private void triggerAfterMessageHandled(Message<?> message, Exception ex,
+												Deque<ExecutorChannelInterceptor> interceptorStack) {
+			Iterator<ExecutorChannelInterceptor> iterator = interceptorStack.descendingIterator();
+			while (iterator.hasNext()) {
+				ExecutorChannelInterceptor interceptor = iterator.next();
+				try {
+					interceptor.afterMessageHandled(message, AbstractExecutorChannel.this,
+							this.delegate.getMessageHandler(), ex);
+				}
+				catch (Throwable ex2) {//NOSONAR
+					logger.error("Exception from afterMessageHandled in " + interceptor, ex2);
+				}
+			}
+		}
+
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -65,7 +65,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics,
 		ConfigurableMetricsAware<AbstractMessageChannelMetrics> {
 
-	private final ChannelInterceptorList interceptors;
+	protected final ChannelInterceptorList interceptors;
 
 	private final Comparator<Object> orderComparator = new OrderComparator();
 
@@ -520,7 +520,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 		private final Log logger;
 
-		private final List<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();
+		protected final List<ChannelInterceptor> interceptors = new CopyOnWriteArrayList<ChannelInterceptor>();
 
 		private volatile int size;
 

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java

@@ -18,11 +18,14 @@ package org.springframework.integration.channel;
 
 import java.util.ArrayDeque;
 import java.util.Deque;
+import java.util.List;
 
 import org.springframework.integration.channel.management.PollableChannelManagement;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Base class for all pollable channels.
@@ -30,10 +33,12 @@ import org.springframework.messaging.support.ChannelInterceptor;
  * @author Mark Fisher
  * @author Artem Bilan
  * @author Gary Russell
+ * @author Artem Bilan
  */
-public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel,
-		PollableChannelManagement {
+public abstract class AbstractPollableChannel extends AbstractMessageChannel
+		implements PollableChannel, PollableChannelManagement, ExecutorChannelInterceptorAware {
 
+	protected volatile int executorInterceptorsSize;
 
 	@Override
 	public int getReceiveCount() {
@@ -90,7 +95,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 			if (logger.isTraceEnabled()) {
 				logger.trace("preReceive on channel '" + this + "'");
 			}
-			if (interceptorList.getInterceptors().size() > 0) {
+			if (interceptorList.getSize() > 0) {
 				interceptorStack = new ArrayDeque<ChannelInterceptor>();
 
 				if (!interceptorList.preReceive(this, interceptorStack)) {
@@ -108,7 +113,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 			else if (logger.isTraceEnabled()) {
 				logger.trace("postReceive on channel '" + this + "', message is null");
 			}
-			if (interceptorStack != null) {
+			if (!CollectionUtils.isEmpty(interceptorStack)) {
 				message = interceptorList.postReceive(message, this);
 				interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
 			}
@@ -118,13 +123,62 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 			if (countsEnabled && !counted) {
 				getMetrics().afterError();
 			}
-			if (interceptorStack != null) {
+			if (!CollectionUtils.isEmpty(interceptorStack)) {
 				interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
 			}
 			throw e;
 		}
 	}
 
+	@Override
+	public void setInterceptors(List<ChannelInterceptor> interceptors) {
+		super.setInterceptors(interceptors);
+		for (ChannelInterceptor interceptor : interceptors) {
+			if (interceptor instanceof ExecutorChannelInterceptor) {
+				this.executorInterceptorsSize++;
+			}
+		}
+	}
+
+	@Override
+	public void addInterceptor(ChannelInterceptor interceptor) {
+		super.addInterceptor(interceptor);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize++;
+		}
+	}
+
+	@Override
+	public void addInterceptor(int index, ChannelInterceptor interceptor) {
+		super.addInterceptor(index, interceptor);
+		if (interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize++;
+		}
+	}
+
+	@Override
+	public boolean removeInterceptor(ChannelInterceptor interceptor) {
+		boolean removed = super.removeInterceptor(interceptor);
+		if (removed && interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize--;
+		}
+		return removed;
+	}
+
+	@Override
+	public ChannelInterceptor removeInterceptor(int index) {
+		ChannelInterceptor interceptor = super.removeInterceptor(index);
+		if (interceptor != null && interceptor instanceof ExecutorChannelInterceptor) {
+			this.executorInterceptorsSize--;
+		}
+		return interceptor;
+	}
+
+	@Override
+	public boolean hasExecutorInterceptors() {
+		return this.executorInterceptorsSize > 0;
+	}
+
 	/**
 	 * Subclasses must implement this method. A non-negative timeout indicates
 	 * how long to wait if the channel is empty (if the value is 0, it must

spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,11 +20,13 @@ import java.util.concurrent.Executor;
 
 import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.dispatcher.LoadBalancingStrategy;
+import org.springframework.integration.dispatcher.MessageHandlingTaskDecorator;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
 import org.springframework.integration.dispatcher.UnicastingDispatcher;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.MessageHandlingRunnable;
 import org.springframework.util.Assert;
 import org.springframework.util.ErrorHandler;
 
@@ -45,25 +47,17 @@ import org.springframework.util.ErrorHandler;
  * @author Artem Bilan
  * @since 1.0.3
  */
-public class ExecutorChannel extends AbstractSubscribableChannel {
-
-	private volatile UnicastingDispatcher dispatcher;
-
-	private volatile Executor executor;
+public class ExecutorChannel extends AbstractExecutorChannel {
 
 	private volatile boolean failover = true;
 
-	private volatile Integer maxSubscribers;
-
 	private volatile LoadBalancingStrategy loadBalancingStrategy;
 
-
 	/**
 	 * Create an ExecutorChannel that delegates to the provided
 	 * {@link Executor} when dispatching Messages.
 	 * <p>
 	 * The Executor must not be null.
-	 *
 	 * @param executor The executor.
 	 */
 	public ExecutorChannel(Executor executor) {
@@ -75,46 +69,34 @@ public class ExecutorChannel extends AbstractSubscribableChannel {
 	 * delegates to the provided {@link Executor} when dispatching Messages.
 	 * <p>
 	 * The Executor must not be null.
-	 *
 	 * @param executor The executor.
 	 * @param loadBalancingStrategy The load balancing strategy implementation.
 	 */
 	public ExecutorChannel(Executor executor, LoadBalancingStrategy loadBalancingStrategy) {
+		super(executor);
 		Assert.notNull(executor, "executor must not be null");
-		this.executor = executor;
-		this.dispatcher = new UnicastingDispatcher(executor);
+		UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(executor);
 		if (loadBalancingStrategy != null) {
 			this.loadBalancingStrategy = loadBalancingStrategy;
-			this.dispatcher.setLoadBalancingStrategy(loadBalancingStrategy);
+			unicastingDispatcher.setLoadBalancingStrategy(loadBalancingStrategy);
 		}
+		this.dispatcher = unicastingDispatcher;
 	}
 
 
 	/**
 	 * Specify whether the channel's dispatcher should have failover enabled.
 	 * By default, it will. Set this value to 'false' to disable it.
-	 *
 	 * @param failover The failover boolean.
 	 */
 	public void setFailover(boolean failover) {
 		this.failover = failover;
-		this.dispatcher.setFailover(failover);
-	}
-
-	/**
-	 * Specify the maximum number of subscribers supported by the
-	 * channel's dispatcher.
-	 *
-	 * @param maxSubscribers The maximum number of subscribers allowed.
-	 */
-	public void setMaxSubscribers(int maxSubscribers) {
-		this.maxSubscribers = maxSubscribers;
-		this.dispatcher.setMaxSubscribers(maxSubscribers);
+		getDispatcher().setFailover(failover);
 	}
 
 	@Override
 	protected UnicastingDispatcher getDispatcher() {
-		return this.dispatcher;
+		return (UnicastingDispatcher) this.dispatcher;
 	}
 
 	@Override
@@ -124,15 +106,33 @@ public class ExecutorChannel extends AbstractSubscribableChannel {
 					new BeanFactoryChannelResolver(this.getBeanFactory()));
 			this.executor = new ErrorHandlingTaskExecutor(this.executor, errorHandler);
 		}
-		this.dispatcher = new UnicastingDispatcher(this.executor);
-		this.dispatcher.setFailover(this.failover);
+		UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher(this.executor);
+		unicastingDispatcher.setFailover(this.failover);
 		if (this.maxSubscribers == null) {
-			this.maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
+			this.maxSubscribers =
+					getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_UNICAST_SUBSCRIBERS, Integer.class);
 		}
-		this.dispatcher.setMaxSubscribers(this.maxSubscribers);
+		unicastingDispatcher.setMaxSubscribers(this.maxSubscribers);
 		if (this.loadBalancingStrategy != null) {
-			this.dispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy);
+			unicastingDispatcher.setLoadBalancingStrategy(this.loadBalancingStrategy);
 		}
+
+		unicastingDispatcher.setMessageHandlingTaskDecorator(new MessageHandlingTaskDecorator() {
+
+			@Override
+			public Runnable decorate(MessageHandlingRunnable task) {
+				if (ExecutorChannel.this.executorInterceptorsSize > 0) {
+					return new MessageHandlingTask(task);
+				}
+				else {
+					return task;
+				}
+			}
+
+		});
+
+		this.dispatcher = unicastingDispatcher;
 	}
 
+
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/ExecutorChannelInterceptorAware.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ *  Licensed 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.springframework.integration.channel;
+
+/**
+ * The {@link ChannelInterceptorAware} extension for the cases when
+ * the {@link org.springframework.messaging.support.ExecutorChannelInterceptor}s
+ * may have reason (e.g. {@link ExecutorChannel} or {@link QueueChannel})
+ * and the implementors require to know if they should make the
+ * {@link org.springframework.messaging.support.ExecutorChannelInterceptor}
+ * or not.
+ * @author Artem Bilan
+ * @since 4.2
+ */
+public interface ExecutorChannelInterceptorAware extends ChannelInterceptorAware {
+
+	boolean hasExecutorInterceptors();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,8 +20,10 @@ import java.util.concurrent.Executor;
 
 import org.springframework.integration.context.IntegrationProperties;
 import org.springframework.integration.dispatcher.BroadcastingDispatcher;
+import org.springframework.integration.dispatcher.MessageHandlingTaskDecorator;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.util.ErrorHandlingTaskExecutor;
+import org.springframework.messaging.support.MessageHandlingRunnable;
 import org.springframework.util.ErrorHandler;
 
 /**
@@ -30,12 +32,9 @@ import org.springframework.util.ErrorHandler;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
-public class PublishSubscribeChannel extends AbstractSubscribableChannel {
-
-	private volatile BroadcastingDispatcher dispatcher;
-
-	private volatile Executor executor;
+public class PublishSubscribeChannel extends AbstractExecutorChannel {
 
 	private volatile ErrorHandler errorHandler;
 
@@ -45,9 +44,6 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 
 	private volatile int minSubscribers;
 
-	private volatile Integer maxSubscribers;
-
-
 	/**
 	 * Create a PublishSubscribeChannel that will use an {@link Executor}
 	 * to invoke the handlers. If this is null, each invocation will occur in
@@ -56,7 +52,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 	 * @param executor The executor.
 	 */
 	public PublishSubscribeChannel(Executor executor) {
-		this.executor = executor;
+		super(executor);
 		this.dispatcher = new BroadcastingDispatcher(executor);
 	}
 
@@ -98,12 +94,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 	 * ignored. By default this is <code>false</code> meaning that an Exception
 	 * will be thrown whenever a handler fails. To override this and suppress
 	 * Exceptions, set the value to <code>true</code>.
-	 *
 	 * @param ignoreFailures true if failures should be ignored.
 	 */
 	public void setIgnoreFailures(boolean ignoreFailures) {
 		this.ignoreFailures = ignoreFailures;
-		this.getDispatcher().setIgnoreFailures(ignoreFailures);
+		getDispatcher().setIgnoreFailures(ignoreFailures);
 	}
 
 	/**
@@ -113,23 +108,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 	 * <em>not</em> be applied. If planning to use an Aggregator downstream
 	 * with the default correlation and completion strategies, you should set
 	 * this flag to <code>true</code>.
-	 *
 	 * @param applySequence true if the sequence information should be applied.
 	 */
 	public void setApplySequence(boolean applySequence) {
 		this.applySequence = applySequence;
-		this.getDispatcher().setApplySequence(applySequence);
-	}
-
-	/**
-	 * Specify the maximum number of subscribers supported by the
-	 * channel's dispatcher.
-	 *
-	 * @param maxSubscribers The maximum number of subscribers allowed.
-	 */
-	public void setMaxSubscribers(int maxSubscribers) {
-		this.maxSubscribers = maxSubscribers;
-		this.getDispatcher().setMaxSubscribers(maxSubscribers);
+		getDispatcher().setApplySequence(applySequence);
 	}
 
 	/**
@@ -141,7 +124,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 	 */
 	public void setMinSubscribers(int minSubscribers) {
 		this.minSubscribers = minSubscribers;
-		this.getDispatcher().setMinSubscribers(minSubscribers);
+		getDispatcher().setMinSubscribers(minSubscribers);
 	}
 
 	/**
@@ -160,20 +143,36 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 				this.executor = new ErrorHandlingTaskExecutor(this.executor, this.errorHandler);
 			}
 			this.dispatcher = new BroadcastingDispatcher(this.executor);
-			this.dispatcher.setIgnoreFailures(this.ignoreFailures);
-			this.dispatcher.setApplySequence(this.applySequence);
-			this.dispatcher.setMinSubscribers(this.minSubscribers);
+			getDispatcher().setIgnoreFailures(this.ignoreFailures);
+			getDispatcher().setApplySequence(this.applySequence);
+			getDispatcher().setMinSubscribers(this.minSubscribers);
 		}
 		if (this.maxSubscribers == null) {
-			Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
+			Integer maxSubscribers =
+					getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
 			this.setMaxSubscribers(maxSubscribers);
 		}
-		this.dispatcher.setBeanFactory(this.getBeanFactory());
+		getDispatcher().setBeanFactory(this.getBeanFactory());
+
+		getDispatcher().setMessageHandlingTaskDecorator(new MessageHandlingTaskDecorator() {
+
+			@Override
+			public Runnable decorate(MessageHandlingRunnable task) {
+				if (PublishSubscribeChannel.this.executorInterceptorsSize > 0) {
+					return new MessageHandlingTask(task);
+				}
+				else {
+					return task;
+				}
+			}
+
+		});
+
 	}
 
 	@Override
 	protected BroadcastingDispatcher getDispatcher() {
-		return this.dispatcher;
+		return (BroadcastingDispatcher) this.dispatcher;
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ThreadStatePropagationChannelInterceptor.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ *  Licensed 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.springframework.integration.channel.interceptor;
+
+import java.io.Serializable;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
+
+/**
+ * The {@link ExecutorChannelInterceptor} implementation responsible for
+ * the {@link Thread} (any?) state propagation from one message flow's thread to another
+ * through the {@link MessageChannel}s involved in the flow.
+ * <p>
+ * The propagation is done from the {@link #preSend(Message, MessageChannel)}
+ * implementation using some internal {@link Message} extension which keeps the message
+ * to send and the state to propagate.
+ * <p>
+ * The propagated state context extraction and population is done from the {@link #postReceive}
+ * implementation for the {@link org.springframework.messaging.PollableChannel}s, and from
+ * the {@link #beforeHandle} for the
+ * {@link org.springframework.integration.channel.AbstractExecutorChannel}s and
+ * {@link org.springframework.messaging.support.ExecutorSubscribableChannel}s
+ * <p>
+ * Important. Any further interceptor, which modifies the message to send
+ * (e.g. {@code MessageBuilder.withPayload(...)...build()}), may drop the state to propagate.
+ * Such kind of interceptors combination should be revised properly.
+ * In most cases the interceptors reordering is enough to overcome the issue.
+ *
+ * @param <S> the propagated state object type.
+ *
+ * @author Artem Bilan
+ * @since 4.2
+ */
+public abstract class ThreadStatePropagationChannelInterceptor<S extends Serializable>
+		extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {
+
+	@Override
+	public final Message<?> preSend(Message<?> message, MessageChannel channel) {
+		S threadContext = obtainPropagatingContext(message, channel);
+		if (threadContext != null) {
+			return new MessageWithThreadState<S>(message, threadContext);
+		}
+		else {
+			return message;
+		}
+	}
+
+	@Override
+	@SuppressWarnings("unchecked")
+	public final Message<?> postReceive(Message<?> message, MessageChannel channel) {
+		if (message != null && message instanceof MessageWithThreadState) {
+			MessageWithThreadState<S> messageWithThreadState = (MessageWithThreadState<S>) message;
+			Message<?> messageToHandle = messageWithThreadState.message;
+			populatePropagatedContext(messageWithThreadState.state, messageToHandle, channel);
+			return messageToHandle;
+		}
+		return message;
+	}
+
+	@Override
+	public final Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
+		return postReceive(message, channel);
+	}
+
+	@Override
+	public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler,
+										  Exception ex) {
+		// No-op
+	}
+
+	protected abstract S obtainPropagatingContext(Message<?> message, MessageChannel channel);
+
+	protected abstract void populatePropagatedContext(S state, Message<?> message, MessageChannel channel);
+
+
+	private static class MessageWithThreadState<S> implements Message<Object>, Serializable {
+
+		private static final long serialVersionUID = 1548216539234073073L;
+
+		final Message<?> message;
+
+		final S state;
+
+		public MessageWithThreadState(Message<?> message, S state) {
+			this.message = message;
+			this.state = state;
+		}
+
+		@Override
+		public Object getPayload() {
+			return this.message.getPayload();
+		}
+
+		@Override
+		public MessageHeaders getHeaders() {
+			return this.message.getHeaders();
+		}
+
+		@Override
+		public String toString() {
+			return "MessageWithThreadState{" +
+					"message=" + message +
+					", state=" + state +
+					'}';
+		}
+
+	}
+
+}
+

spring-integration-core/src/main/java/org/springframework/integration/config/GlobalChannelInterceptorProcessor.java

@@ -32,7 +32,7 @@ import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.beans.factory.ListableBeanFactory;
-import org.springframework.context.SmartLifecycle;
+import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.core.OrderComparator;
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;
@@ -52,21 +52,21 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @since 2.0
  */
-final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, SmartLifecycle {
+final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, SmartInitializingSingleton {
 
 	private static final Log logger = LogFactory.getLog(GlobalChannelInterceptorProcessor.class);
 
 
 	private final OrderComparator comparator = new OrderComparator();
 
-	private final Set<GlobalChannelInterceptorWrapper> positiveOrderInterceptors = new LinkedHashSet<GlobalChannelInterceptorWrapper>();
+	private final Set<GlobalChannelInterceptorWrapper> positiveOrderInterceptors =
+			new LinkedHashSet<GlobalChannelInterceptorWrapper>();
 
-	private final Set<GlobalChannelInterceptorWrapper> negativeOrderInterceptors = new LinkedHashSet<GlobalChannelInterceptorWrapper>();
+	private final Set<GlobalChannelInterceptorWrapper> negativeOrderInterceptors =
+			new LinkedHashSet<GlobalChannelInterceptorWrapper>();
 
 	private ListableBeanFactory beanFactory;
 
-	private volatile boolean processed;
-
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		Assert.isInstanceOf(ListableBeanFactory.class, beanFactory);
@@ -74,51 +74,27 @@ final class GlobalChannelInterceptorProcessor implements BeanFactoryAware, Smart
 	}
 
 	@Override
-	public synchronized void start() {
-		if (!this.processed) {
-			Collection<GlobalChannelInterceptorWrapper> interceptors = this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class).values();
-			if (CollectionUtils.isEmpty(interceptors)) {
-				logger.debug("No global channel interceptors.");
-			}
-			else {
-				for (GlobalChannelInterceptorWrapper channelInterceptor : interceptors) {
-					if (channelInterceptor.getOrder() >= 0) {
-						this.positiveOrderInterceptors.add(channelInterceptor);
-					}
-					else {
-						this.negativeOrderInterceptors.add(channelInterceptor);
-					}
-				}
-				Map<String, ChannelInterceptorAware> channels = this.beanFactory.getBeansOfType(ChannelInterceptorAware.class);
-				for (Entry<String, ChannelInterceptorAware> entry : channels.entrySet()) {
-					this.addMatchingInterceptors(entry.getValue(), entry.getKey());
-				}
-			}
-			this.processed = true;
+	public void afterSingletonsInstantiated() {
+		Collection<GlobalChannelInterceptorWrapper> interceptors =
+				this.beanFactory.getBeansOfType(GlobalChannelInterceptorWrapper.class).values();
+		if (CollectionUtils.isEmpty(interceptors)) {
+			logger.debug("No global channel interceptors.");
+		}
+		else {
+			for (GlobalChannelInterceptorWrapper channelInterceptor : interceptors) {
+				if (channelInterceptor.getOrder() >= 0) {
+					this.positiveOrderInterceptors.add(channelInterceptor);
+				}
+				else {
+					this.negativeOrderInterceptors.add(channelInterceptor);
+				}
+			}
+			Map<String, ChannelInterceptorAware> channels =
+					this.beanFactory.getBeansOfType(ChannelInterceptorAware.class);
+			for (Entry<String, ChannelInterceptorAware> entry : channels.entrySet()) {
+				addMatchingInterceptors(entry.getValue(), entry.getKey());
+			}
 		}
-	}
-
-	@Override
-	public void stop() {
-	}
-
-	@Override
-	public boolean isRunning() {
-		return false;
-	}
-
-	@Override
-	public int getPhase() {
-		return Integer.MIN_VALUE;
-	}
-
-	@Override
-	public boolean isAutoStartup() {
-		return true;
-	}
-
-	@Override
-	public void stop(Runnable callback) {
 	}
 
 	/**

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java

@@ -254,6 +254,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
 			catch (DestinationResolutionException e) {
 				inputChannel = new DirectChannel();
 				this.beanFactory.registerSingleton(inputChannelName, inputChannel);
+				this.beanFactory.initializeBean(inputChannel, inputChannelName);
 				inputChannel = (MessageChannel) this.beanFactory.initializeBean(inputChannel, inputChannelName);
 			}
 			Assert.notNull(inputChannel, "failed to resolve inputChannel '" + inputChannelName + "'");

spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,16 +34,12 @@ import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
-import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.beans.factory.SmartInitializingSingleton;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
-import org.springframework.context.ApplicationEvent;
-import org.springframework.context.ApplicationListener;
 import org.springframework.context.EnvironmentAware;
-import org.springframework.context.Lifecycle;
 import org.springframework.core.annotation.AnnotatedElementUtils;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.core.env.Environment;
@@ -79,8 +75,7 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  */
 public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware,
-		InitializingBean, Lifecycle, ApplicationListener<ApplicationEvent>, EnvironmentAware,
-		SmartInitializingSingleton {
+		InitializingBean, EnvironmentAware, SmartInitializingSingleton {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -91,12 +86,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 	private final Map<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>> postProcessors =
 			new HashMap<Class<? extends Annotation>, MethodAnnotationPostProcessor<?>>();
 
-	private final Set<ApplicationListener<ApplicationEvent>> listeners = new HashSet<ApplicationListener<ApplicationEvent>>();
-
-	private final Set<Lifecycle> lifecycles = new HashSet<Lifecycle>();
-
-	private volatile boolean running = true;
-
 	private final MultiValueMap<String, String> lazyLifecyleRoles = new LinkedMultiValueMap<String, String>();
 
 	@Override
@@ -141,7 +130,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 			}
 		}
 		catch (NoSuchBeanDefinitionException e) {
-			logger.error("No lifecyle role controller in context");
+			logger.error("No LifecycleRoleController in the context");
 		}
 	}
 
@@ -154,8 +143,9 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 			return bean;
 		}
 		ReflectionUtils.doWithMethods(beanClass, new ReflectionUtils.MethodCallback() {
+
 			@Override
-			@SuppressWarnings({ "unchecked", "rawtypes" })
+			@SuppressWarnings({"unchecked", "rawtypes"})
 			public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
 				Map<Class<? extends Annotation>, List<Annotation>> annotationChains =
 						new HashMap<Class<? extends Annotation>, List<Annotation>>();
@@ -181,7 +171,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 							catch (NoSuchMethodException e) {
 								throw new IllegalArgumentException("Service methods must be extracted to the service "
 										+ "interface for JdkDynamicProxy. The affected bean is: '" + beanName + "' "
-										+ "and its method: '"  + method + "'", e);
+										+ "and its method: '" + method + "'", e);
 							}
 						}
 						Object result = postProcessor.postProcess(bean, beanName, targetMethod, annotations);
@@ -211,20 +201,8 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 							String endpointBeanName = generateBeanName(beanName, method, annotationType);
 							endpoint.setBeanName(endpointBeanName);
 							beanFactory.registerSingleton(endpointBeanName, endpoint);
-							endpoint.setBeanFactory(beanFactory);
-							try {
-								endpoint.afterPropertiesSet();
-							}
-							catch (Exception e) {
-								throw new BeanInitializationException("failed to initialize annotated component", e);
-							}
-							lifecycles.add(endpoint);
-							if (endpoint.isAutoStartup()) {
-								endpoint.start();
-							}
-							if (result instanceof ApplicationListener) {
-								listeners.add((ApplicationListener) result);
-							}
+							beanFactory.initializeBean(endpoint, endpointBeanName);
+
 							Role role = AnnotationUtils.findAnnotation(method, Role.class);
 							if (role != null) {
 								lazyLifecyleRoles.add(role.value(), endpointBeanName);
@@ -239,7 +217,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 	}
 
 	/**
-	 * @param method the method.
+	 * @param method         the method.
 	 * @param annotationType the annotation type.
 	 * @return the hierarchical list of annotations in top-bottom order.
 	 */
@@ -258,7 +236,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 	}
 
 	private boolean recursiveFindAnnotation(Class<? extends Annotation> annotationType, Annotation ann,
-			List<Annotation> annotationChain, Set<Annotation> visited) {
+											List<Annotation> annotationChain, Set<Annotation> visited) {
 		if (ann.annotationType().equals(annotationType)) {
 			annotationChain.add(ann);
 			return true;
@@ -281,8 +259,10 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 		return (targetClass != null) ? targetClass : bean.getClass();
 	}
 
-	private String generateBeanName(String originalBeanName, Method method, Class<? extends Annotation> annotationType) {
-		String baseName = originalBeanName + "." + method.getName() + "." + ClassUtils.getShortNameAsProperty(annotationType);
+	private String generateBeanName(String originalBeanName, Method method,
+									Class<? extends Annotation> annotationType) {
+		String baseName = originalBeanName + "." + method.getName() + "."
+				+ ClassUtils.getShortNameAsProperty(annotationType);
 		String name = baseName;
 		int count = 1;
 		while (this.beanFactory.containsBean(name)) {
@@ -291,47 +271,4 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean
 		return name;
 	}
 
-
-	@Override
-	public void onApplicationEvent(ApplicationEvent event) {
-		for (ApplicationListener<ApplicationEvent> listener : listeners) {
-			try  {
-				listener.onApplicationEvent(event);
-			}
-			catch (ClassCastException e) {
-				if (logger.isWarnEnabled() && event != null) {
-					logger.warn("ApplicationEvent of type [" + event.getClass() +
-							"] not accepted by ApplicationListener [" + listener + "]");
-				}
-			}
-		}
-	}
-
-	// Lifecycle implementation
-
-	@Override
-	public boolean isRunning() {
-		return this.running;
-	}
-
-	@Override
-	public void start() {
-		for (Lifecycle lifecycle : this.lifecycles) {
-			if (!lifecycle.isRunning()) {
-				lifecycle.start();
-			}
-		}
-		this.running = true;
-	}
-
-	@Override
-	public void stop() {
-		for (Lifecycle lifecycle : this.lifecycles) {
-			if (lifecycle.isRunning()) {
-				lifecycle.stop();
-			}
-		}
-		this.running = false;
-	}
-
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java

@@ -29,6 +29,8 @@ import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.MessageHandlingRunnable;
+import org.springframework.util.Assert;
 
 /**
  * A broadcasting dispatcher implementation. If the 'ignoreFailures' property is set to <code>false</code> (the
@@ -62,6 +64,16 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
 
 	private volatile boolean messageBuilderFactorySet;
 
+	private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator =
+			new MessageHandlingTaskDecorator() {
+
+				@Override
+				public Runnable decorate(MessageHandlingRunnable task) {
+					return task;
+				}
+
+			};
+
 	private BeanFactory beanFactory;
 
 
@@ -117,6 +129,11 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
 		this.minSubscribers = minSubscribers;
 	}
 
+	public void setMessageHandlingTaskDecorator(MessageHandlingTaskDecorator messageHandlingTaskDecorator) {
+		Assert.notNull(messageHandlingTaskDecorator, "'messageHandlingTaskDecorator' must not be null.");
+		this.messageHandlingTaskDecorator = messageHandlingTaskDecorator;
+	}
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -141,16 +158,18 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
 			throw new MessageDispatchingException(message, "Dispatcher has no subscribers");
 		}
 		int sequenceSize = handlers.size();
-		for (final MessageHandler handler : handlers) {
-			final Message<?> messageToSend = (!this.applySequence) ? message : getMessageBuilderFactory().fromMessage(message)
-					.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build();
+		for (MessageHandler handler : handlers) {
+			Message<?> messageToSend = message;
+			if (this.applySequence) {
+				messageToSend = getMessageBuilderFactory()
+						.fromMessage(message)
+						.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize)
+						.build();
+			}
+
 			if (this.executor != null) {
-				this.executor.execute(new Runnable() {
-					@Override
-					public void run() {
-						invokeHandler(handler, messageToSend);
-					}
-				});
+				Runnable task = createMessageHandlingTask(handler, messageToSend);
+				this.executor.execute(task);
 				dispatched++;
 			}
 			else {
@@ -170,6 +189,39 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
 		return dispatched >= minSubscribers;
 	}
 
+
+	private Runnable createMessageHandlingTask(final MessageHandler handler, final Message<?> message) {
+		MessageHandlingRunnable task = new MessageHandlingRunnable() {
+
+			final MessageHandler delegate = new MessageHandler() {
+
+				@Override
+				public void handleMessage(Message<?> message) throws MessagingException {
+					invokeHandler(handler, message);
+				}
+
+			};
+
+			@Override
+			public void run() {
+				invokeHandler(handler, message);
+			}
+
+			@Override
+			public Message<?> getMessage() {
+				return message;
+			}
+
+			@Override
+			public MessageHandler getMessageHandler() {
+				return this.delegate;
+			}
+
+		};
+
+		return this.messageHandlingTaskDecorator.decorate(task);
+	}
+
 	private boolean invokeHandler(MessageHandler handler, Message<?> message) {
 		try {
 			handler.handleMessage(message);

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlingTaskDecorator.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ *  Licensed 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.springframework.integration.dispatcher;
+
+import org.springframework.messaging.support.MessageHandlingRunnable;
+
+/**
+ * The strategy to decorate {@link MessageHandlingRunnable} tasks
+ * to be used with the {@link java.util.concurrent.Executor}.
+ *
+ * @author Artem Bilan
+ * @since 4.2
+ * @see UnicastingDispatcher
+ * @see BroadcastingDispatcher
+ */
+public interface MessageHandlingTaskDecorator {
+
+	Runnable decorate(MessageHandlingRunnable task);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java

@@ -1,4 +1,4 @@
-/* Copyright 2002-2014 the original author or authors.
+/* Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,6 +24,9 @@ import org.springframework.integration.MessageDispatchingException;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.MessageHandlingRunnable;
+import org.springframework.util.Assert;
 
 /**
  * Implementation of {@link MessageDispatcher} that will attempt to send a
@@ -43,16 +46,35 @@ import org.springframework.messaging.MessageHandler;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Oleg Zhurakousky
+ * @author Artem Bilan
  * @since 1.0.2
  */
 public class UnicastingDispatcher extends AbstractDispatcher {
 
+	private final MessageHandler dispatchHandler = new MessageHandler() {
+
+		@Override
+		public void handleMessage(Message<?> message) throws MessagingException {
+			doDispatch(message);
+		}
+
+	};
+
+	private final Executor executor;
+
 	private volatile boolean failover = true;
 
 	private volatile LoadBalancingStrategy loadBalancingStrategy;
 
-	private final Executor executor;
+	private volatile MessageHandlingTaskDecorator messageHandlingTaskDecorator =
+			new MessageHandlingTaskDecorator() {
 
+				@Override
+				public Runnable decorate(MessageHandlingRunnable task) {
+					return task;
+				}
+
+			};
 
 	public UnicastingDispatcher() {
 		this.executor = null;
@@ -83,20 +105,44 @@ public class UnicastingDispatcher extends AbstractDispatcher {
 		this.loadBalancingStrategy = loadBalancingStrategy;
 	}
 
+	public void setMessageHandlingTaskDecorator(MessageHandlingTaskDecorator messageHandlingTaskDecorator) {
+		Assert.notNull(messageHandlingTaskDecorator, "'messageHandlingTaskDecorator' must not be null.");
+		this.messageHandlingTaskDecorator = messageHandlingTaskDecorator;
+	}
+
 	@Override
 	public final boolean dispatch(final Message<?> message) {
 		if (this.executor != null) {
-			this.executor.execute(new Runnable() {
-				@Override
-				public void run() {
-					doDispatch(message);
-				}
-			});
+			Runnable task = createMessageHandlingTask(message);
+			this.executor.execute(task);
 			return true;
 		}
 		return this.doDispatch(message);
 	}
 
+	private Runnable createMessageHandlingTask(final Message<?> message) {
+		MessageHandlingRunnable task = new MessageHandlingRunnable() {
+
+			@Override
+			public void run() {
+				doDispatch(message);
+			}
+
+			@Override
+			public Message<?> getMessage() {
+				return message;
+			}
+
+			@Override
+			public MessageHandler getMessageHandler() {
+				return UnicastingDispatcher.this.dispatchHandler;
+			}
+
+		};
+
+		return this.messageHandlingTaskDecorator.decorate(task);
+	}
+
 	private boolean doDispatch(Message<?> message) {
 		if (this.tryOptimizedDispatch(message)) {
 			return true;
@@ -107,7 +153,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
 			throw new MessageDispatchingException(message, "Dispatcher has no subscribers");
 		}
 		List<RuntimeException> exceptions = new ArrayList<RuntimeException>();
-		while (success == false && handlerIterator.hasNext()) {
+		while (!success && handlerIterator.hasNext()) {
 			MessageHandler handler = handlerIterator.next();
 			try {
 				handler.handleMessage(message);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2013 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,12 +16,23 @@
 
 package org.springframework.integration.endpoint;
 
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.Iterator;
+import java.util.List;
+
 import org.springframework.context.Lifecycle;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHandler;
-import org.springframework.messaging.PollableChannel;
+import org.springframework.integration.channel.ExecutorChannelInterceptorAware;
 import org.springframework.integration.transaction.IntegrationResourceHolder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageDeliveryException;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.messaging.support.ChannelInterceptor;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
 import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
 
 /**
  * Message Endpoint that connects any {@link MessageHandler} implementation
@@ -30,6 +41,7 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public class PollingConsumer extends AbstractPollingEndpoint {
 
@@ -37,6 +49,8 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 
 	private final MessageHandler handler;
 
+	private final List<ChannelInterceptor> channelInterceptors;
+
 	private volatile long receiveTimeout = 1000;
 
 	public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) {
@@ -44,6 +58,12 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 		Assert.notNull(handler, "handler must not be null");
 		this.inputChannel = inputChannel;
 		this.handler = handler;
+		if (this.inputChannel instanceof ExecutorChannelInterceptorAware) {
+			this.channelInterceptors = ((ExecutorChannelInterceptorAware) this.inputChannel).getChannelInterceptors();
+		}
+		else {
+			channelInterceptors = null;
+		}
 	}
 
 
@@ -59,7 +79,6 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 		super.doStart();
 	}
 
-
 	@Override
 	protected void doStop() {
 		if (this.handler instanceof Lifecycle) {
@@ -68,18 +87,82 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 		super.doStop();
 	}
 
-
 	@Override
 	protected void handleMessage(Message<?> message) {
-		this.handler.handleMessage(message);
+		Deque<ExecutorChannelInterceptor> interceptorStack = null;
+		try {
+			if (this.channelInterceptors != null
+					&& ((ExecutorChannelInterceptorAware) this.inputChannel).hasExecutorInterceptors()) {
+				interceptorStack = new ArrayDeque<ExecutorChannelInterceptor>();
+				message = applyBeforeHandle(message, interceptorStack);
+				if (message == null) {
+					return;
+				}
+			}
+			this.handler.handleMessage(message);
+			if (!CollectionUtils.isEmpty(interceptorStack)) {
+				triggerAfterMessageHandled(message, null, interceptorStack);
+			}
+		}
+		catch (Exception ex) {
+			if (!CollectionUtils.isEmpty(interceptorStack)) {
+				triggerAfterMessageHandled(message, ex, interceptorStack);
+			}
+			if (ex instanceof MessagingException) {
+				throw (MessagingException) ex;
+			}
+			String description = "Failed to handle " + message + " to " + this + " in " + this.handler;
+			throw new MessageDeliveryException(message, description, ex);
+		}
+		catch (Error ex) {//NOSONAR - ok, we re-throw below
+			if (!CollectionUtils.isEmpty(interceptorStack)) {
+				String description = "Failed to handle " + message + " to " + this + " in " + this.handler;
+				triggerAfterMessageHandled(message,
+						new MessageDeliveryException(message, description, ex),
+						interceptorStack);
+			}
+			throw ex;
+		}
+	}
+
+	private Message<?> applyBeforeHandle(Message<?> message, Deque<ExecutorChannelInterceptor> interceptorStack) {
+		for (ChannelInterceptor interceptor : this.channelInterceptors) {
+			if (interceptor instanceof ExecutorChannelInterceptor) {
+				ExecutorChannelInterceptor executorInterceptor = (ExecutorChannelInterceptor) interceptor;
+				message = executorInterceptor.beforeHandle(message, this.inputChannel, this.handler);
+				if (message == null) {
+					if (logger.isDebugEnabled()) {
+						logger.debug(executorInterceptor.getClass().getSimpleName()
+								+ " returned null from beforeHandle, i.e. precluding the send.");
+					}
+					triggerAfterMessageHandled(null, null, interceptorStack);
+					return null;
+				}
+				interceptorStack.add(executorInterceptor);
+			}
+		}
+		return message;
+	}
+
+	private void triggerAfterMessageHandled(Message<?> message, Exception ex,
+	                                        Deque<ExecutorChannelInterceptor> interceptorStack) {
+		Iterator<ExecutorChannelInterceptor> iterator = interceptorStack.descendingIterator();
+		while (iterator.hasNext()) {
+			ExecutorChannelInterceptor interceptor = iterator.next();
+			try {
+				interceptor.afterMessageHandled(message, this.inputChannel, this.handler, ex);
+			}
+			catch (Throwable ex2) {//NOSONAR
+				logger.error("Exception from afterMessageHandled in " + interceptor, ex2);
+			}
+		}
 	}
 
 	@Override
 	protected Message<?> receiveMessage() {
-		Message<?> message = (this.receiveTimeout >= 0)
+		return (this.receiveTimeout >= 0)
 				? this.inputChannel.receive(this.receiveTimeout)
 				: this.inputChannel.receive();
-		return message;
 	}
 
 	@Override
@@ -91,4 +174,5 @@ public class PollingConsumer extends AbstractPollingEndpoint {
 	protected String getResourceKey() {
 		return IntegrationResourceHolder.INPUT_CHANNEL;
 	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/ExecutorChannelTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,25 +20,43 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.BDDMockito.willThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.Before;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.core.task.SyncTaskExecutor;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
+import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.scheduling.concurrent.ConcurrentTaskExecutor;
 import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 
 /**
  * @author Mark Fisher
+ * @author Artem Bilan
  */
 public class ExecutorChannelTests {
 
@@ -156,6 +174,50 @@ public class ExecutorChannelTests {
 		assertEquals(numberOfMessages, handler2.count.get());
 	}
 
+	@Test
+	public void interceptorWithModifiedMessage() {
+		ExecutorChannel channel = new ExecutorChannel(new SyncTaskExecutor());
+		channel.setBeanFactory(mock(BeanFactory.class));
+		channel.afterPropertiesSet();
+
+		MessageHandler handler = mock(MessageHandler.class);
+		Message<?> expected = mock(Message.class);
+		BeforeHandleInterceptor interceptor = new BeforeHandleInterceptor();
+		interceptor.setMessageToReturn(expected);
+		channel.addInterceptor(interceptor);
+		channel.subscribe(handler);
+		channel.send(new GenericMessage<Object>("foo"));
+		verify(handler).handleMessage(expected);
+		assertEquals(1, interceptor.getCounter().get());
+		assertTrue(interceptor.wasAfterHandledInvoked());
+	}
+
+	@Test
+	public void interceptorWithException() {
+		ExecutorChannel channel = new ExecutorChannel(new SyncTaskExecutor());
+		channel.setBeanFactory(mock(BeanFactory.class));
+		channel.afterPropertiesSet();
+
+		Message<Object> message = new GenericMessage<Object>("foo");
+
+		MessageHandler handler = mock(MessageHandler.class);
+		IllegalStateException expected = new IllegalStateException("Fake exception");
+		willThrow(expected).given(handler).handleMessage(message);
+		BeforeHandleInterceptor interceptor = new BeforeHandleInterceptor();
+		channel.addInterceptor(interceptor);
+		channel.subscribe(handler);
+		try {
+			channel.send(message);
+		}
+		catch (MessageDeliveryException actual) {
+			assertSame(expected, actual.getCause());
+		}
+		verify(handler).handleMessage(message);
+		assertEquals(1, interceptor.getCounter().get());
+		assertTrue(interceptor.wasAfterHandledInvoked());
+	}
+
+
 
 	private static class TestHandler implements MessageHandler {
 
@@ -181,4 +243,40 @@ public class ExecutorChannelTests {
 		}
 	}
 
+	private static class BeforeHandleInterceptor extends ChannelInterceptorAdapter
+			implements ExecutorChannelInterceptor {
+
+		private AtomicInteger counter = new AtomicInteger();
+
+		private volatile boolean afterHandledInvoked;
+
+		private Message<?> messageToReturn;
+
+		public void setMessageToReturn(Message<?> messageToReturn) {
+			this.messageToReturn = messageToReturn;
+		}
+
+		public AtomicInteger getCounter() {
+			return this.counter;
+		}
+
+		public boolean wasAfterHandledInvoked() {
+			return this.afterHandledInvoked;
+		}
+
+		@Override
+		public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
+			assertNotNull(message);
+			this.counter.incrementAndGet();
+			return (this.messageToReturn != null ? this.messageToReturn : message);
+		}
+
+		@Override
+		public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler,
+										Exception ex) {
+			this.afterHandledInvoked = true;
+		}
+
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,7 +24,10 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
+import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -36,11 +39,16 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.endpoint.PollingConsumer;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ChannelInterceptorAdapter;
+import org.springframework.messaging.support.ExecutorChannelInterceptor;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.util.StringUtils;
 
@@ -253,6 +261,45 @@ public class ChannelInterceptorTests {
 		ac.close();
 	}
 
+	@Test
+	public void testPollingConsumerWithExecutorInterceptor() throws InterruptedException {
+		TestUtils.TestApplicationContext testApplicationContext = TestUtils.createTestApplicationContext();
+
+		QueueChannel channel = new QueueChannel();
+
+		final CountDownLatch latch1 = new CountDownLatch(1);
+		final CountDownLatch latch2 = new CountDownLatch(2);
+		final List<Message<?>> messages = new ArrayList<>();
+
+		PollingConsumer consumer = new PollingConsumer(channel, new MessageHandler() {
+
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				messages.add(message);
+				latch1.countDown();
+				latch2.countDown();
+			}
+
+		});
+
+		testApplicationContext.registerBean("consumer", consumer);
+		testApplicationContext.refresh();
+
+		channel.send(new GenericMessage<>("foo"));
+
+		assertTrue(latch1.await(10, TimeUnit.SECONDS));
+
+		channel.addInterceptor(new TestExecutorInterceptor());
+		channel.send(new GenericMessage<>("foo"));
+
+		assertTrue(latch2.await(10, TimeUnit.SECONDS));
+		assertEquals(2, messages.size());
+
+		assertEquals("foo", messages.get(0).getPayload());
+		assertEquals("FOO", messages.get(1).getPayload());
+
+		testApplicationContext.close();
+	}
 
 	public static class PreSendReturnsMessageInterceptor extends ChannelInterceptorAdapter {
 		private String foo;
@@ -374,6 +421,7 @@ public class ChannelInterceptorTests {
 		public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
 			this.afterCompletionInvoked = true;
 		}
+
 	}
 
 
@@ -386,6 +434,26 @@ public class ChannelInterceptorTests {
 			counter.incrementAndGet();
 			return false;
 		}
+
 	}
 
+	private static class TestExecutorInterceptor extends ChannelInterceptorAdapter
+			implements ExecutorChannelInterceptor {
+
+		@Override
+		public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
+			return MessageBuilder.withPayload(((String) message.getPayload()).toUpperCase())
+					.copyHeaders(message.getHeaders())
+					.build();
+		}
+
+		@Override
+		public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler,
+										Exception ex) {
+
+		}
+
+	}
+
+
 }

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests-context.xml

@@ -11,6 +11,8 @@
 
 	<annotation-config/>	<!-- Second declaration should not be a problem - see INT-3445 -->
 
+	<beans:bean class="org.springframework.integration.config.annotation.AnnotatedEndpointActivationTests.AnnotatedEndpoint"/>
+
 	<channel id="input"/>
 
 	<channel id="output">

spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AnnotatedEndpointActivationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.GenericMessage;
+import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
@@ -41,10 +42,11 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
  * @author Dave Syer
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
-@MessageEndpoint
+@DirtiesContext
 public class AnnotatedEndpointActivationTests {
 
 	@Autowired
@@ -63,21 +65,6 @@ public class AnnotatedEndpointActivationTests {
 	// them will get the message.
 	private static volatile int count = 0;
 
-
-	@ServiceActivator(inputChannel = "input", outputChannel = "output")
-	public String process(String message) {
-		count++;
-		String result = message + ": " + count;
-		return result;
-	}
-
-	@ServiceActivator(inputChannel = "inputImplicit", outputChannel = "output")
-	public String processImplicit(String message) {
-		count++;
-		String result = message + ": " + count;
-		return result;
-	}
-
 	@Before
 	public void resetCount() {
 		count = 0;
@@ -108,12 +95,14 @@ public class AnnotatedEndpointActivationTests {
 	}
 
 	@Test(expected = MessageDeliveryException.class)
+	@DirtiesContext
 	public void stopContext() {
 		applicationContext.stop();
 		this.input.send(new GenericMessage<String>("foo"));
 	}
 
 	@Test
+	@DirtiesContext
 	public void stopAndRestartContext() {
 		applicationContext.stop();
 		applicationContext.start();
@@ -124,4 +113,21 @@ public class AnnotatedEndpointActivationTests {
 		assertEquals(1, count);
 	}
 
+	@MessageEndpoint
+	private static class AnnotatedEndpoint {
+
+		@ServiceActivator(inputChannel = "input", outputChannel = "output")
+		public String process(String message) {
+			count++;
+			return message + ": " + count;
+		}
+
+		@ServiceActivator(inputChannel = "inputImplicit", outputChannel = "output")
+		public String processImplicit(String message) {
+			count++;
+			return message + ": " + count;
+		}
+
+	}
+
 }