diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java index 617571f8fe..4f86a86624 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -21,7 +21,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; @@ -41,8 +40,6 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial private MessageMapper mapper = new SimplePayloadMessageMapper(); - private ConsumerPolicy consumerPolicy; - private long sendTimeout = -1; private volatile boolean initialized = false; @@ -70,15 +67,6 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial return this.mapper; } - public void setConsumerPolicy(ConsumerPolicy consumerPolicy) { - Assert.notNull(consumerPolicy, "'consumerPolicy' must not be null"); - this.consumerPolicy = consumerPolicy; - } - - public ConsumerPolicy getConsumerPolicy() { - return this.consumerPolicy; - } - public final void afterPropertiesSet() { if (this.channel == null) { throw new MessagingConfigurationException("'channel' is required"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java index 08c5267c11..89828577d1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java @@ -19,11 +19,12 @@ package org.springframework.integration.adapter; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; import org.springframework.integration.message.SimplePayloadMessageMapper; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; /** @@ -41,7 +42,7 @@ public abstract class AbstractTargetAdapter implements TargetAdapter { private MessageMapper mapper = new SimplePayloadMessageMapper(); - private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(5); + private Schedule schedule = new PollingSchedule(5); public void setName(String name) { @@ -70,13 +71,13 @@ public abstract class AbstractTargetAdapter implements TargetAdapter { return this.mapper; } - public void setConsumerPolicy(ConsumerPolicy policy) { - Assert.notNull(policy, "'policy' must not be null"); - this.policy = policy; + public void setSchedule(Schedule schedule) { + Assert.notNull(schedule, "'schedule' must not be null"); + this.schedule = schedule; } - public ConsumerPolicy getConsumerPolicy() { - return this.policy; + public Schedule getSchedule() { + return this.schedule; } public final Message handle(Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index 2a63965c81..d0fa26fe5e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -18,11 +18,16 @@ package org.springframework.integration.adapter; import java.util.Collection; +import org.springframework.context.Lifecycle; import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.bus.MessageDispatcher; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.scheduling.MessagingTask; +import org.springframework.integration.scheduling.MessagingTaskScheduler; +import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.util.Assert; /** @@ -32,19 +37,44 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class PollingSourceAdapter extends AbstractSourceAdapter implements MessageDispatcher { - - private static int DEFAULT_PERIOD = 1000; +public class PollingSourceAdapter extends AbstractSourceAdapter implements MessagingTaskSchedulerAware, Lifecycle { private PollableSource source; + private PollingSchedule schedule = new PollingSchedule(1000); + + private MessagingTaskScheduler scheduler; + + private int maxMessagesPerTask = 1; + + private volatile boolean starting; + private volatile boolean running; public PollingSourceAdapter(PollableSource source) { Assert.notNull(source, "'source' must not be null"); this.source = source; - this.setConsumerPolicy(ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD)); + } + + + public void setInitialDelay(long intialDelay) { + Assert.isTrue(intialDelay >= 0, "'intialDelay' must not be negative"); + this.schedule.setInitialDelay(intialDelay); + } + + public void setPeriod(long period) { + this.schedule.setPeriod(period); + } + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least one"); + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) { + Assert.notNull(scheduler, "scheduler must not be null"); + this.scheduler = scheduler; } protected PollableSource getSource() { @@ -56,32 +86,42 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements } public void start() { + if (this.isRunning() || this.starting) { + return; + } + this.starting = true; if (!this.isInitialized()) { this.afterPropertiesSet(); } + if (this.scheduler == null) { + if (logger.isInfoEnabled()) { + logger.info("no task scheduler has been provided, will create one"); + } + SimpleMessagingTaskScheduler taskScheduler = new SimpleMessagingTaskScheduler(); + taskScheduler.setCorePoolSize(1); + this.scheduler = taskScheduler; + } + if (!this.scheduler.isRunning()) { + this.scheduler.start(); + } + this.scheduler.schedule(new PollingSourceAdapterTask()); this.running = true; + this.starting = false; } public void stop() { this.running = false; } - public void setPeriod(int period) { - Assert.isTrue(period > 0, "'period' must be a positive value"); - this.getConsumerPolicy().setPeriod(period); - } - - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value"); - this.getConsumerPolicy().setMaxMessagesPerTask(maxMessagesPerTask); - } - - public int dispatch() { + public int processMessages() { if (!this.isRunning()) { + if (logger.isDebugEnabled()) { + logger.debug("source adapter not polling since it has not yet been started"); + } return 0; } int messagesProcessed = 0; - int limit = this.getConsumerPolicy().getMaxMessagesPerTask(); + int limit = this.maxMessagesPerTask; Collection results = this.source.poll(limit); if (results != null) { if (results.size() > limit) { @@ -96,4 +136,16 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements return messagesProcessed; } + + private class PollingSourceAdapterTask implements MessagingTask { + + public void run() { + processMessages(); + } + + public Schedule getSchedule() { + return schedule; + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java index 161e6b3b5c..5fe72ad111 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java @@ -16,7 +16,6 @@ package org.springframework.integration.adapter; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; /** @@ -26,8 +25,6 @@ import org.springframework.integration.channel.MessageChannel; */ public interface SourceAdapter { - ConsumerPolicy getConsumerPolicy(); - void setChannel(MessageChannel channel); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java index 3e8ffdf627..6f7b7a3ae7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/TargetAdapter.java @@ -16,7 +16,6 @@ package org.springframework.integration.adapter; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; @@ -31,6 +30,4 @@ public interface TargetAdapter extends MessageHandler { void setChannel(MessageChannel channel); - ConsumerPolicy getConsumerPolicy(); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java index 78ee301a9c..d7fd93adc4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java @@ -24,7 +24,6 @@ import org.springframework.context.Lifecycle; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.adapter.AbstractSourceAdapter; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.jms.listener.AbstractJmsListeningContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.support.converter.MessageConverter; @@ -50,7 +49,15 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter private TaskExecutor taskExecutor; - private ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); + private long receiveTimeout = 1000; + + private int concurrentConsumers = 1; + + private int maxConcurrentConsumers = 1; + + private int maxMessagesPerTask = Integer.MIN_VALUE; + + private int idleTaskExecutionLimit = 1; private long sendTimeout = -1; @@ -104,10 +111,11 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter if (this.destinationName != null) { dmlc.setDestinationName(this.destinationName); } - dmlc.setReceiveTimeout(this.policy.getReceiveTimeout()); - dmlc.setConcurrentConsumers(this.policy.getConcurrency()); - dmlc.setMaxConcurrentConsumers(this.policy.getMaxConcurrency()); - dmlc.setMaxMessagesPerTask(this.policy.getMaxMessagesPerTask()); + dmlc.setReceiveTimeout(this.receiveTimeout); + dmlc.setConcurrentConsumers(this.concurrentConsumers); + dmlc.setMaxConcurrentConsumers(this.maxConcurrentConsumers); + dmlc.setMaxMessagesPerTask(this.maxMessagesPerTask); + dmlc.setIdleTaskExecutionLimit(this.idleTaskExecutionLimit); dmlc.setAutoStartup(false); ChannelPublishingJmsListener listener = new ChannelPublishingJmsListener(this.getChannel()); listener.setMessageConverter(this.messageConverter); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java deleted file mode 100644 index a3e5ca33af..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java +++ /dev/null @@ -1,121 +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.bus; - -import java.util.Collection; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.context.Lifecycle; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.Message; -import org.springframework.util.Assert; - -/** - * Abstract base class for message dispatchers. Delegates to a - * {@link MessageRetriever} strategy. - * - * @author Mark Fisher - */ -public abstract class AbstractMessageDispatcher implements MessageDispatcher { - - protected Log logger = LogFactory.getLog(this.getClass()); - - private MessageRetriever retriever; - - private List handlers = new CopyOnWriteArrayList(); - - private volatile boolean running; - - private Object lifecycleMonitor = new Object(); - - - public AbstractMessageDispatcher(MessageRetriever retriever) { - this.retriever = retriever; - } - - - public void addHandler(MessageHandler handler) { - Assert.notNull(handler, "'handler' must not be null"); - if (this.isRunning() && handler instanceof Lifecycle) { - ((Lifecycle) handler).start(); - } - this.handlers.add(handler); - } - - protected List getHandlers() { - return this.handlers; - } - - public boolean isRunning() { - return this.running; - } - - public void start() { - synchronized (this.lifecycleMonitor) { - if (!this.isRunning()) { - for (MessageHandler handler : this.handlers) { - if (handler instanceof Lifecycle) { - ((Lifecycle) handler).start(); - } - } - this.running = true; - } - } - } - - public void stop() { - synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - for (MessageHandler handler : this.handlers) { - if (handler instanceof Lifecycle) { - ((Lifecycle) handler).stop(); - } - } - this.running = false; - } - } - } - - /** - * Retrieves messages and dispatches to the executors. - * - * @return the number of messages processed - */ - public int dispatch() { - if (!this.isRunning()) { - return 0; - } - int messagesProcessed = 0; - Collection> messages = this.retriever.retrieveMessages(); - if (messages == null) { - return 0; - } - for (Message message : messages) { - if (dispatchMessage(message)) { - messagesProcessed++; - } - } - return messagesProcessed; - } - - - protected abstract boolean dispatchMessage(Message message); - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java deleted file mode 100644 index b1e5ca4011..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java +++ /dev/null @@ -1,181 +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.bus; - -import java.util.concurrent.TimeUnit; - -import org.springframework.util.Assert; - -/** - * 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 = 1; - - private static final int DEFAULT_REJECTION_LIMIT = 5; - - private static final int DEFAULT_RETRY_INTERVAL = 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 retryInterval = DEFAULT_RETRY_INTERVAL; - - private int initialDelay = 0; - - private int period = 5; - - private TimeUnit timeUnit = TimeUnit.MILLISECONDS; - - private boolean fixedRate = false; - - private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - - - /** - * Factory method for a basic polling policy. - * - * @param period the polling interval - */ - public static ConsumerPolicy newPollingPolicy(int period) { - Assert.isTrue(period > 0, "'period' must be a positive value"); - ConsumerPolicy policy = new ConsumerPolicy(); - policy.setPeriod(period); - policy.setConcurrency(1); - policy.setMaxConcurrency(1); - return policy; - } - - /** - * Factory method for a basic event-driven policy. - */ - public static ConsumerPolicy newEventDrivenPolicy() { - ConsumerPolicy policy = new ConsumerPolicy(); - policy.setPeriod(-1); - policy.setConcurrency(1); - policy.setMaxConcurrency(10); - return policy; - } - - 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("'rejectionLimit' must be at least 1"); - } - this.rejectionLimit = rejectionLimit; - } - - public int getRetryInterval() { - return this.retryInterval; - } - - public void setRetryInterval(int retryInterval) { - this.retryInterval = retryInterval; - } - - public long getReceiveTimeout() { - return this.receiveTimeout; - } - - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 83ce9fe74d..db12e23be7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -20,7 +20,6 @@ import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.ScheduledThreadPoolExecutor; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -37,9 +36,17 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.DefaultMessageDispatcher; +import org.springframework.integration.dispatcher.DispatcherPolicy; +import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.scheduling.MessagePublishingErrorHandler; +import org.springframework.integration.scheduling.MessagingTaskScheduler; +import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -58,23 +65,27 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private Map handlers = new ConcurrentHashMap(); - private List dispatchers = new CopyOnWriteArrayList(); + private Map dispatchers = new ConcurrentHashMap(); - private List dispatcherTasks = new CopyOnWriteArrayList(); + private List lifecycleSourceAdapters = new CopyOnWriteArrayList(); - private ScheduledThreadPoolExecutor dispatcherExecutor; + private MessagingTaskScheduler taskScheduler; private int dispatcherPoolSize = 10; private boolean autoCreateChannels; - private boolean running; + private volatile boolean initialized; + + private volatile boolean starting; + + private volatile boolean running; private Object lifecycleMonitor = new Object(); public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - Assert.notNull(applicationContext, "applicationContext must not be null"); + Assert.notNull(applicationContext, "'applicationContext' must not be null"); this.registerChannels(applicationContext); this.registerEndpoints(applicationContext); this.registerSourceAdapters(applicationContext); @@ -88,8 +99,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif public void setDispatcherPoolSize(int dispatcherPoolSize) { Assert.isTrue(dispatcherPoolSize > 0, "'dispatcherPoolSize' must be at least 1"); this.dispatcherPoolSize = dispatcherPoolSize; - if (this.dispatcherExecutor != null) { - this.dispatcherExecutor.setCorePoolSize(dispatcherPoolSize); + if (this.taskScheduler != null && this.taskScheduler instanceof SimpleMessagingTaskScheduler) { + ((SimpleMessagingTaskScheduler) this.taskScheduler).setCorePoolSize(dispatcherPoolSize); } } @@ -145,23 +156,29 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.activateSubscription(subscription); if (logger.isInfoEnabled()) { logger.info("activated subscription to channel '" + subscription.getChannel() + - "' for receiver '" + subscription.getReceiver() + "'"); + "' for handler '" + subscription.getHandler() + "'"); } } } public void initialize() { - initDispatcherExecutor(); if (this.getInvalidMessageChannel() == null) { this.setInvalidMessageChannel(new SimpleChannel(Integer.MAX_VALUE)); } + initScheduler(); + this.initialized = true; } - private void initDispatcherExecutor() { + private void initScheduler() { CustomizableThreadFactory threadFactory = new CustomizableThreadFactory(); threadFactory.setThreadNamePrefix("dispatcher-executor-"); threadFactory.setThreadGroup(new ThreadGroup("dispatcher-executors")); - this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherPoolSize, threadFactory); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); + scheduler.setCorePoolSize(this.dispatcherPoolSize); + scheduler.setThreadFactory(threadFactory); + scheduler.setErrorHandler(new MessagePublishingErrorHandler(this.getInvalidMessageChannel())); + scheduler.afterPropertiesSet(); + this.taskScheduler = scheduler; } public MessageChannel getInvalidMessageChannel() { @@ -177,23 +194,42 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void registerChannel(String name, MessageChannel channel) { + this.registerChannel(name, channel, null); + } + + public void registerChannel(String name, MessageChannel channel, DispatcherPolicy dispatcherPolicy) { + if (!this.initialized) { + this.initialize(); + } + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.setMessagingTaskScheduler(this.taskScheduler); + if (dispatcherPolicy != null) { + dispatcher.setMaxMessagesPerTask(dispatcherPolicy.getMaxMessagesPerTask()); + dispatcher.setReceiveTimeout(dispatcherPolicy.getReceiveTimeout()); + dispatcher.setRejectionLimit(dispatcherPolicy.getRejectionLimit()); + dispatcher.setRetryInterval(dispatcherPolicy.getRetryInterval()); + } + this.dispatchers.put(channel, dispatcher); this.channelRegistry.registerChannel(name, channel); } public void registerEndpoint(String name, MessageEndpoint endpoint) { + if (!this.initialized) { + this.initialize(); + } Assert.notNull(name, "'name' must not be null"); Assert.notNull(endpoint, "'endpoint' must not be null"); endpoint.setName(name); this.handlers.put(name, endpoint); - if (logger.isInfoEnabled()) { - logger.info("registered endpoint '" + name + "'"); - } endpoint.setChannelRegistry(this); - if (endpoint.getInputChannelName() != null && endpoint.getConsumerPolicy() != null) { + Schedule schedule = endpoint.getSchedule(); + if (endpoint.getInputChannelName() != null) { Subscription subscription = new Subscription(); + subscription.setHandler(name); subscription.setChannel(endpoint.getInputChannelName()); - subscription.setReceiver(name); - subscription.setPolicy(endpoint.getConsumerPolicy()); + if (schedule != null) { + subscription.setSchedule(schedule); + } this.activateSubscription(subscription); } if (this.autoCreateChannels) { @@ -205,12 +241,17 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void registerSourceAdapter(String name, SourceAdapter adapter) { - ConsumerPolicy policy = adapter.getConsumerPolicy(); - if (adapter instanceof MessageDispatcher) { - MessageDispatcher dispatcher = (MessageDispatcher) adapter; - this.dispatchers.add(dispatcher); - DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); - this.addDispatcherTask(dispatcherTask); + if (!this.initialized) { + this.initialize(); + } + if (adapter instanceof MessagingTaskSchedulerAware) { + ((MessagingTaskSchedulerAware) adapter).setMessagingTaskScheduler(this.taskScheduler); + } + if (adapter instanceof Lifecycle) { + this.lifecycleSourceAdapters.add((Lifecycle) adapter); + if (this.isRunning()) { + ((Lifecycle) adapter).start(); + } } if (logger.isInfoEnabled()) { logger.info("registered source adapter '" + name + "'"); @@ -221,10 +262,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif if (targetAdapter instanceof AbstractTargetAdapter) { AbstractTargetAdapter adapter = (AbstractTargetAdapter) targetAdapter; adapter.setName(name); - this.handlers.put(name, targetAdapter); + this.handlers.put(name, adapter); MessageChannel channel = adapter.getChannel(); - ConsumerPolicy policy = adapter.getConsumerPolicy(); - this.doActivate(channel, adapter, policy); + Schedule schedule = adapter.getSchedule(); + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); + concurrencyPolicy.setCoreConcurrency(1); + concurrencyPolicy.setMaxConcurrency(1); + this.doActivate(channel, adapter, schedule, concurrencyPolicy); } if (logger.isInfoEnabled()) { logger.info("registered target adapter '" + name + "'"); @@ -233,8 +277,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif public void activateSubscription(Subscription subscription) { String channelName = subscription.getChannel(); - String handlerName = subscription.getReceiver(); - ConsumerPolicy policy = subscription.getPolicy(); + String handlerName = subscription.getHandler(); + Schedule schedule = subscription.getSchedule(); + ConcurrencyPolicy concurrencyPolicy = subscription.getConcurrencyPolicy(); MessageHandler handler = this.handlers.get(handlerName); if (handler == null) { throw new MessagingException("Cannot activate subscription, unknown handler '" + handlerName + "'"); @@ -251,58 +296,27 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif channel = new SimpleChannel(); this.registerChannel(channelName, channel); } - this.doActivate(channel, handler, policy); + this.doActivate(channel, handler, schedule, concurrencyPolicy); if (logger.isInfoEnabled()) { logger.info("activated subscription to channel '" + channelName + "' for handler '" + handlerName + "'"); } } - private void doActivate(MessageChannel channel, MessageHandler handler, ConsumerPolicy policy) { - PooledMessageHandler pooledHandler = new PooledMessageHandler(handler, policy.getConcurrency(), policy.getMaxConcurrency()); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setRejectionLimit(policy.getRejectionLimit()); - dispatcher.setRetryInterval(policy.getRetryInterval()); - dispatcher.addHandler(pooledHandler); - DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); + private void doActivate(MessageChannel channel, MessageHandler handler, Schedule schedule, ConcurrencyPolicy concurrencyPolicy) { + MessageDispatcher dispatcher = dispatchers.get(channel); + if (dispatcher == null) { + if (logger.isWarnEnabled()) { + logger.warn("no dispatcher available for channel '" + channel + "', be sure to register the channel"); + } + } + if (concurrencyPolicy != null) { + handler = new PooledMessageHandler(handler, concurrencyPolicy.getCoreConcurrency(), concurrencyPolicy.getMaxConcurrency()); + } + dispatcher.addHandler(handler, schedule); if (this.isRunning()) { dispatcher.start(); } - this.dispatchers.add(dispatcher); - this.addDispatcherTask(dispatcherTask); - if (this.logger.isInfoEnabled()) { - logger.info("registered dispatcher task: channel='" + - channel.getName() + "' handler='" + handler + "'"); - } - } - - private void addDispatcherTask(DispatcherTask dispatcherTask) { - this.dispatcherTasks.add(dispatcherTask); - if (this.isRunning()) { - scheduleDispatcherTask(dispatcherTask); - if (this.logger.isInfoEnabled()) { - logger.info("scheduled dispatcher task"); - } - } - } - - 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() { @@ -312,22 +326,33 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } public void start() { - if (this.dispatcherExecutor == null) { + if (!this.initialized) { this.initialize(); } + if (this.isRunning() || this.starting) { + return; + } + this.starting = true; synchronized (this.lifecycleMonitor) { - if (!this.isRunning()) { - this.running = true; - for (MessageDispatcher dispatcher : this.dispatchers) { - dispatcher.start(); - if (logger.isInfoEnabled()) { - logger.info("started dispatcher '" + dispatcher + "'"); - } - } - for (DispatcherTask task : this.dispatcherTasks) { - scheduleDispatcherTask(task); + this.taskScheduler.start(); + this.running = true; + for (MessageDispatcher dispatcher : this.dispatchers.values()) { + dispatcher.start(); + if (logger.isInfoEnabled()) { + logger.info("started dispatcher '" + dispatcher + "'"); } } + for (Lifecycle adapter : this.lifecycleSourceAdapters) { + adapter.start(); + if (logger.isInfoEnabled()) { + logger.info("started source adapter '" + adapter + "'"); + } + } + } + this.running = true; + this.starting = false; + if (logger.isInfoEnabled()) { + logger.info("message bus started"); } } @@ -335,8 +360,14 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (this.isRunning()) { this.running = false; - this.dispatcherExecutor.shutdownNow(); - for (MessageDispatcher dispatcher : this.dispatchers) { + this.taskScheduler.stop(); + for (Lifecycle adapter : this.lifecycleSourceAdapters) { + adapter.stop(); + if (logger.isInfoEnabled()) { + logger.info("stopped source adapter '" + adapter + "'"); + } + } + for (MessageDispatcher dispatcher : this.dispatchers.values()) { dispatcher.stop(); if (logger.isInfoEnabled()) { logger.info("stopped dispatcher '" + dispatcher + "'"); @@ -346,59 +377,4 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - private void handleDispatchError(DispatcherTask task, Throwable t) { - try { - this.getInvalidMessageChannel().send(new ErrorMessage(t), 1000); - } - catch (Throwable ignore) { // message will be logged only - } - if (logger.isWarnEnabled()) { - logger.warn("failure occurred while dispatching message", t); - } - } - - - private class DispatcherTask implements Runnable { - - private MessageDispatcher dispatcher; - - private ConsumerPolicy policy; - - - public DispatcherTask(MessageDispatcher dispatcher, ConsumerPolicy policy) { - this.dispatcher = dispatcher; - this.policy = policy; - } - - - public ConsumerPolicy getPolicy() { - return this.policy; - } - - public void run() { - try { - dispatcher.dispatch(); - } - catch (Throwable t) { - handleDispatchError(this, t); - } - } - - } - - - private class RepeatingDispatcherTask implements Runnable { - - private DispatcherTask task; - - RepeatingDispatcherTask(DispatcherTask task) { - this.task = task; - } - - public void run() { - task.run(); - dispatcherExecutor.execute(new RepeatingDispatcherTask(task)); - } - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/PooledMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/PooledMessageHandler.java index d15d22fe7a..f93c2042be 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/PooledMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/PooledMessageHandler.java @@ -27,7 +27,9 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; -import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.dispatcher.MessageHandlerNotRunningException; +import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; +import org.springframework.integration.dispatcher.MessageSelectorRejectedException; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; import org.springframework.integration.message.selector.MessageSelector; @@ -112,7 +114,7 @@ public class PooledMessageHandler implements MessageHandler, Lifecycle { } } - public Message handle(Message message) { + public Message handle(Message message) { if (!this.isRunning()) { throw new MessageHandlerNotRunningException(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java index 9904a0ae06..15b6f7f0b1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/Subscription.java @@ -16,6 +16,9 @@ package org.springframework.integration.bus; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.scheduling.Schedule; + /** * Configuration metadata for activating a subscription. * @@ -25,9 +28,11 @@ public class Subscription { private String channel; - private String receiver; + private String handler; - private ConsumerPolicy policy = new ConsumerPolicy(); + private Schedule schedule; + + private ConcurrencyPolicy concurrencyPolicy; public String getChannel() { @@ -38,20 +43,28 @@ public class Subscription { this.channel = channel; } - public String getReceiver() { - return this.receiver; + public String getHandler() { + return this.handler; } - public void setReceiver(String receiver) { - this.receiver = receiver; + public void setHandler(String handler) { + this.handler = handler; } - public ConsumerPolicy getPolicy() { - return this.policy; + public Schedule getSchedule() { + return this.schedule; } - public void setPolicy(ConsumerPolicy policy) { - this.policy = policy; + public void setSchedule(Schedule schedule) { + this.schedule = schedule; + } + + public ConcurrencyPolicy getConcurrencyPolicy() { + return this.concurrencyPolicy; + } + + public void setConcurrencyPolicy(ConcurrencyPolicy concurrencyPolicy) { + this.concurrencyPolicy = concurrencyPolicy; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index a04077145f..a8550614da 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -31,10 +31,11 @@ import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.DefaultMessageHandlerAdapter; import org.springframework.integration.handler.MessageHandlerChain; +import org.springframework.integration.scheduling.PollingSchedule; import org.springframework.util.StringUtils; /** @@ -74,15 +75,19 @@ public class EndpointParser implements BeanDefinitionParser { private static final String PERIOD_ATTRIBUTE = "period"; - private static final String PERIOD_PROPERTY = "period"; + private static final String SCHEDULE_PROPERTY = "schedule"; - private static final String CONSUMER_ELEMENT = "consumer"; + private static final String CONCURRENCY_ELEMENT = "concurrency"; - private static final String CONSUMER_POLICY_PROPERTY = "consumerPolicy"; + private static final String CORE_CONCURRENCY_ATTRIBUTE = "core"; + + private static final String MAX_CONCURRENCY_ATTRIBUTE = "max"; + + private static final String CONCURRENCY_POLICY_PROPERTY = "concurrencyPolicy"; public BeanDefinition parse(Element element, ParserContext parserContext) { - RootBeanDefinition endpointDef = new RootBeanDefinition(GenericMessageEndpoint.class); + RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class); endpointDef.setSource(parserContext.extractSource(element)); String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE); if (StringUtils.hasText(inputChannel)) { @@ -98,10 +103,8 @@ public class EndpointParser implements BeanDefinitionParser { Node child = childNodes.item(i); if (child.getNodeType() == Node.ELEMENT_NODE) { String localName = child.getLocalName(); - if (CONSUMER_ELEMENT.equals(localName)) { - String consumerBeanName = parseConsumer((Element) child, parserContext); - endpointDef.getPropertyValues().addPropertyValue( - CONSUMER_POLICY_PROPERTY, new RuntimeBeanReference(consumerBeanName)); + if (CONCURRENCY_ELEMENT.equals(localName)) { + parseConcurrencyPolicy((Element) child, endpointDef); } else if (HANDLER_ELEMENT.equals(localName)) { String ref = ((Element) child).getAttribute(REF_ATTRIBUTE); @@ -150,15 +153,26 @@ public class EndpointParser implements BeanDefinitionParser { return endpointDef; } - private String parseConsumer(Element element, ParserContext parserContext) { - RootBeanDefinition consumerDef = new RootBeanDefinition(ConsumerPolicy.class); - String period = element.getAttribute(PERIOD_ATTRIBUTE); - if (StringUtils.hasText(period)) { - consumerDef.getPropertyValues().addPropertyValue(PERIOD_PROPERTY, Integer.parseInt(period)); + private void parseConcurrencyPolicy(Element concurrencyElement, RootBeanDefinition endpointDefinition) { + ConcurrencyPolicy policy = new ConcurrencyPolicy(); + String coreConcurrency = concurrencyElement.getAttribute(CORE_CONCURRENCY_ATTRIBUTE); + String maxConcurrency = concurrencyElement.getAttribute(MAX_CONCURRENCY_ATTRIBUTE); + if (StringUtils.hasText(coreConcurrency)) { + policy.setCoreConcurrency(Integer.parseInt(coreConcurrency)); } - String beanName = parserContext.getReaderContext().generateBeanName(consumerDef); - parserContext.registerBeanComponent(new BeanComponentDefinition(consumerDef, beanName)); - return beanName; + if (StringUtils.hasText(maxConcurrency)) { + policy.setMaxConcurrency(Integer.parseInt(maxConcurrency)); + } + endpointDefinition.getPropertyValues().addPropertyValue(CONCURRENCY_POLICY_PROPERTY, policy); + } + + private void parseSchedule(Element scheduleElement, RootBeanDefinition endpointDefinition) { + PollingSchedule schedule = new PollingSchedule(5); + String period = scheduleElement.getAttribute(PERIOD_ATTRIBUTE); + if (StringUtils.hasText(period)) { + schedule.setPeriod(Integer.parseInt(period)); + } + endpointDefinition.getPropertyValues().addPropertyValue(SCHEDULE_PROPERTY, schedule); } private String parseHandlerAdapter(String handlerRef, String handlerMethod, ParserContext parserContext) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 678e8dcc2e..c2408fc877 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -43,18 +43,20 @@ import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.Polled; import org.springframework.integration.annotation.Router; import org.springframework.integration.annotation.Splitter; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.handler.config.DefaultMessageHandlerCreator; import org.springframework.integration.handler.config.MessageHandlerCreator; import org.springframework.integration.handler.config.RouterMessageHandlerCreator; import org.springframework.integration.handler.config.SplitterMessageHandlerCreator; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; @@ -103,7 +105,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor if (endpointAnnotation == null) { return bean; } - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); this.configureInputChannel(bean, beanName, endpointAnnotation, endpoint); this.configureDefaultOutputChannel(bean, beanName, endpointAnnotation, endpoint); MessageHandlerChain handlerChain = this.createHandlerChain(bean); @@ -115,13 +117,12 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } private void configureInputChannel(final Object bean, final String beanName, - MessageEndpoint annotation, final GenericMessageEndpoint endpoint) { + MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) { String channelName = annotation.input(); if (StringUtils.hasText(channelName)) { endpoint.setInputChannelName(channelName); - ConsumerPolicy consumerPolicy = new ConsumerPolicy(); - consumerPolicy.setPeriod(annotation.pollPeriod()); - endpoint.setConsumerPolicy(consumerPolicy); + Schedule schedule = new PollingSchedule(annotation.pollPeriod()); + endpoint.setSchedule(schedule); return; } ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() { @@ -140,11 +141,13 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor messageBus.registerChannel(channelName, channel); messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter); endpoint.setInputChannelName(channelName); - endpoint.getConsumerPolicy().setPeriod(period); + Schedule schedule = new PollingSchedule(period); + endpoint.setSchedule(schedule); if (period > 0) { - endpoint.getConsumerPolicy().setConcurrency(1); - endpoint.getConsumerPolicy().setMaxConcurrency(1); - endpoint.getConsumerPolicy().setMaxMessagesPerTask(1); + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); + concurrencyPolicy.setCoreConcurrency(1); + concurrencyPolicy.setMaxConcurrency(1); + endpoint.setConcurrencyPolicy(concurrencyPolicy); } return; } @@ -153,7 +156,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor } private void configureDefaultOutputChannel(final Object bean, final String beanName, - final MessageEndpoint annotation, final GenericMessageEndpoint endpoint) { + final MessageEndpoint annotation, final DefaultMessageEndpoint endpoint) { String channelName = annotation.defaultOutput(); if (StringUtils.hasText(channelName)) { endpoint.setDefaultOutputChannelName(channelName); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java index a0a5aaa64e..69cb8fc20d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java @@ -28,7 +28,7 @@ import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.annotation.Subscriber; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.handler.DefaultMessageHandlerAdapter; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -85,7 +85,7 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { adapter.setMethodName(method.getName()); adapter.setObject(bean); adapter.afterPropertiesSet(); - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); endpoint.setInputChannelName(channelName); endpoint.setChannelRegistry(messageBus); endpoint.setHandler(adapter); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 3a8b77ef6d..97900b37e6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -99,7 +99,8 @@ - + + @@ -111,17 +112,29 @@ - + - Defines a consumer policy. + Defines a schedule. + + + + + Defines a concurrency policy. + + + + + + + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java similarity index 57% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java index 2dc262b91c..844f1c0635 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/ChannelPollingMessageRetriever.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; import java.util.Collection; import java.util.LinkedList; @@ -22,14 +22,14 @@ import java.util.List; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; +import org.springframework.util.Assert; /** * Message retriever that polls a {@link MessageChannel}. The number of * messages retrieved per poll is limited by the 'maxMessagesPerTask' - * property of the provided {@link ConsumerPolicy}, and the timeout for each - * receive call is determined by the policy's 'receiveTimeout' + * property, and the timeout for each receive call is determined by the 'receiveTimeout' * property. In general, it is recommended to use a value of 1 for - * 'maxMessagesPerTask' whenever a non-zero timeout is provided. Otherwise the + * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise the * retriever may be holding on to available messages while waiting for * additional messages. * @@ -39,19 +39,40 @@ public class ChannelPollingMessageRetriever implements MessageRetriever { private MessageChannel channel; - private ConsumerPolicy policy; + private int maxMessagesPerTask = DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK; + + private long receiveTimeout = DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT; - public ChannelPollingMessageRetriever(MessageChannel channel, ConsumerPolicy policy) { + public ChannelPollingMessageRetriever(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); this.channel = channel; - this.policy = policy; } + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least 1"); + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public MessageChannel getChannel() { + return this.channel; + } + public Collection> retrieveMessages() { List> messages = new LinkedList>(); - while (messages.size() < this.policy.getMaxMessagesPerTask()) { - Message message = this.channel.receive(this.policy.getReceiveTimeout()); + while (messages.size() < this.maxMessagesPerTask) { + Message message = null; + if (this.receiveTimeout < 0) { + message = this.channel.receive(); + } + else { + message = this.channel.receive(this.receiveTimeout); + } if (message == null) { return messages; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java new file mode 100644 index 0000000000..b01c76e797 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java @@ -0,0 +1,206 @@ +/* + * 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.dispatcher; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ScheduledFuture; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.context.Lifecycle; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.scheduling.MessagingTaskScheduler; +import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; +import org.springframework.util.Assert; + +/** + * Default implementation of {@link MessageDispatcher}. For a non-broadcasting + * {@link MessageChannel} (point-to-point), each handler can be registered with + * a {@link Schedule}. If the channel is broadcasting (publish-subscribe), the + * handlers will all be scheduled together according to the dispatcher's + * {@link #defaultSchedule}. + * + * @author Mark Fisher + */ +public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTaskSchedulerAware { + + protected Log logger = LogFactory.getLog(this.getClass()); + + private MessageChannel channel; + + private int maxMessagesPerTask = DispatcherPolicy.DEFAULT_MAX_MESSAGES_PER_TASK; + + private long receiveTimeout = DispatcherPolicy.DEFAULT_RECEIVE_TIMEOUT; + + private int rejectionLimit = DispatcherPolicy.DEFAULT_REJECTION_LIMIT; + + private long retryInterval = DispatcherPolicy.DEFAULT_RETRY_INTERVAL; + + private boolean shouldFailOnRejectionLimit = true; + + private MessagingTaskScheduler scheduler; + + private Schedule defaultSchedule = new PollingSchedule(5); + + private Map> scheduledHandlers = new ConcurrentHashMap>(); + + private List> futures = new CopyOnWriteArrayList>(); + + private volatile boolean running; + + private Object lifecycleMonitor = new Object(); + + + public DefaultMessageDispatcher(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); + this.channel = channel; + } + + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be at least 1"); + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public void setRejectionLimit(int rejectionLimit) { + Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1"); + this.rejectionLimit = rejectionLimit; + } + + public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) { + this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit; + } + + public void setRetryInterval(long retryInterval) { + Assert.isTrue(retryInterval >= 0, "'retryInterval' must not be negative"); + this.retryInterval = retryInterval; + } + + public void setMessagingTaskScheduler(MessagingTaskScheduler scheduler) { + Assert.notNull(scheduler, "'scheduler' must not be null"); + this.scheduler = scheduler; + } + + public void setDefaultSchedule(Schedule defaultSchedule) { + Assert.notNull(defaultSchedule, "'defaultSchedule' must not be null"); + this.defaultSchedule = defaultSchedule; + } + + public void addHandler(MessageHandler handler) { + this.addHandler(handler, null); + } + + public void addHandler(MessageHandler handler, Schedule schedule) { + Assert.notNull(handler, "'handler' must not be null"); + if (schedule == null) { + schedule = this.defaultSchedule; + } + else if (this.channel.isBroadcaster()) { + if (logger.isInfoEnabled()) { + logger.info("This dispatcher's channel is a broadcaster, and therefore all handlers are " + + "scheduled with its 'defaultSchedule'. The provided schedule will be ignored."); + } + schedule = this.defaultSchedule; + } + if (this.isRunning() && handler instanceof Lifecycle) { + ((Lifecycle) handler).start(); + } + if (this.scheduledHandlers.containsKey(schedule)) { + this.scheduledHandlers.get(schedule).add(handler); + } + else { + List handlerList = new CopyOnWriteArrayList(); + handlerList.add(handler); + this.scheduledHandlers.put(schedule, handlerList); + } + } + + public boolean isRunning() { + return this.running; + } + + public void start() { + if (this.scheduler == null) { + if (logger.isInfoEnabled()) { + logger.info("no scheduler was provided, will create one"); + } + this.scheduler = new SimpleMessagingTaskScheduler(); + } + if (!this.scheduler.isRunning()) { + this.scheduler.start(); + } + synchronized (this.lifecycleMonitor) { + if (!this.isRunning()) { + for (Map.Entry> entry : this.scheduledHandlers.entrySet()) { + Schedule schedule = entry.getKey(); + List handlers = entry.getValue(); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + retriever.setMaxMessagesPerTask(this.maxMessagesPerTask); + retriever.setReceiveTimeout(this.receiveTimeout); + DispatcherTask task = new DispatcherTask(retriever); + task.setSchedule(schedule); + task.setRejectionLimit(this.rejectionLimit); + task.setRetryInterval(this.retryInterval); + task.setBroadcast(channel.isBroadcaster()); + task.setShouldFailOnRejectionLimit(this.shouldFailOnRejectionLimit); + for (MessageHandler handler : handlers) { + if (handler instanceof Lifecycle) { + ((Lifecycle) handler).start(); + } + task.addHandler(handler); + } + ScheduledFuture future = this.scheduler.schedule(task); + if (future != null) { + futures.add(future); + } + } + this.running = true; + } + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.isRunning()) { + for (ScheduledFuture future : this.futures) { + future.cancel(true); + for (List handlerList : scheduledHandlers.values()) { + for (MessageHandler handler : handlerList) { + if (handler instanceof Lifecycle) { + ((Lifecycle) handler).stop(); + } + } + } + } + this.running = false; + } + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java new file mode 100644 index 0000000000..24d455720c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherPolicy.java @@ -0,0 +1,82 @@ +/* + * 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.dispatcher; + +import org.springframework.integration.endpoint.EndpointPolicy; +import org.springframework.util.Assert; + +/** + * Metadata for a {@link MessageDispatcher}. + * + * @author Mark Fisher + */ +public class DispatcherPolicy implements EndpointPolicy { + + public final static int DEFAULT_MAX_MESSAGES_PER_TASK = 1; + + public final static long DEFAULT_RECEIVE_TIMEOUT = 1000; + + public final static int DEFAULT_REJECTION_LIMIT = 5; + + public final static long DEFAULT_RETRY_INTERVAL = 1000; + + + private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK; + + private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + + private int rejectionLimit = DEFAULT_REJECTION_LIMIT; + + private long retryInterval = DEFAULT_RETRY_INTERVAL; + + + public int getMaxMessagesPerTask() { + return this.maxMessagesPerTask; + } + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagePerTask' must be at least 1"); + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public long getReceiveTimeout() { + return this.receiveTimeout; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public int getRejectionLimit() { + return this.rejectionLimit; + } + + public void setRejectionLimit(int rejectionLimit) { + Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1"); + this.rejectionLimit = rejectionLimit; + } + + public long getRetryInterval() { + return this.retryInterval; + } + + public void setRetryInterval(long retryInterval) { + Assert.isTrue(retryInterval >= 0, "'retryInterval' must not be negative"); + this.retryInterval = retryInterval; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/BasePollingMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java similarity index 65% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/BasePollingMessageDispatcher.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java index 191b6528d0..011fc265a7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/BasePollingMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java @@ -14,39 +14,65 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; import java.util.ArrayList; +import java.util.Collection; import java.util.Iterator; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; +import org.springframework.integration.scheduling.MessagingTask; +import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; /** - * The base implementation of a polling {@link MessageDispatcher}. If + * A task for polling {@link MessageDispatcher MessageDispatchers}. If * {@link #broadcast} is set to false (the default), each message * will be sent to a single {@link MessageHandler}. Otherwise, each - * retrieved {@link Message} will be sent to all handlers. + * retrieved {@link Message} will be sent to all of the handlers. * * @author Mark Fisher */ -public class BasePollingMessageDispatcher extends AbstractMessageDispatcher { +public class DispatcherTask implements MessagingTask { + + private Log logger = LogFactory.getLog(this.getClass()); private boolean broadcast = false; + private Schedule schedule; + private int rejectionLimit = 5; private long retryInterval = 1000; private boolean shouldFailOnRejectionLimit = true; + private MessageRetriever retriever; - public BasePollingMessageDispatcher(MessageRetriever retriever) { - super(retriever); + private List handlers = new CopyOnWriteArrayList(); + + + public DispatcherTask(MessageChannel channel) { + Assert.notNull(channel, "'channel' must not be null"); + this.retriever = new ChannelPollingMessageRetriever(channel); + this.broadcast = channel.isBroadcaster(); + } + + public DispatcherTask(MessageRetriever retriever) { + Assert.notNull(retriever, "'retriever' must not be null"); + if (retriever instanceof ChannelPollingMessageRetriever) { + this.broadcast = ((ChannelPollingMessageRetriever) retriever).getChannel().isBroadcaster(); + } + this.retriever = retriever; } @@ -54,6 +80,15 @@ public class BasePollingMessageDispatcher extends AbstractMessageDispatcher { this.broadcast = broadcast; } + public void setSchedule(Schedule schedule) { + Assert.notNull(schedule, "'schedule' must not be null"); + this.schedule = schedule; + } + + public Schedule getSchedule() { + return this.schedule; + } + public void setRejectionLimit(int rejectionLimit) { Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1"); this.rejectionLimit = rejectionLimit; @@ -72,10 +107,33 @@ public class BasePollingMessageDispatcher extends AbstractMessageDispatcher { this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit; } - @Override + public void addHandler(MessageHandler handler) { + Assert.notNull(handler, "'handler' must not be null"); + this.handlers.add(handler); + } + + /** + * Retrieves messages and dispatches to the executors. + * + * @return the number of messages processed + */ + public int dispatch() { + int messagesProcessed = 0; + Collection> messages = this.retriever.retrieveMessages(); + if (messages == null) { + return 0; + } + for (Message message : messages) { + if (dispatchMessage(message)) { + messagesProcessed++; + } + } + return messagesProcessed; + } + protected boolean dispatchMessage(Message message) { int attempts = 0; - List targets = new ArrayList(this.getHandlers()); + List targets = new ArrayList(this.handlers); while (attempts < this.rejectionLimit) { if (attempts > 0) { if (logger.isDebugEnabled()) { @@ -137,4 +195,8 @@ public class BasePollingMessageDispatcher extends AbstractMessageDispatcher { return false; } + public void run() { + this.dispatch(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java similarity index 54% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageDispatcher.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index 387b9c65ac..154df946f1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -14,23 +14,21 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; -import org.springframework.integration.channel.MessageChannel; +import org.springframework.context.Lifecycle; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.scheduling.Schedule; /** - * A {@link MessageDispatcher} that polls a {@link MessageChannel}. + * Strategy interface for dispatching messages. * * @author Mark Fisher */ -public class ChannelPollingMessageDispatcher extends BasePollingMessageDispatcher { +public interface MessageDispatcher extends Lifecycle { - public ChannelPollingMessageDispatcher(MessageChannel channel, int period) { - this(channel, ConsumerPolicy.newPollingPolicy(period)); - } + void addHandler(MessageHandler handler); - public ChannelPollingMessageDispatcher(MessageChannel channel, ConsumerPolicy policy) { - super(new ChannelPollingMessageRetriever(channel, policy)); - } + void addHandler(MessageHandler handler, Schedule schedule); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageHandlerNotRunningException.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerNotRunningException.java similarity index 94% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/MessageHandlerNotRunningException.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerNotRunningException.java index 2406f4b4ce..0d63fa7fed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageHandlerNotRunningException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerNotRunningException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; import org.springframework.integration.MessageHandlingException; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageHandlerRejectedExecutionException.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerRejectedExecutionException.java similarity index 90% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/MessageHandlerRejectedExecutionException.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerRejectedExecutionException.java index b2e98ca635..07c3b76d7e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageHandlerRejectedExecutionException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageHandlerRejectedExecutionException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; import org.springframework.integration.MessageHandlingException; @@ -26,6 +26,10 @@ import org.springframework.integration.MessageHandlingException; */ public class MessageHandlerRejectedExecutionException extends MessageHandlingException { + public MessageHandlerRejectedExecutionException() { + super(); + } + public MessageHandlerRejectedExecutionException(Throwable cause) { super("handler rejected execution", cause); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java similarity index 94% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/MessageRetriever.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java index f728cdd3a9..ee1cdaeeed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageRetriever.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageRetriever.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; import java.util.Collection; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageSelectorRejectedException.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java similarity index 95% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/MessageSelectorRejectedException.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java index 77e4dccbb2..55b7cac638 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageSelectorRejectedException.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageSelectorRejectedException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.bus; +package org.springframework.integration.dispatcher; import org.springframework.integration.MessageHandlingException; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java new file mode 100644 index 0000000000..b24e7d985c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrencyPolicy.java @@ -0,0 +1,51 @@ +/* + * 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.endpoint; + +import org.springframework.util.Assert; + +/** + * Metadata for configuring a pool of concurrent threads. + * + * @author Mark Fisher + */ +public class ConcurrencyPolicy implements EndpointPolicy { + + private int coreConcurrency; + + private int maxConcurrency; + + + public int getCoreConcurrency() { + return this.coreConcurrency; + } + + public void setCoreConcurrency(int coreConcurrency) { + Assert.isTrue(coreConcurrency > 0, "'coreConcurrency' must be at least 1"); + this.coreConcurrency = coreConcurrency; + } + + public int getMaxConcurrency() { + return this.maxConcurrency; + } + + public void setMaxConcurrency(int maxConcurrency) { + Assert.isTrue(maxConcurrency > 0, "'maxConcurrency' must be at least 1"); + this.maxConcurrency = maxConcurrency; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java similarity index 81% rename from spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index aec445f8bb..0b428e1390 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -19,26 +19,18 @@ package org.springframework.integration.endpoint; import org.springframework.beans.factory.BeanNameAware; import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessagingConfigurationException; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; +import org.springframework.integration.scheduling.Schedule; /** - * A generic endpoint implementation designed to accommodate a variety of - * strategies including: - *
    - *
  • source channel-adapter: source adapter + target channel
    • - *
  • target channel-adapter: source channel + target adapter
    • - *
  • one-way: source + handler that returns null and no target
    • - *
  • request-reply: source + handler and either a reply channel - * specified on the request message or a default target on the endpoint - *
+ * Default implementation of the {@link MessageEndpoint} interface. * * @author Mark Fisher */ -public class GenericMessageEndpoint implements MessageEndpoint, BeanNameAware { +public class DefaultMessageEndpoint implements MessageEndpoint, BeanNameAware { private String name; @@ -48,9 +40,11 @@ public class GenericMessageEndpoint implements MessageEndpoint, BeanNameAware { private MessageHandler handler; - private ChannelRegistry channelRegistry; + private Schedule schedule; - private ConsumerPolicy consumerPolicy = new ConsumerPolicy(); + private ConcurrencyPolicy concurrencyPolicy; + + private ChannelRegistry channelRegistry; public String getName() { @@ -79,14 +73,6 @@ public class GenericMessageEndpoint implements MessageEndpoint, BeanNameAware { return this.inputChannelName; } - public void setConsumerPolicy(ConsumerPolicy consumerPolicy) { - this.consumerPolicy = consumerPolicy; - } - - public ConsumerPolicy getConsumerPolicy() { - return this.consumerPolicy; - } - /** * Set the name of the channel to which this endpoint can send reply messages by default. */ @@ -105,6 +91,22 @@ public class GenericMessageEndpoint implements MessageEndpoint, BeanNameAware { this.handler = handler; } + public Schedule getSchedule() { + return this.schedule; + } + + public void setSchedule(Schedule schedule) { + this.schedule = schedule; + } + + public ConcurrencyPolicy getConcurrencyPolicy() { + return this.concurrencyPolicy; + } + + public void setConcurrencyPolicy(ConcurrencyPolicy concurrencyPolicy) { + this.concurrencyPolicy = concurrencyPolicy; + } + /** * Set the channel registry to use for looking up channels by name. */ @@ -112,7 +114,6 @@ public class GenericMessageEndpoint implements MessageEndpoint, BeanNameAware { this.channelRegistry = channelRegistry; } - public Message handle(Message message) { if (this.handler == null) { if (this.defaultOutputChannelName == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointPolicy.java similarity index 75% rename from spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java rename to spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointPolicy.java index d5e6656471..815b6ca272 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/EndpointPolicy.java @@ -14,17 +14,13 @@ * limitations under the License. */ -package org.springframework.integration.bus; - -import org.springframework.context.Lifecycle; +package org.springframework.integration.endpoint; /** - * Strategy interface for dispatching messages. + * A marker interface for endpoint metadata. * * @author Mark Fisher */ -public interface MessageDispatcher extends Lifecycle { - - int dispatch(); +public interface EndpointPolicy { } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 09872edd5e..464c3fbe53 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -16,9 +16,9 @@ package org.springframework.integration.endpoint; -import org.springframework.integration.bus.ConsumerPolicy; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.scheduling.Schedule; /** * Base interface for message endpoints. @@ -37,8 +37,10 @@ public interface MessageEndpoint extends MessageHandler { String getDefaultOutputChannelName(); - ConsumerPolicy getConsumerPolicy(); - void setChannelRegistry(ChannelRegistry channelRegistry); + Schedule getSchedule(); + + ConcurrencyPolicy getConcurrencyPolicy(); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java index 4dc81cb3e1..d8a2898517 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/AbstractMessagingTaskScheduler.java @@ -33,9 +33,9 @@ public abstract class AbstractMessagingTaskScheduler implements MessagingTaskSch * Submit a task to be run once. */ public void execute(Runnable task) { - this.schedule(new DefaultMessagingTask(task)); + this.schedule(task); } - public abstract ScheduledFuture schedule(MessagingTask task); + public abstract ScheduledFuture schedule(Runnable task); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java deleted file mode 100644 index e3a8ab29bd..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/DefaultMessagingTask.java +++ /dev/null @@ -1,61 +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.scheduling; - -/** - * A wrapper for {@link Runnable Runnables} that provides a schedule and also - * captures any error that may be thrown from the run() method. - * - * @author Mark Fisher - */ -public class DefaultMessagingTask implements MessagingTask { - - private Runnable runnable; - - private Schedule schedule; - - private Throwable lastError; - - - public DefaultMessagingTask(Runnable runnable) { - this(runnable, null); - } - - public DefaultMessagingTask(Runnable runnable, Schedule schedule) { - this.runnable = runnable; - this.schedule = schedule; - } - - - public Schedule getSchedule() { - return this.schedule; - } - - public Throwable getLastError() { - return this.lastError; - } - - public void run() { - try { - this.runnable.run(); - } - catch (Throwable t) { - this.lastError = t; - } - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java new file mode 100644 index 0000000000..ab54183a0a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagePublishingErrorHandler.java @@ -0,0 +1,68 @@ +/* + * 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.scheduling; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.util.ErrorHandler; + +/** + * {@link ErrorHandler} implementation that sends an {@link ErrorMessage} to a + * {@link MessageChannel}. + * + * @author Mark Fisher + */ +public class MessagePublishingErrorHandler implements ErrorHandler { + + private Log logger = LogFactory.getLog(this.getClass()); + + private MessageChannel errorChannel; + + private long sendTimeout = 1000; + + + public MessagePublishingErrorHandler() { + } + + public MessagePublishingErrorHandler(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + } + + + public void setErrorChannel(MessageChannel errorChannel) { + this.errorChannel = errorChannel; + } + + public void handle(Throwable t) { + if (this.errorChannel != null) { + try { + if(this.errorChannel.send(new ErrorMessage(t), this.sendTimeout)) { + return; + } + } + catch (Throwable ignore) { // message will be logged only + } + } + if (logger.isWarnEnabled()) { + logger.warn("failure occurred in messaging task", t); + } + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java index 6a132571bc..94314f2d40 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTask.java @@ -26,8 +26,4 @@ public interface MessagingTask extends Runnable { Schedule getSchedule(); - Throwable getLastError(); - - void run(); - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java index 8d3446bb5a..e611a5c70e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java @@ -18,6 +18,7 @@ package org.springframework.integration.scheduling; import java.util.concurrent.ScheduledFuture; +import org.springframework.context.Lifecycle; import org.springframework.scheduling.SchedulingTaskExecutor; /** @@ -25,8 +26,8 @@ import org.springframework.scheduling.SchedulingTaskExecutor; * * @author Mark Fisher */ -public interface MessagingTaskScheduler extends SchedulingTaskExecutor { +public interface MessagingTaskScheduler extends SchedulingTaskExecutor, Lifecycle { - ScheduledFuture schedule(MessagingTask task); + ScheduledFuture schedule(Runnable task); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java new file mode 100644 index 0000000000..092308eb5c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/MessagingTaskSchedulerAware.java @@ -0,0 +1,29 @@ +/* + * 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.scheduling; + +/** + * Callback interface for components that require the + * {@link MessagingTaskScheduler}. + * + * @author Mark Fisher + */ +public interface MessagingTaskSchedulerAware { + + void setMessagingTaskScheduler(MessagingTaskScheduler scheduler); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java index 5975092234..a8a8404397 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java @@ -16,12 +16,17 @@ package org.springframework.integration.scheduling; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.util.ErrorHandler; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; import org.springframework.util.Assert; /** @@ -36,6 +41,18 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler private int corePoolSize = 10; + private ThreadFactory threadFactory; + + private String threadNamePrefix = this.getClass().getSimpleName() + "-"; + + private ErrorHandler errorHandler; + + private Set pendingTasks = new CopyOnWriteArraySet(); + + private volatile boolean running; + + private Object lifecycleMonitor = new Object(); + public void setExecutor(ScheduledExecutorService executor) { Assert.notNull(executor, "'executor' must not be null"); @@ -47,47 +64,108 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler this.corePoolSize = corePoolSize; } + public void setThreadFactory(ThreadFactory threadFactory) { + this.threadFactory = threadFactory; + } + + public void setThreadNamePrefix(String threadNamePrefix) { + Assert.notNull(threadNamePrefix, "'threadNamePrefix' must not be null"); + this.threadNamePrefix = threadNamePrefix; + } + + public void setErrorHandler(ErrorHandler errorHandler) { + this.errorHandler = errorHandler; + } + public void afterPropertiesSet() { if (this.executor == null) { - this.executor = new ScheduledThreadPoolExecutor(this.corePoolSize); + if (this.threadFactory == null) { + this.threadFactory = new CustomizableThreadFactory(this.threadNamePrefix); + } + this.executor = new ScheduledThreadPoolExecutor(this.corePoolSize, this.threadFactory); + } + } + + public boolean isRunning() { + return this.running; + } + + public void start() { + if (this.executor == null) { + this.afterPropertiesSet(); + } + synchronized (this.lifecycleMonitor) { + this.running = true; + for (Runnable task : this.pendingTasks) { + this.schedule(task); + } + this.pendingTasks.clear(); + } + } + + public void stop() { + if (this.isRunning()) { + this.running = false; + this.executor.shutdownNow(); } } @Override - public ScheduledFuture schedule(MessagingTask task) { - if (this.executor == null) { - this.afterPropertiesSet(); + public ScheduledFuture schedule(Runnable task) { + if (!this.isRunning()) { + this.pendingTasks.add(task); + return null; } - Schedule schedule = task.getSchedule(); + Schedule schedule = (task instanceof MessagingTask) ? ((MessagingTask) task).getSchedule() : null; + MessagingTaskRunner runner = new MessagingTaskRunner(task); if (schedule == null) { - return this.executor.schedule(task, 0, TimeUnit.MILLISECONDS); + return this.executor.schedule(runner, 0, TimeUnit.MILLISECONDS); } if (schedule instanceof PollingSchedule) { PollingSchedule ps = (PollingSchedule) schedule; if (ps.getPeriod() <= 0) { - return this.executor.schedule(new RepeatingTask(task), ps.getInitialDelay(), ps.getTimeUnit()); + runner.setShouldRepeat(true); + return this.executor.schedule(runner, ps.getInitialDelay(), ps.getTimeUnit()); } if (ps.getFixedRate()) { - return this.executor.scheduleAtFixedRate(task, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); + return this.executor.scheduleAtFixedRate(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); } - return this.executor.scheduleWithFixedDelay(task, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); + return this.executor.scheduleWithFixedDelay(runner, ps.getInitialDelay(), ps.getPeriod(), ps.getTimeUnit()); } throw new UnsupportedOperationException(this.getClass().getName() + " does not support schedule type '" + schedule.getClass().getName() + "'"); } - private class RepeatingTask implements Runnable { + private class MessagingTaskRunner implements Runnable { - private MessagingTask task; + private Runnable task; - RepeatingTask(MessagingTask task) { + private boolean shouldRepeat; + + + public MessagingTaskRunner(Runnable task) { this.task = task; } + public void setShouldRepeat(boolean shouldRepeat) { + this.shouldRepeat = shouldRepeat; + } + public void run() { - task.run(); - executor.execute(new RepeatingTask(task)); + try { + this.task.run(); + } + catch (Throwable t) { + if (errorHandler != null) { + errorHandler.handle(t); + } + } + if (this.shouldRepeat) { + MessagingTaskRunner runner = new MessagingTaskRunner(this.task); + runner.setShouldRepeat(true); + executor.execute(runner); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ConsumerType.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandler.java similarity index 77% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/ConsumerType.java rename to spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandler.java index c0e1a3388b..36d26d402f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ConsumerType.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/ErrorHandler.java @@ -14,21 +14,15 @@ * limitations under the License. */ -package org.springframework.integration.channel; +package org.springframework.integration.util; /** - * Enumeration of the different types of message consumer. + * Strategy for handling a {@link Throwable}. * * @author Mark Fisher */ -public enum ConsumerType { +public interface ErrorHandler { - EVENT_DRIVEN, - - FIXED_RATE, - - FIXED_DELAY, - - SCHEDULED + void handle(Throwable t); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterTests.java index 904a1bcfa4..29118ef2dc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterTests.java @@ -20,6 +20,8 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.Test; @@ -35,14 +37,13 @@ public class AdapterTests { public void testAdaptersWithBeanDefinitions() throws IOException, InterruptedException { AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTests.xml", this.getClass()); TestSink sink = (TestSink) context.getBean("sink"); + CountDownLatch latch = new CountDownLatch(1); + sink.setLatch(latch); assertNull(sink.get()); context.start(); String result = null; - int attempts = 0; - while (result == null && attempts++ < 100) { - Thread.sleep(5); - result = sink.get(); - } + latch.await(3000, TimeUnit.MILLISECONDS); + result = sink.get(); assertNotNull(result); context.close(); } @@ -51,14 +52,13 @@ public class AdapterTests { public void testAdaptersWithNamespace() throws IOException, InterruptedException { AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTestsWithNamespace.xml", this.getClass()); TestSink sink = (TestSink) context.getBean("sink"); + CountDownLatch latch = new CountDownLatch(1); + sink.setLatch(latch); assertNull(sink.get()); context.start(); String result = null; - int attempts = 0; - while (result == null && attempts++ < 100) { - Thread.sleep(5); - result = sink.get(); - } + latch.await(3000, TimeUnit.MILLISECONDS); + result = sink.get(); assertNotNull(result); context.close(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index c3aad89bad..8ef5d141e8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -44,7 +44,6 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(100); adapter.start(); - adapter.dispatch(); Message message = channel.receive(); assertNotNull("message should not be null", message); assertEquals("testing.1", message.getPayload()); @@ -56,18 +55,20 @@ public class PollingSourceAdapterTests { SimpleChannel channel = new SimpleChannel(1); PollingSourceAdapter adapter = new PollingSourceAdapter(source); adapter.setChannel(channel); - adapter.setPeriod(500); + adapter.setInitialDelay(10000); adapter.setSendTimeout(10); adapter.start(); - adapter.dispatch(); - adapter.dispatch(); + adapter.processMessages(); + adapter.processMessages(); + adapter.stop(); Message message1 = channel.receive(); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); Message message2 = channel.receive(0); assertNull("second message should be null", message2); - adapter.dispatch(); - Message message3 = channel.receive(0); + adapter.start(); + adapter.processMessages(); + Message message3 = channel.receive(100); assertNotNull("third message should not be null", message3); assertEquals("testing.3", message3.getPayload()); } @@ -78,10 +79,10 @@ public class PollingSourceAdapterTests { SimpleChannel channel = new SimpleChannel(); PollingSourceAdapter adapter = new PollingSourceAdapter(source); adapter.setChannel(channel); - adapter.setPeriod(1000); + adapter.setInitialDelay(10000); adapter.setMaxMessagesPerTask(5); adapter.start(); - adapter.dispatch(); + adapter.processMessages(); Message message1 = channel.receive(0); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); @@ -104,7 +105,7 @@ public class PollingSourceAdapterTests { adapter.setPeriod(1000); adapter.setMaxMessagesPerTask(2); adapter.start(); - adapter.dispatch(); + adapter.processMessages(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java index e7953172d0..3836669a3e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java @@ -16,6 +16,8 @@ package org.springframework.integration.adapter; +import java.util.concurrent.CountDownLatch; + /** * @author Mark Fisher */ @@ -23,6 +25,13 @@ public class TestSink { private String result; + private CountDownLatch latch; + + + public void setLatch(CountDownLatch latch) { + this.latch = latch; + } + public void validMethod(String s) { } @@ -34,6 +43,9 @@ public class TestSink { } public void store(String s) { + if (this.latch != null) { + this.latch.countDown(); + } this.result = s; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml index 4294c1226b..78a1d4d15e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml @@ -34,14 +34,9 @@ - + - - - - - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java index 2db48434cc..0ada773d62 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java @@ -39,10 +39,10 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithJmsTemplate.xml", this.getClass()); context.start(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.dispatch(); - Message message = channel.receive(100); + adapter.processMessages(); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message message = channel.receive(500); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); context.stop(); @@ -53,10 +53,10 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithConnectionFactoryAndDestination.xml", this.getClass()); context.start(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.dispatch(); - Message message = channel.receive(100); + adapter.processMessages(); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message message = channel.receive(500); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); context.stop(); @@ -67,10 +67,10 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "pollingAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); context.start(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter"); - adapter.dispatch(); - Message message = channel.receive(100); + adapter.processMessages(); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message message = channel.receive(500); assertNotNull("message should not be null", message); assertEquals("polling-test", message.getPayload()); context.stop(); @@ -81,10 +81,10 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageDrivenAdapterWithConnectionFactoryAndDestination.xml", this.getClass()); context.start(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter"); assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass()); - Message message = channel.receive(100); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message message = channel.receive(3000); assertNotNull("message should not be null", message); assertEquals("message-driven-test", message.getPayload()); context.stop(); @@ -95,10 +95,10 @@ public class JmsSourceAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass()); context.start(); - MessageChannel channel = (MessageChannel) context.getBean("channel"); JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter"); assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass()); - Message message = channel.receive(100); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + Message message = channel.receive(3000); assertNotNull("message should not be null", message); assertEquals("message-driven-test", message.getPayload()); context.stop(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java index c5dbfb70c3..f72cd991c7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java @@ -40,7 +40,7 @@ public class ByteStreamSourceAdapterTests { ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); adapter.setChannel(channel); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); byte[] payload = (byte[]) message1.getPayload(); @@ -50,7 +50,7 @@ public class ByteStreamSourceAdapterTests { assertEquals(3, payload[2]); Message message2 = channel.receive(0); assertNull(message2); - adapter.dispatch(); + adapter.processMessages(); Message message3 = channel.receive(0); assertNull(message3); } @@ -65,7 +65,7 @@ public class ByteStreamSourceAdapterTests { adapter.setBytesPerMessage(8); adapter.setMaxMessagesPerTask(5); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); assertEquals(8, ((byte[]) message1.getPayload()).length); @@ -80,10 +80,11 @@ public class ByteStreamSourceAdapterTests { MessageChannel channel = new SimpleChannel(); ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); adapter.setBytesPerMessage(4); + adapter.setInitialDelay(10000); adapter.setMaxMessagesPerTask(1); adapter.setChannel(channel); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); byte[] bytes1 = (byte[]) message1.getPayload(); @@ -91,7 +92,7 @@ public class ByteStreamSourceAdapterTests { assertEquals(0, bytes1[0]); Message message2 = channel.receive(0); assertNull(message2); - adapter.dispatch(); + adapter.processMessages(); Message message3 = channel.receive(0); byte[] bytes3 = (byte[]) message3.getPayload(); assertEquals(4, bytes3.length); @@ -104,11 +105,12 @@ public class ByteStreamSourceAdapterTests { ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); + adapter.setInitialDelay(10000); adapter.setChannel(channel); adapter.setBytesPerMessage(4); adapter.setMaxMessagesPerTask(5); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(2, count); Message message1 = channel.receive(0); byte[] bytes1 = (byte[]) message1.getPayload(); @@ -128,17 +130,18 @@ public class ByteStreamSourceAdapterTests { ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); + adapter.setInitialDelay(10000); adapter.setBytesPerMessage(4); adapter.setMaxMessagesPerTask(1); adapter.setChannel(channel); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - adapter.dispatch(); + adapter.processMessages(); Message message3 = channel.receive(0); assertEquals(2, ((byte[]) message3.getPayload()).length); } @@ -149,18 +152,19 @@ public class ByteStreamSourceAdapterTests { ByteArrayInputStream stream = new ByteArrayInputStream(bytes); MessageChannel channel = new SimpleChannel(); ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream); + adapter.setInitialDelay(10000); adapter.setBytesPerMessage(4); adapter.setShouldTruncate(false); adapter.setMaxMessagesPerTask(1); adapter.setChannel(channel); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); assertEquals(4, ((byte[]) message1.getPayload()).length); Message message2 = channel.receive(0); assertNull(message2); - adapter.dispatch(); + adapter.processMessages(); Message message3 = channel.receive(0); assertEquals(4, ((byte[]) message3.getPayload()).length); assertEquals(0, ((byte[]) message3.getPayload())[3]); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java index d5fa7a836f..52d32815b5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java @@ -40,13 +40,13 @@ public class CharacterStreamSourceAdapterTests { CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream); adapter.setChannel(channel); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - adapter.dispatch(); + adapter.processMessages(); Message message3 = channel.receive(0); assertNull(message3); } @@ -60,7 +60,7 @@ public class CharacterStreamSourceAdapterTests { adapter.setChannel(channel); adapter.setMaxMessagesPerTask(5); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); assertEquals("test", message1.getPayload()); @@ -74,16 +74,17 @@ public class CharacterStreamSourceAdapterTests { ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes()); MessageChannel channel = new SimpleChannel(); CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream); + adapter.setInitialDelay(10000); adapter.setMaxMessagesPerTask(1); adapter.setChannel(channel); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(1, count); Message message1 = channel.receive(0); assertEquals("test1", message1.getPayload()); Message message2 = channel.receive(0); assertNull(message2); - adapter.dispatch(); + adapter.processMessages(); Message message3 = channel.receive(0); assertEquals("test2", message3.getPayload()); } @@ -97,7 +98,7 @@ public class CharacterStreamSourceAdapterTests { adapter.setChannel(channel); adapter.setMaxMessagesPerTask(5); adapter.start(); - int count = adapter.dispatch(); + int count = adapter.processMessages(); assertEquals(2, count); Message message1 = channel.receive(0); assertEquals("test1", message1.getPayload()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 2d291a851d..8741873be6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -22,10 +22,10 @@ import java.io.ByteArrayOutputStream; import org.junit.Test; -import org.springframework.integration.bus.ConsumerPolicy; -import org.springframework.integration.bus.ChannelPollingMessageDispatcher; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.ChannelPollingMessageRetriever; +import org.springframework.integration.dispatcher.DispatcherTask; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -40,12 +40,10 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + DispatcherTask dispatcherTask = new DispatcherTask(channel); + dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); - int count = dispatcher.dispatch(); + int count = dispatcherTask.dispatch(); assertEquals(1, count); String result = new String(stream.toByteArray()); assertEquals("foo", result); @@ -57,16 +55,14 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + DispatcherTask dispatcherTask = new DispatcherTask(channel); + dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); channel.send(new StringMessage("bar")); - assertEquals(1, dispatcher.dispatch()); + assertEquals(1, dispatcherTask.dispatch()); String result1 = new String(stream.toByteArray()); assertEquals("foo", result1); - assertEquals(1, dispatcher.dispatch()); + assertEquals(1, dispatcherTask.dispatch()); String result2 = new String(stream.toByteArray()); assertEquals("foobar", result2); } @@ -78,17 +74,15 @@ public class CharacterStreamTargetAdapterTests { CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); adapter.setShouldAppendNewLine(true); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + DispatcherTask dispatcherTask = new DispatcherTask(channel); + dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); channel.send(new StringMessage("bar")); - assertEquals(1, dispatcher.dispatch()); + assertEquals(1, dispatcherTask.dispatch()); String result1 = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine, result1); - assertEquals(1, dispatcher.dispatch()); + assertEquals(1, dispatcherTask.dispatch()); String result2 = new String(stream.toByteArray()); assertEquals("foo" + newLine + "bar" + newLine, result2); } @@ -99,14 +93,13 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - policy.setMaxMessagesPerTask(2); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + retriever.setMaxMessagesPerTask(2); + DispatcherTask dispatcherTask = new DispatcherTask(retriever); + dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); channel.send(new StringMessage("bar")); - assertEquals(2, dispatcher.dispatch()); + assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); assertEquals("foobar", result); } @@ -118,15 +111,14 @@ public class CharacterStreamTargetAdapterTests { CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); adapter.setShouldAppendNewLine(true); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - policy.setReceiveTimeout(0); - policy.setMaxMessagesPerTask(10); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + retriever.setReceiveTimeout(0); + retriever.setMaxMessagesPerTask(10); + DispatcherTask dispatcherTask = new DispatcherTask(retriever); + dispatcherTask.addHandler(adapter); channel.send(new StringMessage("foo")); channel.send(new StringMessage("bar")); - assertEquals(2, dispatcher.dispatch()); + assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, result); @@ -138,13 +130,11 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + DispatcherTask dispatcherTask = new DispatcherTask(channel); + dispatcherTask.addHandler(adapter); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); - int count = dispatcher.dispatch(); + int count = dispatcherTask.dispatch(); assertEquals(1, count); String result = new String(stream.toByteArray()); assertEquals("foo", result); @@ -156,17 +146,16 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - policy.setReceiveTimeout(0); - policy.setMaxMessagesPerTask(2); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + retriever.setReceiveTimeout(0); + retriever.setMaxMessagesPerTask(2); + DispatcherTask dispatcherTask = new DispatcherTask(retriever); + dispatcherTask.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1)); channel.send(new GenericMessage(testObject2)); - assertEquals(2, dispatcher.dispatch()); + assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); assertEquals("foobar", result); } @@ -178,17 +167,16 @@ public class CharacterStreamTargetAdapterTests { CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setChannel(channel); adapter.setShouldAppendNewLine(true); - ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy(); - policy.setReceiveTimeout(0); - policy.setMaxMessagesPerTask(2); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.addHandler(adapter); - dispatcher.start(); + ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel); + retriever.setReceiveTimeout(0); + retriever.setMaxMessagesPerTask(2); + DispatcherTask dispatcherTask = new DispatcherTask(retriever); + dispatcherTask.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1)); channel.send(new GenericMessage(testObject2)); - assertEquals(2, dispatcher.dispatch()); + assertEquals(2, dispatcherTask.dispatch()); String result = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, result); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/ChannelPollingMessageDispatcherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java similarity index 78% rename from spring-integration-core/src/test/java/org/springframework/integration/bus/ChannelPollingMessageDispatcherTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java index 17f41387fe..f2f2f5e0a5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/ChannelPollingMessageDispatcherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java @@ -17,6 +17,8 @@ package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -26,15 +28,20 @@ import org.junit.Test; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.dispatcher.DefaultMessageDispatcher; +import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.PayloadTypeSelector; +import org.springframework.integration.scheduling.MessagePublishingErrorHandler; +import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler; /** * @author Mark Fisher */ -public class ChannelPollingMessageDispatcherTests { +public class DefaultMessageDispatcherTests { @Test public void testNonBroadcastingDispatcherSendsToExactlyOneEndpoint() throws InterruptedException { @@ -43,14 +50,15 @@ public class ChannelPollingMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); + scheduler.start(); + dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("exactly one endpoint should have received message", 1, counter1.get() + counter2.get()); } @@ -62,15 +70,13 @@ public class ChannelPollingMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(2); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); + channel.setBroadcaster(true); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setBroadcast(true); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("both endpoints should have received message", 2, counter1.get() + counter2.get()); } @@ -84,10 +90,9 @@ public class ChannelPollingMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { @Override public void start() { @@ -96,7 +101,6 @@ public class ChannelPollingMessageDispatcherTests { dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("inactive endpoint should not have received message", 0, counter1.get()); assertEquals("exactly one endpoint should have received message", 1, counter2.get() + counter3.get()); @@ -111,11 +115,10 @@ public class ChannelPollingMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); + channel.setBroadcaster(true); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setBroadcast(true); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override @@ -124,23 +127,21 @@ public class ChannelPollingMessageDispatcherTests { }); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("inactive endpoint should not have received message", 0, counter2.get()); assertEquals("both active endpoints should have received message", 2, counter1.get() + counter3.get()); } @Test - public void testDispatcherWithNoExecutors() { - ConsumerPolicy policy = new ConsumerPolicy(); + public void testDispatcherWithNoExecutorsDoesNotFail() { SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - assertEquals(0, dispatcher.dispatch()); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.start(); } - @Test(expected=MessageDeliveryException.class) - public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() { + @Test + public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger counter3 = new AtomicInteger(); @@ -148,23 +149,30 @@ public class ChannelPollingMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); + channel.setBroadcaster(true); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setBroadcast(true); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { - throw new MessageHandlerRejectedExecutionException(null); + public Message handle(Message message) { + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); + SimpleChannel errorChannel = new SimpleChannel(); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); + scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); + dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); - dispatcher.dispatch(); + latch.await(500, TimeUnit.MILLISECONDS); + Message errorMessage = errorChannel.receive(100); + assertNotNull(errorMessage); + assertTrue(errorMessage instanceof ErrorMessage); + assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass()); } @Test @@ -176,90 +184,95 @@ public class ChannelPollingMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); + channel.setBroadcaster(true); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setBroadcast(true); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1)); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { - throw new MessageHandlerRejectedExecutionException(null); + public Message handle(Message message) { + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1)); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("rejecting endpoint should not have received message", 0, counter2.get()); assertEquals("both non-rejecting endpoints should have received message", 2, counter1.get() + counter3.get()); } - @Test(expected=MessageDeliveryException.class) + @Test public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(1); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { @Override - public Message handle(Message message) { - throw new MessageHandlerRejectedExecutionException(null); + public Message handle(Message message) { + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { - throw new MessageHandlerRejectedExecutionException(null); + public Message handle(Message message) { + throw new MessageHandlerRejectedExecutionException(); } }); + SimpleChannel errorChannel = new SimpleChannel(); + SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(); + scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); + dispatcher.setMessagingTaskScheduler(scheduler); dispatcher.start(); - dispatcher.dispatch(); + Message errorMessage = errorChannel.receive(100); + assertNotNull(errorMessage); + assertTrue(errorMessage instanceof ErrorMessage); + assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass()); } @Test - public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() { + public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() throws InterruptedException { final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); final AtomicInteger rejectedCounter1 = new AtomicInteger(); final AtomicInteger rejectedCounter2 = new AtomicInteger(); - final CountDownLatch latch = new CountDownLatch(1); + final CountDownLatch latch = new CountDownLatch(4); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { rejectedCounter1.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + latch.countDown(); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { rejectedCounter2.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + latch.countDown(); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.start(); - dispatcher.dispatch(); + latch.await(100, TimeUnit.MILLISECONDS); assertEquals("rejecting endpoints should not have received message", 0, counter1.get() + counter2.get()); assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get()); assertEquals("endpoint2 should have rejected two times", 2, rejectedCounter2.get()); @@ -277,39 +290,37 @@ public class ChannelPollingMessageDispatcherTests { TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); TestEndpoint endpoint3 = new TestEndpoint(counter3, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(2); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { rejectedCounter1.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { if (rejectedCounter2.get() == 1) { return super.handle(message); } rejectedCounter2.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { rejectedCounter3.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("endpoint1 should not have received message", 0, counter1.get()); assertEquals("endpoint2 should have received message the second time", 1, counter2.get()); @@ -328,36 +339,34 @@ public class ChannelPollingMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(2); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); + channel.setBroadcaster(true); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setBroadcast(true); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); dispatcher.setRejectionLimit(5); dispatcher.setRetryInterval(3); dispatcher.setShouldFailOnRejectionLimit(false); dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { if (rejectedCounter1.get() == 2) { return super.handle(message); } rejectedCounter1.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { if (rejectedCounter2.get() == 4) { return super.handle(message); } rejectedCounter2.incrementAndGet(); - throw new MessageHandlerRejectedExecutionException(null); + throw new MessageHandlerRejectedExecutionException(); } }); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("endpoint1 should have received one message", 1, counter1.get()); assertEquals("endpoint2 should have received one message", 1, counter2.get()); @@ -372,10 +381,9 @@ public class ChannelPollingMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1); PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1); executor1.addMessageSelector(new PayloadTypeSelector(Integer.class)); @@ -383,7 +391,6 @@ public class ChannelPollingMessageDispatcherTests { dispatcher.addHandler(executor1); dispatcher.addHandler(executor2); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); @@ -399,13 +406,12 @@ public class ChannelPollingMessageDispatcherTests { final CountDownLatch endpointLatch = new CountDownLatch(1); TestEndpoint endpoint1 = new TestEndpoint(counter1, endpointLatch); TestEndpoint endpoint2 = new TestEndpoint(counter2, endpointLatch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { attemptedCounter1.incrementAndGet(); attemptedLatch.countDown(); return super.handle(message); @@ -413,7 +419,7 @@ public class ChannelPollingMessageDispatcherTests { }; PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1) { @Override - public Message handle(Message message) { + public Message handle(Message message) { attemptedCounter2.incrementAndGet(); attemptedLatch.countDown(); return super.handle(message); @@ -424,7 +430,6 @@ public class ChannelPollingMessageDispatcherTests { dispatcher.addHandler(executor1); dispatcher.addHandler(executor2); dispatcher.start(); - dispatcher.dispatch(); attemptedLatch.await(100, TimeUnit.MILLISECONDS); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); assertEquals("endpoint2 should not have accepted the message", 0, counter2.get()); @@ -441,11 +446,10 @@ public class ChannelPollingMessageDispatcherTests { final CountDownLatch latch = new CountDownLatch(1); TestEndpoint endpoint1 = new TestEndpoint(counter1, latch); TestEndpoint endpoint2 = new TestEndpoint(counter2, latch); - ConsumerPolicy policy = new ConsumerPolicy(); SimpleChannel channel = new SimpleChannel(); + channel.setBroadcaster(true); channel.send(new StringMessage(1, "test")); - ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy); - dispatcher.setBroadcast(true); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1); PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1); executor1.addMessageSelector(new PayloadTypeSelector(Integer.class)); @@ -453,14 +457,12 @@ public class ChannelPollingMessageDispatcherTests { dispatcher.addHandler(executor1); dispatcher.addHandler(executor2); dispatcher.start(); - dispatcher.dispatch(); latch.await(100, TimeUnit.MILLISECONDS); assertEquals("endpoint1 should not have accepted the message", 0, counter1.get()); assertEquals("endpoint2 should have accepted the message", 1, counter2.get()); } - - private static class TestEndpoint extends GenericMessageEndpoint { + private static class TestEndpoint extends DefaultMessageEndpoint { private AtomicInteger counter; @@ -473,7 +475,7 @@ public class ChannelPollingMessageDispatcherTests { } @Override - public Message handle(Message message) { + public Message handle(Message message) { counter.incrementAndGet(); latch.countDown(); return null; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index 115d41a176..d19508630c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -24,24 +24,27 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; + import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher */ public class FixedDelayConsumerTests { - //@Test + @Test public void testAllSentMessagesAreReceivedWithinTimeLimit() throws Exception { int messagesToSend = 20; final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new GenericMessageEndpoint() { + MessageEndpoint endpoint = new DefaultMessageEndpoint() { @Override public Message handle(Message message) { counter.incrementAndGet(); @@ -50,18 +53,19 @@ public class FixedDelayConsumerTests { } }; MessageBus bus = new MessageBus(); + bus.initialize(); 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); + PollingSchedule schedule = new PollingSchedule(10); + schedule.setFixedRate(false); + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); + concurrencyPolicy.setCoreConcurrency(1); + concurrencyPolicy.setMaxConcurrency(1); Subscription subscription = new Subscription(); + subscription.setSchedule(schedule); + subscription.setConcurrencyPolicy(concurrencyPolicy); subscription.setChannel("testChannel"); - subscription.setReceiver("testEndpoint"); - subscription.setPolicy(policy); + subscription.setHandler("testEndpoint"); bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { @@ -77,7 +81,7 @@ public class FixedDelayConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new GenericMessageEndpoint() { + MessageEndpoint endpoint = new DefaultMessageEndpoint() { @Override public Message handle(Message message) { counter.incrementAndGet(); @@ -86,18 +90,15 @@ public class FixedDelayConsumerTests { } }; MessageBus bus = new MessageBus(); + bus.initialize(); 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); + PollingSchedule schedule = new PollingSchedule(10); + schedule.setFixedRate(false); Subscription subscription = new Subscription(); subscription.setChannel("testChannel"); - subscription.setReceiver("testEndpoint"); - subscription.setPolicy(policy); + subscription.setHandler("testEndpoint"); + subscription.setSchedule(schedule); bus.activateSubscription(subscription); for (int i = 0; i < messagesToSend; i++) { channel.send(new GenericMessage(1, "test " + (i+1))); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index 8d2f65fb07..b66f8cef84 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -26,10 +26,12 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.ConcurrencyPolicy; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -42,7 +44,7 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new GenericMessageEndpoint() { + MessageEndpoint endpoint = new DefaultMessageEndpoint() { @Override public Message handle(Message message) { counter.incrementAndGet(); @@ -51,15 +53,15 @@ public class FixedRateConsumerTests { } }; MessageBus bus = new MessageBus(); + bus.initialize(); bus.registerChannel("testChannel", channel); bus.registerEndpoint("testEndpoint", endpoint); - ConsumerPolicy policy = new ConsumerPolicy(); - policy.setFixedRate(true); - policy.setPeriod(10); + PollingSchedule schedule = new PollingSchedule(10); + schedule.setFixedRate(true); Subscription subscription = new Subscription(); subscription.setChannel("testChannel"); - subscription.setReceiver("testEndpoint"); - subscription.setPolicy(policy); + subscription.setHandler("testEndpoint"); + subscription.setSchedule(schedule); bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { @@ -75,7 +77,7 @@ public class FixedRateConsumerTests { final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); SimpleChannel channel = new SimpleChannel(); - MessageEndpoint endpoint = new GenericMessageEndpoint() { + MessageEndpoint endpoint = new DefaultMessageEndpoint() { @Override public Message handle(Message message) { counter.incrementAndGet(); @@ -84,18 +86,19 @@ public class FixedRateConsumerTests { } }; MessageBus bus = new MessageBus(); + bus.initialize(); 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(20); + PollingSchedule schedule = new PollingSchedule(5); + schedule.setFixedRate(true); + ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(); + concurrencyPolicy.setCoreConcurrency(1); + concurrencyPolicy.setMaxConcurrency(1); Subscription subscription = new Subscription(); subscription.setChannel("testChannel"); - subscription.setReceiver("testEndpoint"); - subscription.setPolicy(policy); + subscription.setHandler("testEndpoint"); + subscription.setSchedule(schedule); + subscription.setConcurrencyPolicy(concurrencyPolicy); bus.activateSubscription(subscription); bus.start(); for (int i = 0; i < messagesToSend; i++) { @@ -103,8 +106,8 @@ public class FixedRateConsumerTests { } latch.await(80, TimeUnit.MILLISECONDS); int count = counter.get(); - assertTrue("received " + count + ", but expected less than 7", count < 7); - assertTrue("received " + count + ", but expected more than 3", count > 3); + assertTrue("received " + count + ", but expected less than 20", count < 20); + assertTrue("received " + count + ", but expected more than 5", count > 5); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 3c1e5086d7..ddd660b8c9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -21,13 +21,19 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import java.util.Collection; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.adapter.PollableSource; +import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; -import org.springframework.integration.endpoint.GenericMessageEndpoint; +import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; @@ -46,7 +52,7 @@ public class MessageBusTests { bus.registerChannel("sourceChannel", sourceChannel); sourceChannel.send(new StringMessage("123", "test")); bus.registerChannel("targetChannel", targetChannel); - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); endpoint.setInputChannelName("sourceChannel"); endpoint.setDefaultOutputChannelName("targetChannel"); bus.registerEndpoint("endpoint", endpoint); @@ -78,11 +84,9 @@ public class MessageBusTests { sourceChannel.send(new GenericMessage("123", "test")); MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel"); MessageBus bus = (MessageBus) context.getBean("bus"); - ConsumerPolicy policy = new ConsumerPolicy(); Subscription subscription = new Subscription(); subscription.setChannel("sourceChannel"); - subscription.setReceiver("endpoint"); - subscription.setPolicy(policy); + subscription.setHandler("endpoint"); bus.activateSubscription(subscription); Message result = targetChannel.receive(100); assertEquals("test", result.getPayload()); @@ -93,10 +97,10 @@ public class MessageBusTests { SimpleChannel inputChannel = new SimpleChannel(); SimpleChannel outputChannel1 = new SimpleChannel(); SimpleChannel outputChannel2 = new SimpleChannel(); - GenericMessageEndpoint endpoint1 = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(); endpoint1.setDefaultOutputChannelName("output1"); endpoint1.setInputChannelName("input"); - GenericMessageEndpoint endpoint2 = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(); endpoint2.setDefaultOutputChannelName("output2"); endpoint2.setInputChannelName("input"); MessageBus bus = new MessageBus(); @@ -114,11 +118,14 @@ public class MessageBusTests { } @Test - public void testInvalidMessageChannelWithFailedDispatch() { + public void testInvalidMessageChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new MessageBus(); - SourceAdapter sourceAdapter = new FailingSourceAdapter(); + CountDownLatch latch = new CountDownLatch(1); + SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch)); + sourceAdapter.setChannel(new SimpleChannel()); bus.registerSourceAdapter("testAdapter", sourceAdapter); bus.start(); + latch.await(1000, TimeUnit.MILLISECONDS); Message message = bus.getInvalidMessageChannel().receive(100); assertNotNull("message should not be null", message); assertTrue(message instanceof ErrorMessage); @@ -127,27 +134,18 @@ public class MessageBusTests { } - private static class FailingSourceAdapter implements SourceAdapter, MessageDispatcher { + private static class FailingSource implements PollableSource { - public void setChannel(MessageChannel channel) { + private CountDownLatch latch; + + public FailingSource(CountDownLatch latch) { + this.latch = latch; } - public int dispatch() { + public Collection poll(int limit) { + latch.countDown(); throw new RuntimeException("intentional test failure"); } - - public ConsumerPolicy getConsumerPolicy() { - return ConsumerPolicy.newPollingPolicy(1000); - } - - public boolean isRunning() { - return true; - } - - public void start() { - } - - public void stop() { - } } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index 9a8df7c65a..e1fe6f7f5d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -10,7 +10,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 99c6eb1a06..cfc87b8e3d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -35,7 +35,7 @@ public class EndpointParserTests { @Test public void testSimpleEndpoint() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "genericEndpointTests.xml", this.getClass()); + "simpleEndpointTests.xml", this.getClass()); context.start(); MessageChannel channel = (MessageChannel) context.getBean("testChannel"); TestHandler handler = (TestHandler) context.getBean("testHandler"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterEndpointTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterEndpointTests.xml index e8497b9d32..eae3e29c2c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterEndpointTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterEndpointTests.xml @@ -12,7 +12,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/genericEndpointTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml similarity index 96% rename from spring-integration-core/src/test/java/org/springframework/integration/config/genericEndpointTests.xml rename to spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml index 195242c1ab..58fd2a9bd6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/genericEndpointTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml @@ -12,7 +12,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java similarity index 89% rename from spring-integration-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java index 03b0ca312c..b06f4ecb1d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java @@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import org.junit.Test; + import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; @@ -30,18 +31,18 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class GenericMessageEndpointTests { +public class DefaultMessageEndpointTests { @Test public void testDefaultReplyChannel() throws Exception { MessageChannel channel = new SimpleChannel(); MessageChannel replyChannel = new SimpleChannel(); MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + public Message handle(Message message) { return new StringMessage("123", "hello " + message.getPayload()); } }; - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); endpoint.setInputChannelName("testChannel"); endpoint.setHandler(handler); endpoint.setDefaultOutputChannelName("replyChannel"); @@ -62,11 +63,11 @@ public class GenericMessageEndpointTests { MessageChannel channel = new SimpleChannel(); final MessageChannel replyChannel = new SimpleChannel(); MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { + public Message handle(Message message) { return new StringMessage("123", "hello " + message.getPayload()); } }; - GenericMessageEndpoint endpoint = new GenericMessageEndpoint(); + DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(); endpoint.setInputChannelName("testChannel"); endpoint.setHandler(handler); MessageBus bus = new MessageBus(); @@ -77,7 +78,7 @@ public class GenericMessageEndpointTests { StringMessage testMessage = new StringMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel"); channel.send(testMessage); - Message reply = replyChannel.receive(50); + Message reply = replyChannel.receive(50); assertNotNull(reply); assertEquals("hello test", reply.getPayload()); }