Added new message bus implementation assuming all subscription and consumer activation responsibilities.

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

@@ -1,159 +0,0 @@
-/*
- * Copyright 2002-2007 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;
-
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.beans.BeansException;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.ApplicationContextAware;
-import org.springframework.context.Lifecycle;
-import org.springframework.integration.channel.ChannelResolver;
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.channel.consumer.AbstractConsumer;
-import org.springframework.integration.channel.consumer.ConsumerType;
-import org.springframework.integration.channel.consumer.EventDrivenConsumer;
-import org.springframework.integration.channel.consumer.FixedDelayConsumer;
-import org.springframework.integration.channel.consumer.FixedRateConsumer;
-import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.util.Assert;
-
-/**
- * A central component for registering channels and endpoints. The message bus
- * will autodetect channels and endpoints from its host application context.
- * 
- * @author Mark Fisher
- */
-public class MessageBus implements ChannelResolver, ApplicationContextAware, Lifecycle {
-
-	private final Log logger = LogFactory.getLog(getClass());
-
-	private Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
-
-	private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
-
-	private List<AbstractConsumer> consumers = new CopyOnWriteArrayList<AbstractConsumer>();
-
-	private ApplicationContext applicationContext;
-
-	private boolean running;
-
-	private Object lifecycleMonitor = new Object();
-
-
-	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-		Assert.notNull(applicationContext, "applicationContext must not be null");
-		this.applicationContext = applicationContext;
-		this.initChannels();
-		this.initEndpoints();
-	}
-
-	@SuppressWarnings("unchecked")
-	private void initChannels() {
-		Map<String, MessageChannel> channelBeans = (Map<String, MessageChannel>) this.applicationContext
-				.getBeansOfType(MessageChannel.class);
-		for (Map.Entry<String, MessageChannel> entry : channelBeans.entrySet()) {
-			this.channels.put(entry.getKey(), entry.getValue());
-			if (logger.isInfoEnabled()) {
-				logger.info("registered channel '" + entry.getKey() + "'");
-			}
-		}
-	}
-
-	@SuppressWarnings("unchecked")
-	private void initEndpoints() {
-		Map<String, MessageEndpoint> endpointBeans = (Map<String, MessageEndpoint>) this.applicationContext
-				.getBeansOfType(MessageEndpoint.class);
-		for (Map.Entry<String, MessageEndpoint> entry : endpointBeans.entrySet()) {
-			this.endpoints.put(entry.getKey(), entry.getValue());
-			if (logger.isInfoEnabled()) {
-				logger.info("registered endpoint '" + entry.getKey() + "'");
-			}
-		}
-	}
-
-	public MessageChannel resolve(String channelName) {
-		return this.channels.get(channelName);
-	}
-
-	public boolean isRunning() {
-		synchronized (this.lifecycleMonitor) {
-			return this.running;
-		}
-	}
-
-	public void start() {
-		synchronized (this.lifecycleMonitor) {
-			if (!this.isRunning()) {
-				this.running = true;
-				this.activateEndpoints();
-			}
-		}
-	}
-
-	public void stop() {
-		synchronized (this.lifecycleMonitor) {
-			if (this.isRunning()) {
-				this.running = false;
-				this.deactivateEndpoints();
-			}
-		}
-	}
-
-	private void activateEndpoints() {
-		for (MessageEndpoint endpoint : this.endpoints.values()) {
-			MessageSource source = endpoint.getSource();
-			ConsumerType consumerType = endpoint.getConsumerType();
-			AbstractConsumer consumer = createConsumer(consumerType, source, endpoint);
-			consumer.initialize();
-			consumer.start();
-		}
-	}
-
-	private void deactivateEndpoints() {
-		for (AbstractConsumer consumer : this.consumers) {
-			consumer.stop();
-		}
-	}
-
-
-	/**
-	 * Create a consumer based upon the specified consumer type.
-	 */
-	private AbstractConsumer createConsumer(ConsumerType type, MessageSource source, MessageEndpoint endpoint) {
-		if (type.equals(ConsumerType.EVENT_DRIVEN)) {
-			return new EventDrivenConsumer(source, endpoint);
-		}
-		else if (type.equals(ConsumerType.FIXED_RATE)) {
-			return new FixedRateConsumer(source, endpoint);
-		}
-		else if (type.equals(ConsumerType.FIXED_DELAY)) {
-			return new FixedDelayConsumer(source, endpoint);
-		}
-		else {
-			throw new UnsupportedOperationException("the consumerType '"
-					+ type.name() + "' is not supported.");
-		}
-	}
-
-}

spring-eai-core/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java

@@ -23,7 +23,6 @@ import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
 
-import org.springframework.integration.channel.consumer.ConsumerType;
 import org.springframework.stereotype.Component;
 
 /**
@@ -42,6 +41,4 @@ public @interface MessageEndpoint {
 
 	String target();
 
-	ConsumerType consumerType() default ConsumerType.EVENT_DRIVEN;
-
 }

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

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2002-2007 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.bus;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A container for Message consumer configuration metadata.
+ * 
+ * @author Mark Fisher
+ */
+public class ConsumerPolicy {
+
+	private static final int DEFAULT_CONCURRENCY = 1;
+
+	private static final int DEFAULT_MAX_CONCURRENCY = 10;
+
+	private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10;
+
+	private static final int DEFAULT_REJECTION_LIMIT = 10;
+
+	private static final int DEFAULT_REJECTION_LIMIT_WAIT = 1000;
+
+	private static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
+
+
+	private int concurrency = DEFAULT_CONCURRENCY;
+
+	private int maxConcurrency = DEFAULT_MAX_CONCURRENCY;
+
+	private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
+
+	private int rejectionLimit = DEFAULT_REJECTION_LIMIT;
+
+	private int rejectionLimitWait = DEFAULT_REJECTION_LIMIT_WAIT;
+
+	private int initialDelay = 0;
+
+	private int period = -1;
+
+	private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
+
+	private boolean fixedRate = false;
+
+	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
+
+
+	public int getInitialDelay() {
+		return this.initialDelay;
+	}
+
+	public void setInitialDelay(int initialDelay) {
+		this.initialDelay = initialDelay;
+	}
+
+	public int getPeriod() {
+		return this.period;
+	}
+
+	public void setPeriod(int period) {
+		this.period = period;
+	}
+
+	public TimeUnit getTimeUnit() {
+		return this.timeUnit;
+	}
+
+	public void setTimeUnit(TimeUnit timeUnit) {
+		this.timeUnit = timeUnit;
+	}
+
+	public boolean isFixedRate() {
+		return this.fixedRate;
+	}
+
+	public void setFixedRate(boolean fixedRate) {
+		this.fixedRate = fixedRate;
+	}
+
+	public int getConcurrency() {
+		return this.concurrency;
+	}
+
+	public void setConcurrency(int concurrency) {
+		if (concurrency < 1) {
+			throw new IllegalArgumentException("'concurrency' value must be at least 1");
+		}
+		this.concurrency = concurrency;
+	}
+
+	public int getMaxConcurrency() {
+		return this.maxConcurrency;
+	}
+
+	public void setMaxConcurrency(int maxConcurrency) {
+		if (maxConcurrency < 1) {
+			throw new IllegalArgumentException("'maxConcurrency' value must be at least 1");
+		}
+		this.maxConcurrency = maxConcurrency;
+	}
+
+	public int getMaxMessagesPerTask() {
+		return this.maxMessagesPerTask;
+	}
+
+	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
+		if (maxMessagesPerTask == 0) {
+			throw new IllegalArgumentException("'maxMessagesPerTask' must not be 0");
+		}
+		this.maxMessagesPerTask = maxMessagesPerTask;
+	}
+
+	public int getRejectionLimit() {
+		return this.rejectionLimit;
+	}
+
+	public void setRejectionLimit(int rejectionLimit) {
+		if (rejectionLimit < 1) {
+			throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1");
+		}
+		this.rejectionLimit = rejectionLimit;
+	}
+
+	public int getRejectionLimitWait() {
+		return this.rejectionLimitWait;
+	}
+
+	public void setRejectionLimitWait(int rejectionLimitWait) {
+		this.rejectionLimitWait = rejectionLimitWait;
+	}
+
+	public long getReceiveTimeout() {
+		return this.receiveTimeout;
+	}
+
+	public void setReceiveTimeout(long receiveTimeout) {
+		this.receiveTimeout = receiveTimeout;
+	}
+
+}

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

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2002-2007 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.bus;
+
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.bus.MessageBus.EndpointTask;
+
+/**
+ * A subclass of {@link ThreadPoolExecutor} with configurable error thresholds.
+ * 
+ * @author Mark Fisher
+ */
+public class EndpointExecutor extends ThreadPoolExecutor {
+
+	private Log logger = LogFactory.getLog(this.getClass());
+
+	private int successiveErrorCount;
+
+	private int successiveErrorThreshold = -1;
+
+	private int totalErrorCount;
+
+	private int totalErrorThreshold = -1;
+
+
+	public EndpointExecutor(int corePoolSize, int maximumPoolSize) {
+		super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<Runnable>());
+	}
+
+	/**
+	 * Set the maximum number of errors allowed in <em>successive</em>
+	 * endpoint executions. If this threshold is ever exceeded, the executor
+	 * will shutdown.
+	 */
+	public void setSuccessiveErrorThreshold(int successiveErrorThreshold) {
+		this.successiveErrorThreshold = successiveErrorThreshold;
+	}
+
+	/**
+	 * Set the maximum number of <em>total</em> errors allowed in endpoint
+	 * executions. If this threshold is ever exceeded, the executor will
+	 * shutdown.
+	 */
+	public void setTotalErrorThreshold(int totalErrorThreshold) {
+		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());
+			}
+			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();
+			}
+		}
+		else {
+			successiveErrorCount = 0;
+		}
+	}
+
+}

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

@@ -0,0 +1,315 @@
+/*
+ * Copyright 2002-2007 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.bus;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.MessagingException;
+import org.springframework.integration.channel.ChannelResolver;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.channel.PointToPointChannel;
+import org.springframework.integration.endpoint.MessageEndpoint;
+import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
+
+/**
+ * The messaging bus. Serves as a registry for channels and endpoints, manages their lifecycle,
+ * and all subscriptions.
+ * 
+ * @author Mark Fisher
+ */
+public class MessageBus implements ChannelResolver, ApplicationContextAware, Lifecycle {
+
+	private Log logger = LogFactory.getLog(this.getClass());
+
+	private Map<String, MessageChannel> channels = new ConcurrentHashMap<String, MessageChannel>();
+
+	private Map<String, MessageEndpoint> endpoints = new ConcurrentHashMap<String, MessageEndpoint>();
+
+	private List<DispatcherTask> dispatcherTasks = new CopyOnWriteArrayList<DispatcherTask>();
+
+	private Map<MessageEndpoint, EndpointExecutor> endpointExecutors = new ConcurrentHashMap<MessageEndpoint, EndpointExecutor>();
+
+	private ScheduledThreadPoolExecutor dispatcherExecutor;
+
+	private boolean autoCreateChannels;
+
+	private ApplicationContext applicationContext;
+
+	private boolean running;
+
+	private Object lifecycleMonitor = new Object();
+
+
+	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+		Assert.notNull(applicationContext, "applicationContext must not be null");
+		this.applicationContext = applicationContext;
+		this.registerChannelsFromContext();
+		this.registerEndpointsFromContext();
+	}
+
+	@SuppressWarnings("unchecked")
+	private void registerChannelsFromContext() {
+		Map<String, MessageChannel> channelBeans = (Map<String, MessageChannel>) this.applicationContext
+				.getBeansOfType(MessageChannel.class);
+		for (Map.Entry<String, MessageChannel> entry : channelBeans.entrySet()) {
+			this.registerChannel(entry.getKey(), entry.getValue());
+			if (logger.isInfoEnabled()) {
+				logger.info("registered channel '" + entry.getKey() + "'");
+			}
+		}
+	}
+
+	@SuppressWarnings("unchecked")
+	private void registerEndpointsFromContext() {
+		Map<String, MessageEndpoint> endpointBeans = (Map<String, MessageEndpoint>) this.applicationContext
+				.getBeansOfType(MessageEndpoint.class);
+		for (Map.Entry<String, MessageEndpoint> entry : endpointBeans.entrySet()) {
+			this.registerEndpoint(entry.getKey(), entry.getValue());
+			if (logger.isInfoEnabled()) {
+				logger.info("registered endpoint '" + entry.getKey() + "'");
+			}
+		}
+	}
+
+	public void initialize() {
+		this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherTasks.size() > 0 ? this.dispatcherTasks.size() : 1);
+	}
+
+	public MessageChannel resolve(String channelName) {
+		return this.channels.get(channelName);
+	}
+
+	public void registerChannel(String name, MessageChannel channel) {
+		this.channels.put(name, channel);
+	}
+
+	public void registerEndpoint(String name, MessageEndpoint endpoint) {
+		this.endpoints.put(name, endpoint);
+	}
+
+	public void activateSubscription(String channelName, String endpointName, ConsumerPolicy policy) {
+		MessageChannel channel = this.channels.get(channelName);
+		if (channel == null) {
+			if (this.autoCreateChannels == false) {
+				throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName +
+						"'. Consider enabling the 'autoCreateChannels' option for the message bus.");
+			}
+			this.registerChannel(channelName, new PointToPointChannel());
+			if (this.logger.isInfoEnabled()) {
+				logger.info("created channel '" + channelName + "'");
+			}
+		}
+		MessageEndpoint endpoint = this.endpoints.get(endpointName);
+		if (endpoint == null) {
+			throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'");
+		}
+		EndpointExecutor endpointExecutor = new EndpointExecutor(policy.getConcurrency(), policy.getMaxConcurrency());
+		endpointExecutors.put(endpoint, endpointExecutor);
+		DispatcherTask dispatcherTask = new DispatcherTask(channel, endpoint, policy);
+		this.dispatcherTasks.add(dispatcherTask);
+		if (this.isRunning()) {
+			scheduleDispatcherTask(dispatcherTask);
+		}
+	}
+
+	public int getActiveCountForEndpoint(String endpointName) {
+		MessageEndpoint endpoint = this.endpoints.get(endpointName);
+		if (endpoint != null) {
+			EndpointExecutor executor = this.endpointExecutors.get(endpoint);
+			if (executor != null) {
+				return executor.getActiveCount();
+			}
+		}
+		return 0;
+	}
+
+	private void scheduleDispatcherTask(DispatcherTask task) {
+		ConsumerPolicy policy = task.getPolicy();
+		if (policy.getPeriod() <= 0) {
+			if (policy.getReceiveTimeout() <= 0) {
+				if (logger.isWarnEnabled()) {
+					logger.warn("Scheduling a repeating task with no receive timeout is not recommended! " +
+							"Consider providing a positive value for either 'period' or 'receiveTimeout'");
+				}
+			}
+			dispatcherExecutor.schedule(new RepeatingDispatcherTask(task), policy.getInitialDelay(), policy.getTimeUnit());
+		}
+		else if (policy.isFixedRate()) {
+			dispatcherExecutor.scheduleAtFixedRate(task, policy.getInitialDelay(), policy.getPeriod(), policy.getTimeUnit());
+		}
+		else {
+			dispatcherExecutor.scheduleWithFixedDelay(task, policy.getInitialDelay(), policy.getPeriod(), policy.getTimeUnit());
+		}
+	}
+
+	public boolean isRunning() {
+		synchronized (this.lifecycleMonitor) {
+			return this.running;
+		}
+	}
+
+	public void start() {
+		if (this.dispatcherExecutor == null) {
+			this.initialize();
+		}
+		synchronized (this.lifecycleMonitor) {
+			if (!this.isRunning()) {
+				this.running = true;
+				for (DispatcherTask task : this.dispatcherTasks) {
+					scheduleDispatcherTask(task);
+				}
+			}
+		}
+	}
+
+	public void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (this.isRunning()) {
+				this.running = false;
+				this.dispatcherExecutor.shutdownNow();
+			}
+		}
+	}
+
+
+	private class DispatcherTask implements Runnable {
+
+		private MessageChannel channel;
+
+		private MessageEndpoint endpoint;
+
+		private ConsumerPolicy policy;
+
+
+		public DispatcherTask(MessageChannel channel, MessageEndpoint endpoint, ConsumerPolicy policy) {
+			this.channel = channel;
+			this.endpoint = endpoint;
+			this.policy = policy;
+		}
+
+		public MessageChannel getChannel() {
+			return this.channel;
+		}
+
+		public MessageEndpoint getEndpoint() {
+			return this.endpoint;
+		}
+
+		public ConsumerPolicy getPolicy() {
+			return this.policy;
+		}
+
+		public void run() {
+			EndpointExecutor executor = endpointExecutors.get(this.endpoint);
+			if (executor == null || executor.isShutdown()) {
+				if (logger.isWarnEnabled()) {
+					logger.warn("dispatcher shutting down, endpoint executor is not active");
+				}
+				return;
+			}			
+			for (int i = 0; i < policy.getMaxMessagesPerTask(); i++) {
+				Message message = channel.receive(this.policy.getReceiveTimeout());
+				if (message == null) {
+					return;
+				}
+				else {
+					boolean taskSubmitted = false;
+					int attempts = 0;
+					while (!taskSubmitted) {
+						try {
+							executor.execute(new EndpointTask(this.endpoint, message));
+							taskSubmitted = true;
+						}
+						catch (RejectedExecutionException rex) {
+							attempts++;
+							if (attempts == policy.getRejectionLimit()) {
+								attempts = 0;
+								if (logger.isDebugEnabled()) {
+									logger.debug("reached rejected execution limit");
+								}
+								try {
+									Thread.sleep(policy.getRejectionLimitWait());
+								}
+								catch (InterruptedException iex) {
+									Thread.currentThread().interrupt();
+								}
+							}
+						}
+					}
+				}
+			}
+		}
+	}
+
+
+	private class RepeatingDispatcherTask implements Runnable {
+
+		private DispatcherTask task;
+
+		RepeatingDispatcherTask(DispatcherTask task) {
+			this.task = task;
+		}
+
+		public void run() {
+			task.run();
+			dispatcherExecutor.execute(new RepeatingDispatcherTask(task));
+		}
+	}
+
+
+	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;
+			}
+		}
+	}
+
+}

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

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel.consumer;
+package org.springframework.integration.channel;
 
 /**
  * Enumeration of the different types of message consumer.

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java

@@ -1,146 +0,0 @@
-/*
- * Copyright 2002-2007 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.consumer;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.springframework.context.Lifecycle;
-import org.springframework.integration.MessageSource;
-import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.Message;
-import org.springframework.util.Assert;
-
-/**
- * Base class for consumers defining common properties and behavior.
- * 
- * @author Mark Fisher
- */
-public abstract class AbstractConsumer implements Lifecycle {
-
-	/**
-	 * The default receive timeout: 1000 ms = 1 second.
-	 */
-	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
-
-
-	protected final Log logger = LogFactory.getLog(getClass());
-
-	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
-
-	private MessageSource source;
-
-	private MessageEndpoint endpoint;
-
-	private boolean active = false;
-
-	private boolean running = false;
-
-	protected final Object lifecycleMonitor = new Object();
-
-
-	public AbstractConsumer(MessageSource source, MessageEndpoint endpoint) {
-		Assert.notNull(source, "source must not be null");
-		Assert.notNull(endpoint, "endpoint must not be null");
-		this.source = source;
-		this.endpoint = endpoint;
-	}
-
-
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
-	public final boolean isRunning() {
-		synchronized (this.lifecycleMonitor) {
-			return this.running;
-		}
-	}
-
-	public final void start() {
-		synchronized (this.lifecycleMonitor) {
-			this.running = true;
-			this.lifecycleMonitor.notifyAll();
-		}
-		this.doStart();
-	}
-
-	public final void stop() {
-		synchronized (this.lifecycleMonitor) {
-			this.running = false;
-			this.lifecycleMonitor.notifyAll();
-		}
-		this.doStop();
-	}
-
-	public final boolean isActive() {
-		synchronized (this.lifecycleMonitor) {
-			return this.active;
-		}
-	}
-
-	public final void initialize() {
-		synchronized (this.lifecycleMonitor) {
-			this.active = true;
-			this.lifecycleMonitor.notifyAll();
-		}
-		doInitialize();
-	}
-
-	/**
-	 * If the consumer is active but not yet running, then wait until it is running.
-	 */
-	protected void waitWhileNotRunning() {
-		synchronized (this.lifecycleMonitor) {
-			while (this.active && !this.running) {
-				try {
-					this.lifecycleMonitor.wait();
-				}
-				catch (InterruptedException ex) {
-					Thread.currentThread().interrupt();
-				}
-			}
-		}
-	}
-
-	protected boolean receiveAndPassToEndpoint() {
-		boolean messageReceived = false;
-		Message message = null;
-		if (this.receiveTimeout < 0) { // indefinite timeout
-			message = this.source.receive();
-		}
-		else {
-			message = this.source.receive(this.receiveTimeout);
-		}
-		if (message != null) {
-			messageReceived = true;
-			messageReceived(message);
-			this.endpoint.messageReceived(message);
-		}
-		return messageReceived;
-	}
-
-
-	protected abstract void doInitialize();
-
-	protected abstract void doStart();
-
-	protected abstract void doStop();
-
-	protected abstract void messageReceived(Message message);
-
-}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java

@@ -1,101 +0,0 @@
-/*
- * Copyright 2002-2007 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.consumer;
-
-import java.util.concurrent.ScheduledThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-import org.springframework.integration.MessageSource;
-import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.Message;
-
-/**
- * Base class for consumers that poll on a given interval.
- * 
- * @author Mark Fisher
- */
-public abstract class AbstractPollingConsumer extends AbstractConsumer {
-
-	private static final int DEFAULT_POLL_INTERVAL = 1000;
-
-
-	protected ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
-
-	private int initialDelay = 0;
-
-	private int pollInterval = DEFAULT_POLL_INTERVAL;
-
-	private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
-
-
-	public AbstractPollingConsumer(MessageSource source, MessageEndpoint endpoint) {
-		super(source, endpoint);
-		this.setReceiveTimeout(0);
-	}
-
-
-	public void setPollInterval(int pollInterval) {
-		this.pollInterval = pollInterval;
-	}
-
-	/**
-	 * Specify the {@link TimeUnit} for polling. Default is milliseconds.
-	 */
-	public void setTimeUnit(TimeUnit timeUnit) {
-		this.timeUnit = timeUnit;
-	}
-
-	@Override
-	protected void doStart() {
-		scheduleInvoker(new PollingInvoker(), this.initialDelay, this.pollInterval, this.timeUnit);
-	}
-
-	/**
-	 * Each subclass must implement this method depending on its scheduling
-	 * behavior (e.g. fixed-rate versus fixed-delay).
-	 * 
-	 * @param invoker the invoker task to schedule
-	 * @param initialDelay the time in milliseconds to wait before the first
-	 * poll
-	 * @param pollInterval the polling interval in milliseconds
-	 */
-	protected abstract void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit);
-
-
-	@Override
-	protected void doInitialize() {
-	}
-
-	@Override
-	protected void doStop() {
-		this.executor.shutdown();
-	}
-
-	@Override
-	protected void messageReceived(Message message) {
-	}
-
-
-	private class PollingInvoker implements Runnable {
-
-		public void run() {
-			receiveAndPassToEndpoint();
-		}
-
-	}
-
-}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java

@@ -1,401 +0,0 @@
-/*
- * Copyright 2002-2007 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.consumer;
-
-import java.lang.reflect.Method;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-
-import org.springframework.context.Lifecycle;
-import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.MessageSource;
-import org.springframework.integration.endpoint.MessageEndpoint;
-import org.springframework.integration.message.Message;
-import org.springframework.scheduling.SchedulingAwareRunnable;
-import org.springframework.scheduling.SchedulingTaskExecutor;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.util.Assert;
-
-/**
- * A consumer that runs tasks repeatedly in order to pass to the endpoint as
- * soon as a message is received by one of those tasks.
- * 
- * @author Mark Fisher
- * @author Juergen Hoeller
- */
-public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle {
-
-	private static final int DEFAULT_CONCURRENCY = 1;
-
-	private static final int DEFAULT_MAX_CONCURRENCY = 10;
-
-	private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10;
-
-	private static final int DEFAULT_IDLE_TASK_EXECUTION_LIMIT = 1;
-
-
-	private TaskExecutor executor;
-
-	private int concurrency = DEFAULT_CONCURRENCY;
-
-	private int maxConcurrency = DEFAULT_MAX_CONCURRENCY;
-
-	private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK;
-
-	private int idleTaskExecutionLimit = DEFAULT_IDLE_TASK_EXECUTION_LIMIT;
-
-	private final Set<MessageEndpointInvoker> scheduledInvokers = new HashSet<MessageEndpointInvoker>();
-
-	private int activeInvokerCount = 0;
-
-	private final Object activeInvokerMonitor = new Object();
-
-	private final List<MessageEndpointInvoker> pausedInvokers = new LinkedList<MessageEndpointInvoker>();
-
-
-	public EventDrivenConsumer(MessageSource source, MessageEndpoint endpoint) {
-		super(source, endpoint);
-	}
-
-
-	public void setExecutor(TaskExecutor executor) {
-		Assert.notNull(executor, "executor must not be null");
-		this.executor = executor;
-	}
-
-	public void setConcurrency(int concurrency) {
-		if (concurrency < 1) {
-			throw new IllegalArgumentException("'concurrency' value must be at least 1");
-		}
-		synchronized (this.activeInvokerMonitor) {
-			this.concurrency = concurrency;
-			if (this.maxConcurrency < concurrency) {
-				this.maxConcurrency = concurrency;
-			}
-		}
-	}
-
-	public void setMaxConcurrency(int maxConcurrency) {
-		if (maxConcurrency < 1) {
-			throw new IllegalArgumentException("'maxConcurrency' value must be at least 1");
-		}
-		synchronized (this.activeInvokerMonitor) {
-			this.maxConcurrency = Math.max(maxConcurrency, this.concurrency);
-		}
-	}
-
-	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
-		if (maxMessagesPerTask == 0) {
-			throw new IllegalArgumentException("'maxMessagesPerTask' must not be 0");
-		}
-		synchronized (this.activeInvokerMonitor) {
-			this.maxMessagesPerTask = maxMessagesPerTask;
-		}
-	}
-
-	public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) {
-		if (idleTaskExecutionLimit < 1) {
-			throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1");
-		}
-		synchronized (this.activeInvokerMonitor) {
-			this.idleTaskExecutionLimit = idleTaskExecutionLimit;
-		}
-	}
-
-	public void doStart() {
-		synchronized (this.lifecycleMonitor) {
-			this.resumePausedTasks();
-		}
-	}
-
-	public void doStop() {
-		this.shutdown();
-	}
-
-	public void doInitialize() {
-		synchronized (this.activeInvokerMonitor) {
-			if (this.executor == null) {
-				this.executor = createDefaultExecutor();
-			}
-			else if (this.executor instanceof SchedulingTaskExecutor &&
-					((SchedulingTaskExecutor) this.executor).prefersShortLivedTasks() &&
-					this.maxMessagesPerTask == Integer.MIN_VALUE) {
-				this.maxMessagesPerTask = 1;
-			}
-			initializeExecutorIfPossible();
-			for (int i = 0; i < this.concurrency; i++) {
-				scheduleNewInvoker();
-			}
-		}
-	}
-
-	private void initializeExecutorIfPossible() {
-		try {
-			Method initMethod = this.executor.getClass().getMethod("initialize");
-			initMethod.invoke(this.executor);
-		}
-		catch (Exception e) {
-			// do nothing
-		}
-	}
-
-	private void shutdown() {
-		synchronized (this.lifecycleMonitor) {
-			this.shutdownExecutorIfPossible();
-		}
-	}
-
-	private void shutdownExecutorIfPossible() {
-		try {
-			if (this.executor instanceof Lifecycle) {
-				((Lifecycle) this.executor).stop();
-			}
-			else {
-				Method shutdownMethod = this.executor.getClass().getMethod("shutdown");
-				shutdownMethod.invoke(this.executor);
-			}
-		}
-		catch (Exception e) {
-			// do nothing
-		}
-	}
-
-	private TaskExecutor createDefaultExecutor() {
-		ThreadPoolTaskExecutor defaultExecutor = new ThreadPoolTaskExecutor();
-		defaultExecutor.setCorePoolSize(this.maxConcurrency);
-		defaultExecutor.setMaxPoolSize(this.maxConcurrency);
-		defaultExecutor.setQueueCapacity(5);
-		return defaultExecutor;
-	}
-
-	/**
-	 * Try scheduling a new invoker, since we know messages are being received.
-	 * @see #scheduleNewInvokerIfAppropriate()
-	 */
-	@Override
-	protected void messageReceived(Message message) {
-		scheduleNewInvokerIfAppropriate();
-	}
-
-	/**
-	 * Schedule a new invoker, increasing the total number of scheduled
-	 * invokers for this consumer.
-	 */
-	private void scheduleNewInvoker() {
-		MessageEndpointInvoker invoker = new MessageEndpointInvoker();
-		if (rescheduleInvokerIfNecessary(invoker)) {
-			this.scheduledInvokers.add(invoker);
-		}
-	}
-
-	private boolean rescheduleInvokerIfNecessary(MessageEndpointInvoker invoker) {
-		synchronized (this.lifecycleMonitor) {
-			if (this.isRunning()) {
-				try {
-					doRescheduleInvoker(invoker);
-				}
-				catch (RuntimeException ex) {
-					logRejectedInvoker(invoker, ex);
-					this.pausedInvokers.add(invoker);
-				}
-				return true;
-			}
-			if (this.isActive()) {
-				this.pausedInvokers.add(invoker);
-				return true;
-			}
-			else {
-				return false;
-			}
-		}
-	}
-
-	protected void doRescheduleInvoker(final MessageEndpointInvoker invoker) {
-		this.executor.execute(invoker);
-	}
-
-	private boolean shouldRescheduleInvoker(int idleTaskExecutionCount) {
-		synchronized (this.activeInvokerMonitor) {
-			boolean idle = (idleTaskExecutionCount >= this.idleTaskExecutionLimit);
-			return (this.scheduledInvokers.size() <= (idle ? this.concurrency : this.maxConcurrency));
-		}
-	}
-
-	private boolean hasIdleInvokers() {
-		for (MessageEndpointInvoker invoker : this.scheduledInvokers) {
-			if (invoker.isIdle()) {
-				return true;
-			}
-		}
-		return false;
-	}
-
-	/**
-	 * Try to resume all paused tasks.
-	 * Tasks for which rescheduling failed simply remain in paused mode.
-	 */
-	protected void resumePausedTasks() {
-		synchronized (this.lifecycleMonitor) {
-			if (!this.pausedInvokers.isEmpty()) {
-				for (Iterator<MessageEndpointInvoker> it = this.pausedInvokers.iterator(); it.hasNext();) {
-					MessageEndpointInvoker invoker = it.next();
-					try {
-						doRescheduleInvoker(invoker);
-						it.remove();
-						if (logger.isDebugEnabled()) {
-							logger.debug("Resumed paused invoker: " + invoker);
-						}
-					}
-					catch (RuntimeException e) {
-						logRejectedInvoker(invoker, e);
-						// Keep the task in paused mode...
-					}
-				}
-			}
-		}
-	}
-
-	public int getPausedInvokerCount() {
-		synchronized (this.lifecycleMonitor) {
-			return this.pausedInvokers.size();
-		}
-	}
-
-	/**
-	 * Log an invoker that has been rejected by {@link #doRescheduleInvoker}.
-	 * <p>The default implementation simply logs a corresponding message
-	 * at debug level.
-	 * @param invoker the rejected invoker object
-	 * @param ex the exception thrown from {@link #doRescheduleInvoker}
-	 */
-	protected void logRejectedInvoker(Object invoker, RuntimeException ex) {
-		if (logger.isDebugEnabled()) {
-			logger.debug("Invoker [" + invoker + "] has been rejected and paused: " + ex);
-		}
-	}
-
-	private void scheduleNewInvokerIfAppropriate() {
-		if (this.isRunning()) {
-			this.resumePausedTasks();
-			synchronized (this.activeInvokerMonitor) {
-				if (this.scheduledInvokers.size() < this.maxConcurrency && !hasIdleInvokers()) {
-					scheduleNewInvoker();
-					if (logger.isDebugEnabled()) {
-						logger.debug("Raised scheduled invoker count: " + scheduledInvokers.size());
-					}
-				}
-			}
-		}
-	}
-
-	public final int getScheduledInvokerCount() {
-		synchronized (this.activeInvokerMonitor) {
-			return this.scheduledInvokers.size();
-		}
-	}
-
-	public final int getActiveInvokerCount() {
-		synchronized (this.activeInvokerMonitor) {
-			return this.activeInvokerCount;
-		}
-	}
-
-
-	private class MessageEndpointInvoker implements SchedulingAwareRunnable {
-
-		private int idleTaskExecutionCount = 0;
-
-		private volatile boolean idle = true;
-
-
-		public void run() {
-			synchronized (activeInvokerMonitor) {
-				activeInvokerCount++;
-				activeInvokerMonitor.notifyAll();
-			}
-			boolean messageReceived = false;
-			//TODO: try {
-				if (maxMessagesPerTask < 0) {
-					while (isActive()) {
-						waitWhileNotRunning();
-						if (isActive()) {
-							messageReceived = receiveAndPassToEndpoint();
-							this.idle = !messageReceived;
-						}
-					}
-				}
-				else {
-					int messageCount = 0;
-					while (isRunning() && messageCount < maxMessagesPerTask) {
-						boolean messageHandled = receiveAndPassToEndpoint();
-						this.idle = !messageHandled;
-						messageReceived = (messageHandled || messageReceived);
-						messageCount++;
-					}
-				}
-			// TODO: } catch (Throwable t) { check if last message succeeded, else sleep between recovery attempts } 
-			synchronized (activeInvokerMonitor) {
-				activeInvokerCount--;
-				activeInvokerMonitor.notifyAll();
-			}
-			if (!messageReceived) {
-				this.idleTaskExecutionCount++;
-			}
-			else {
-				this.idleTaskExecutionCount = 0;
-			}
-			if (!shouldRescheduleInvoker(this.idleTaskExecutionCount) || !rescheduleInvokerIfNecessary(this)) {
-				this.shutdown();
-			}
-			else if (isRunning()) {
-				int nonPausedInvokers = getScheduledInvokerCount() - getPausedInvokerCount();
-				if (nonPausedInvokers < 1) {
-					logger.error("All scheduled invokers have been paused, probably due to tasks having been rejected. " +
-							"Check your thread pool configuration! Manual recovery necessary through a start() call.");
-				}
-				else if (nonPausedInvokers < concurrency) {
-					logger.warn("Number of scheduled invokers has dropped below concurrency limit, probably " +
-							"due to tasks having been rejected. Check your thread pool configuration! Automatic recovery " +
-							"to be triggered by remaining invokers.");
-				}
-			}
-		}
-
-		private void shutdown() {
-			synchronized (activeInvokerMonitor) {
-				scheduledInvokers.remove(this);
-				if (logger.isDebugEnabled()) {
-					logger.debug("Lowered scheduled invoker count: " + scheduledInvokers.size());
-				}
-				activeInvokerMonitor.notifyAll();
-			}			
-		}
-
-		public boolean isLongLived() {
-			return (maxMessagesPerTask < 0);
-		}
-
-		public boolean isIdle() {
-			return this.idle;
-		}
-
-	}
-
-}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java

@@ -1,42 +0,0 @@
-/*
- * Copyright 2002-2007 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.consumer;
-
-import java.util.concurrent.TimeUnit;
-
-import org.springframework.integration.MessageSource;
-import org.springframework.integration.endpoint.MessageEndpoint;
-
-/**
- * A consumer that measures the <code>pollInterval</code> between the
- * invoker's execution completion time and the subsequent start.
- * 
- * @author Mark Fisher
- */
-public class FixedDelayConsumer extends AbstractPollingConsumer {
-
-	public FixedDelayConsumer(MessageSource source, MessageEndpoint endpoint) {
-		super(source, endpoint);
-	}
-
-
-	@Override
-	protected void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit) {
-		this.executor.scheduleWithFixedDelay(invoker, initialDelay, pollInterval, timeUnit);
-	}
-
-}

spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java

@@ -1,42 +0,0 @@
-/*
- * Copyright 2002-2007 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.consumer;
-
-import java.util.concurrent.TimeUnit;
-
-import org.springframework.integration.MessageSource;
-import org.springframework.integration.endpoint.MessageEndpoint;
-
-/**
- * A consumer that measures the <code>pollInterval</code> between each
- * execution's start.
- * 
- * @author Mark Fisher
- */
-public class FixedRateConsumer extends AbstractPollingConsumer {
-
-	public FixedRateConsumer(MessageSource source, MessageEndpoint endpoint) {
-		super(source, endpoint);
-	}
-
-
-	@Override
-	protected void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit) {
-		this.executor.scheduleAtFixedRate(invoker, initialDelay, pollInterval, timeUnit);
-	}
-
-}

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

@@ -20,7 +20,6 @@ import org.springframework.integration.MessageHandlingException;
 import org.springframework.integration.MessageSource;
 import org.springframework.integration.MessageTarget;
 import org.springframework.integration.channel.ChannelResolver;
-import org.springframework.integration.channel.consumer.ConsumerType;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
 
@@ -47,8 +46,6 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 
 	private ChannelResolver channelResolver;
 
-	private ConsumerType consumerType = ConsumerType.EVENT_DRIVEN;
-
 
 	/**
 	 * Set the source from which this endpoint receives messages.
@@ -78,20 +75,6 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 		this.handler = handler;
 	}
 
-	/**
-	 * Set the type of consumer to use for this endpoint.
-	 */
-	public void setConsumerType(ConsumerType consumerType) {
-		this.consumerType = consumerType;
-	}
-
-	/**
-	 * Return the type of consumer to use for this endpoint.
-	 */
-	public ConsumerType getConsumerType() {
-		return this.consumerType;
-	}
-
 	/**
 	 * Set the channel resolver strategy to use when a message
 	 * provides a '<i>replyChannelName</i>'.
@@ -104,6 +87,7 @@ public class GenericMessageEndpoint implements MessageEndpoint {
 	public void messageReceived(Message message) {
 		if (this.handler == null) {
 			target.send(message);
+			return;
 		}
 		Message replyMessage = handler.handle(message);
 		if (replyMessage != null) {

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

@@ -18,7 +18,6 @@ package org.springframework.integration.endpoint;
 
 import org.springframework.integration.MessageSource;
 import org.springframework.integration.channel.ChannelResolver;
-import org.springframework.integration.channel.consumer.ConsumerType;
 import org.springframework.integration.message.Message;
 
 /**
@@ -28,8 +27,6 @@ import org.springframework.integration.message.Message;
  */
 public interface MessageEndpoint {
 
-	ConsumerType getConsumerType();
-
 	MessageSource getSource();
 
 	void setChannelResolver(ChannelResolver channelResolver);

spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java

@@ -23,6 +23,8 @@ import org.junit.Test;
 
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
@@ -69,6 +71,10 @@ public class MessageBusTests {
 		MessageChannel sourceChannel = (MessageChannel) context.getBean("sourceChannel");
 		sourceChannel.send(new DocumentMessage("123", "test"));		
 		MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel");
+		// TODO: add metadata for this
+		MessageBus bus = (MessageBus) context.getBean("bus");
+		ConsumerPolicy policy = new ConsumerPolicy();
+		bus.activateSubscription("sourceChannel", "endpoint", policy);
 		Message result = targetChannel.receive(10);
 		assertEquals("test", result.getPayload());
 	}

spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel.consumer;
+package org.springframework.integration.bus;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -24,12 +24,14 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
+
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 /**
  * @author Mark Fisher
@@ -37,49 +39,54 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 public class EventDrivenConsumerTests {
 
 	@Test
+	public void stub() {}
+
+	// TODO: make this a @Test
 	public void testDynamicConcurrency() throws Exception {
 		int messagesToSend = 200;
 		int concurrency = 1;
-		int maxConcurrency = 10;
+		int maxConcurrency = 100;
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
-		final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
 		final AtomicInteger maxActive = new AtomicInteger(0);
 		final AtomicInteger activeSum = new AtomicInteger(0);
-		executor.setCorePoolSize(concurrency);
-		executor.setMaxPoolSize(maxConcurrency);
-		executor.setQueueCapacity(0);
+		final MessageBus bus = new MessageBus();
 		PointToPointChannel channel = new PointToPointChannel();
 		MessageEndpoint endpoint = new GenericMessageEndpoint() {
 			public void messageReceived(Message message) {
 				counter.incrementAndGet();
 				latch.countDown();
-				activeSum.set(activeSum.addAndGet(executor.getActiveCount()));
-				maxActive.set(Math.max(executor.getActiveCount(), maxActive.get()));
+				activeSum.set(activeSum.addAndGet(bus.getActiveCountForEndpoint("testEndpoint")));
+				maxActive.set(Math.max(bus.getActiveCountForEndpoint("testEndpoint"), maxActive.get()));
 			}
 		};
-		EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint);
-		consumer.setExecutor(executor);
-		consumer.setConcurrency(concurrency);
-		consumer.setMaxConcurrency(maxConcurrency);
-		consumer.setIdleTaskExecutionLimit(1);
-		consumer.setMaxMessagesPerTask(1);
-		consumer.setReceiveTimeout(100);
-		consumer.initialize();
-		consumer.start();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setConcurrency(concurrency);
+		policy.setMaxConcurrency(maxConcurrency);
+		policy.setMaxMessagesPerTask(1);
+		policy.setRejectionLimit(1);
+		policy.setPeriod(0);
+		policy.setReceiveTimeout(100);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		for (int i = 0; i < messagesToSend - 110; i++) {
 			channel.send(new DocumentMessage(1, "fast-1." + (i+1)));
 		}
-		int activeCountAfterFirstBurst = executor.getActiveCount();
+		int activeCountAfterFirstBurst = bus.getActiveCountForEndpoint("testEndpoint");
+		System.out.println("after-first: " + activeCountAfterFirstBurst);
 		for (int i = 0; i < 10; i++) {
 			channel.send(new DocumentMessage(1, "slow-1." + (i+1)));
 			Thread.sleep(50);
 		}
-		int activeCountAfterSlowDown = executor.getActiveCount();
+		int activeCountAfterSlowDown = bus.getActiveCountForEndpoint("testEndpoint");
+		System.out.println("after-slowdown: " + activeCountAfterSlowDown);
 		for (int i = 0; i < 100; i++) {
 			channel.send(new DocumentMessage(1, "fast-2." + (i+1)));
 		}
-		int activeCountAfterLastBurst = executor.getActiveCount();
+		int activeCountAfterLastBurst = bus.getActiveCountForEndpoint("testEndpoint");
+		System.out.println("after-last: " + activeCountAfterLastBurst);
 		latch.await(10, TimeUnit.SECONDS);
 		int averageActive = activeSum.get() / messagesToSend;
 		assertTrue(activeCountAfterSlowDown < activeCountAfterFirstBurst);

spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel.consumer;
+package org.springframework.integration.bus;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -24,6 +24,8 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
@@ -47,10 +49,17 @@ public class FixedDelayConsumerTests {
 				latch.countDown();
 			}
 		};
-		FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint);
-		consumer.setPollInterval(10);
-		consumer.initialize();
-		consumer.start();
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setConcurrency(1);
+		policy.setMaxConcurrency(1);
+		policy.setMaxMessagesPerTask(1);
+		policy.setFixedRate(true);
+		policy.setPeriod(10);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new DocumentMessage(1, "test " + (i+1)));
 		}
@@ -70,10 +79,17 @@ public class FixedDelayConsumerTests {
 				latch.countDown();
 			}
 		};
-		FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint);
-		consumer.setPollInterval(10);
-		consumer.initialize();
-		consumer.start();
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setConcurrency(1);
+		policy.setMaxConcurrency(1);
+		policy.setMaxMessagesPerTask(1);
+		policy.setFixedRate(true);
+		policy.setPeriod(10);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new DocumentMessage(1, "test " + (i+1)));
 		}

spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.integration.channel.consumer;
+package org.springframework.integration.bus;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
@@ -24,6 +24,9 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
+
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.PointToPointChannel;
 import org.springframework.integration.endpoint.GenericMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
@@ -47,10 +50,14 @@ public class FixedRateConsumerTests {
 				latch.countDown();
 			}
 		};
-		FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint);
-		consumer.setPollInterval(10);
-		consumer.initialize();
-		consumer.start();
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setFixedRate(true);
+		policy.setPeriod(10);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new DocumentMessage(1, "test " + (i+1)));
 		}
@@ -70,10 +77,17 @@ public class FixedRateConsumerTests {
 				latch.countDown();
 			}
 		};
-		FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint);
-		consumer.setPollInterval(10);
-		consumer.initialize();
-		consumer.start();
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setConcurrency(1);
+		policy.setMaxConcurrency(1);
+		policy.setMaxMessagesPerTask(1);
+		policy.setFixedRate(true);
+		policy.setPeriod(10);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new DocumentMessage(1, "test " + (i+1)));
 		}

spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java

@@ -21,10 +21,11 @@ import static org.junit.Assert.assertNotNull;
 
 import org.junit.Test;
 
+import org.springframework.integration.bus.ConsumerPolicy;
+import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.ChannelResolver;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.PointToPointChannel;
-import org.springframework.integration.channel.consumer.EventDrivenConsumer;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.DocumentMessage;
 import org.springframework.integration.message.Message;
@@ -47,9 +48,13 @@ public class GenericMessageEndpointTests {
 		endpoint.setSource(channel);
 		endpoint.setHandler(handler);
 		endpoint.setTarget(replyChannel);
-		EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint);
-		consumer.initialize();
-		consumer.start();
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setPeriod(0);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		DocumentMessage testMessage = new DocumentMessage(1, "test");
 		channel.send(testMessage);
 		Message reply = replyChannel.receive(10);
@@ -78,9 +83,13 @@ public class GenericMessageEndpointTests {
 		endpoint.setSource(channel);
 		endpoint.setHandler(handler);
 		endpoint.setChannelResolver(channelResolver);
-		EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint);
-		consumer.initialize();
-		consumer.start();
+		MessageBus bus = new MessageBus();
+		bus.registerChannel("testChannel", channel);
+		bus.registerEndpoint("testEndpoint", endpoint);
+		ConsumerPolicy policy = new ConsumerPolicy();
+		policy.setPeriod(0);
+		bus.activateSubscription("testChannel", "testEndpoint", policy);
+		bus.start();
 		DocumentMessage testMessage = new DocumentMessage(1, "test");
 		testMessage.getHeader().setReplyChannelName("replyChannel");
 		channel.send(testMessage);

spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml

@@ -4,7 +4,7 @@
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans.xsd">
 
-	<bean class="org.springframework.integration.MessageBus"/>
+	<bean id="bus" class="org.springframework.integration.bus.MessageBus"/>
 
 	<bean id="sourceChannel" class="org.springframework.integration.channel.PointToPointChannel"/>