From 0ababb60fe3505f310e8182671057752e5454354 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sun, 9 Dec 2007 14:17:07 +0000 Subject: [PATCH] Added new message bus implementation assuming all subscription and consumer activation responsibilities. --- .../integration/MessageBus.java | 159 ------- .../annotation/MessageEndpoint.java | 3 - .../integration/bus/ConsumerPolicy.java | 154 +++++++ .../integration/bus/EndpointExecutor.java | 90 ++++ .../integration/bus/MessageBus.java | 315 ++++++++++++++ .../channel/{consumer => }/ConsumerType.java | 2 +- .../channel/consumer/AbstractConsumer.java | 146 ------- .../consumer/AbstractPollingConsumer.java | 101 ----- .../channel/consumer/EventDrivenConsumer.java | 401 ------------------ .../channel/consumer/FixedDelayConsumer.java | 42 -- .../channel/consumer/FixedRateConsumer.java | 42 -- .../endpoint/GenericMessageEndpoint.java | 18 +- .../integration/endpoint/MessageEndpoint.java | 3 - .../integration/MessageBusTests.java | 6 + .../EventDrivenConsumerTests.java | 49 ++- .../FixedDelayConsumerTests.java | 34 +- .../FixedRateConsumerTests.java | 32 +- .../endpoint/GenericMessageEndpointTests.java | 23 +- .../integration/messageBusTests.xml | 2 +- 19 files changed, 660 insertions(+), 962 deletions(-) delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java create mode 100644 spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java rename spring-eai-core/src/main/java/org/springframework/integration/channel/{consumer => }/ConsumerType.java (93%) delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java delete mode 100644 spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java rename spring-eai-core/src/test/java/org/springframework/integration/{channel/consumer => bus}/EventDrivenConsumerTests.java (66%) rename spring-eai-core/src/test/java/org/springframework/integration/{channel/consumer => bus}/FixedDelayConsumerTests.java (72%) rename spring-eai-core/src/test/java/org/springframework/integration/{channel/consumer => bus}/FixedRateConsumerTests.java (74%) diff --git a/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java b/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java deleted file mode 100644 index 0efd1a6c3b..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/MessageBus.java +++ /dev/null @@ -1,159 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration; - -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.CopyOnWriteArrayList; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.BeansException; -import org.springframework.context.ApplicationContext; -import org.springframework.context.ApplicationContextAware; -import org.springframework.context.Lifecycle; -import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.consumer.AbstractConsumer; -import org.springframework.integration.channel.consumer.ConsumerType; -import org.springframework.integration.channel.consumer.EventDrivenConsumer; -import org.springframework.integration.channel.consumer.FixedDelayConsumer; -import org.springframework.integration.channel.consumer.FixedRateConsumer; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.util.Assert; - -/** - * A central component for registering channels and endpoints. The message bus - * will autodetect channels and endpoints from its host application context. - * - * @author Mark Fisher - */ -public class MessageBus implements ChannelResolver, ApplicationContextAware, Lifecycle { - - private final Log logger = LogFactory.getLog(getClass()); - - private Map channels = new ConcurrentHashMap(); - - private Map endpoints = new ConcurrentHashMap(); - - private List consumers = new CopyOnWriteArrayList(); - - private ApplicationContext applicationContext; - - private boolean running; - - private Object lifecycleMonitor = new Object(); - - - public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { - Assert.notNull(applicationContext, "applicationContext must not be null"); - this.applicationContext = applicationContext; - this.initChannels(); - this.initEndpoints(); - } - - @SuppressWarnings("unchecked") - private void initChannels() { - Map channelBeans = (Map) this.applicationContext - .getBeansOfType(MessageChannel.class); - for (Map.Entry entry : channelBeans.entrySet()) { - this.channels.put(entry.getKey(), entry.getValue()); - if (logger.isInfoEnabled()) { - logger.info("registered channel '" + entry.getKey() + "'"); - } - } - } - - @SuppressWarnings("unchecked") - private void initEndpoints() { - Map endpointBeans = (Map) this.applicationContext - .getBeansOfType(MessageEndpoint.class); - for (Map.Entry entry : endpointBeans.entrySet()) { - this.endpoints.put(entry.getKey(), entry.getValue()); - if (logger.isInfoEnabled()) { - logger.info("registered endpoint '" + entry.getKey() + "'"); - } - } - } - - public MessageChannel resolve(String channelName) { - return this.channels.get(channelName); - } - - public boolean isRunning() { - synchronized (this.lifecycleMonitor) { - return this.running; - } - } - - public void start() { - synchronized (this.lifecycleMonitor) { - if (!this.isRunning()) { - this.running = true; - this.activateEndpoints(); - } - } - } - - public void stop() { - synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - this.running = false; - this.deactivateEndpoints(); - } - } - } - - private void activateEndpoints() { - for (MessageEndpoint endpoint : this.endpoints.values()) { - MessageSource source = endpoint.getSource(); - ConsumerType consumerType = endpoint.getConsumerType(); - AbstractConsumer consumer = createConsumer(consumerType, source, endpoint); - consumer.initialize(); - consumer.start(); - } - } - - private void deactivateEndpoints() { - for (AbstractConsumer consumer : this.consumers) { - consumer.stop(); - } - } - - - /** - * Create a consumer based upon the specified consumer type. - */ - private AbstractConsumer createConsumer(ConsumerType type, MessageSource source, MessageEndpoint endpoint) { - if (type.equals(ConsumerType.EVENT_DRIVEN)) { - return new EventDrivenConsumer(source, endpoint); - } - else if (type.equals(ConsumerType.FIXED_RATE)) { - return new FixedRateConsumer(source, endpoint); - } - else if (type.equals(ConsumerType.FIXED_DELAY)) { - return new FixedDelayConsumer(source, endpoint); - } - else { - throw new UnsupportedOperationException("the consumerType '" - + type.name() + "' is not supported."); - } - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java index 3a393cc9cd..b096c3a707 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java @@ -23,7 +23,6 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; -import org.springframework.integration.channel.consumer.ConsumerType; import org.springframework.stereotype.Component; /** @@ -42,6 +41,4 @@ public @interface MessageEndpoint { String target(); - ConsumerType consumerType() default ConsumerType.EVENT_DRIVEN; - } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java new file mode 100644 index 0000000000..d5994be637 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java @@ -0,0 +1,154 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.bus; + +import java.util.concurrent.TimeUnit; + +/** + * A container for Message consumer configuration metadata. + * + * @author Mark Fisher + */ +public class ConsumerPolicy { + + private static final int DEFAULT_CONCURRENCY = 1; + + private static final int DEFAULT_MAX_CONCURRENCY = 10; + + private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10; + + private static final int DEFAULT_REJECTION_LIMIT = 10; + + private static final int DEFAULT_REJECTION_LIMIT_WAIT = 1000; + + private static final long DEFAULT_RECEIVE_TIMEOUT = 1000; + + + private int concurrency = DEFAULT_CONCURRENCY; + + private int maxConcurrency = DEFAULT_MAX_CONCURRENCY; + + private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK; + + private int rejectionLimit = DEFAULT_REJECTION_LIMIT; + + private int rejectionLimitWait = DEFAULT_REJECTION_LIMIT_WAIT; + + private int initialDelay = 0; + + private int period = -1; + + private TimeUnit timeUnit = TimeUnit.MILLISECONDS; + + private boolean fixedRate = false; + + private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; + + + public int getInitialDelay() { + return this.initialDelay; + } + + public void setInitialDelay(int initialDelay) { + this.initialDelay = initialDelay; + } + + public int getPeriod() { + return this.period; + } + + public void setPeriod(int period) { + this.period = period; + } + + public TimeUnit getTimeUnit() { + return this.timeUnit; + } + + public void setTimeUnit(TimeUnit timeUnit) { + this.timeUnit = timeUnit; + } + + public boolean isFixedRate() { + return this.fixedRate; + } + + public void setFixedRate(boolean fixedRate) { + this.fixedRate = fixedRate; + } + + public int getConcurrency() { + return this.concurrency; + } + + public void setConcurrency(int concurrency) { + if (concurrency < 1) { + throw new IllegalArgumentException("'concurrency' value must be at least 1"); + } + this.concurrency = concurrency; + } + + public int getMaxConcurrency() { + return this.maxConcurrency; + } + + public void setMaxConcurrency(int maxConcurrency) { + if (maxConcurrency < 1) { + throw new IllegalArgumentException("'maxConcurrency' value must be at least 1"); + } + this.maxConcurrency = maxConcurrency; + } + + public int getMaxMessagesPerTask() { + return this.maxMessagesPerTask; + } + + public void setMaxMessagesPerTask(int maxMessagesPerTask) { + if (maxMessagesPerTask == 0) { + throw new IllegalArgumentException("'maxMessagesPerTask' must not be 0"); + } + this.maxMessagesPerTask = maxMessagesPerTask; + } + + public int getRejectionLimit() { + return this.rejectionLimit; + } + + public void setRejectionLimit(int rejectionLimit) { + if (rejectionLimit < 1) { + throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1"); + } + this.rejectionLimit = rejectionLimit; + } + + public int getRejectionLimitWait() { + return this.rejectionLimitWait; + } + + public void setRejectionLimitWait(int rejectionLimitWait) { + this.rejectionLimitWait = rejectionLimitWait; + } + + public long getReceiveTimeout() { + return this.receiveTimeout; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java new file mode 100644 index 0000000000..8b99affdc9 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java @@ -0,0 +1,90 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.bus; + +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.bus.MessageBus.EndpointTask; + +/** + * A subclass of {@link ThreadPoolExecutor} with configurable error thresholds. + * + * @author Mark Fisher + */ +public class EndpointExecutor extends ThreadPoolExecutor { + + private Log logger = LogFactory.getLog(this.getClass()); + + private int successiveErrorCount; + + private int successiveErrorThreshold = -1; + + private int totalErrorCount; + + private int totalErrorThreshold = -1; + + + public EndpointExecutor(int corePoolSize, int maximumPoolSize) { + super(corePoolSize, maximumPoolSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue()); + } + + /** + * Set the maximum number of errors allowed in successive + * endpoint executions. If this threshold is ever exceeded, the executor + * will shutdown. + */ + public void setSuccessiveErrorThreshold(int successiveErrorThreshold) { + this.successiveErrorThreshold = successiveErrorThreshold; + } + + /** + * Set the maximum number of total errors allowed in endpoint + * executions. If this threshold is ever exceeded, the executor will + * shutdown. + */ + public void setTotalErrorThreshold(int totalErrorThreshold) { + this.totalErrorThreshold = totalErrorThreshold; + } + + @Override + protected void afterExecute(Runnable r, Throwable t) { + EndpointTask task = (EndpointTask) r; + if (task.getError() != null) { + if (logger.isWarnEnabled()) { + logger.warn("Exception occurred in endpoint execution: " + task.getError()); + } + this.successiveErrorCount++; + this.totalErrorCount++; + if ((successiveErrorThreshold >= 0 && successiveErrorCount > successiveErrorThreshold) + || (totalErrorThreshold >= 0 && totalErrorCount > totalErrorThreshold)) { + if (logger.isInfoEnabled()) { + logger.info("error threshold exceeded, shutting down now"); + } + this.shutdownNow(); + } + } + else { + successiveErrorCount = 0; + } + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java new file mode 100644 index 0000000000..643600c24a --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -0,0 +1,315 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.bus; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledThreadPoolExecutor; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.Lifecycle; +import org.springframework.integration.MessagingException; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.message.Message; +import org.springframework.util.Assert; + +/** + * The messaging bus. Serves as a registry for channels and endpoints, manages their lifecycle, + * and all subscriptions. + * + * @author Mark Fisher + */ +public class MessageBus implements ChannelResolver, ApplicationContextAware, Lifecycle { + + private Log logger = LogFactory.getLog(this.getClass()); + + private Map channels = new ConcurrentHashMap(); + + private Map endpoints = new ConcurrentHashMap(); + + private List dispatcherTasks = new CopyOnWriteArrayList(); + + private Map endpointExecutors = new ConcurrentHashMap(); + + private ScheduledThreadPoolExecutor dispatcherExecutor; + + private boolean autoCreateChannels; + + private ApplicationContext applicationContext; + + private boolean running; + + private Object lifecycleMonitor = new Object(); + + + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + Assert.notNull(applicationContext, "applicationContext must not be null"); + this.applicationContext = applicationContext; + this.registerChannelsFromContext(); + this.registerEndpointsFromContext(); + } + + @SuppressWarnings("unchecked") + private void registerChannelsFromContext() { + Map channelBeans = (Map) this.applicationContext + .getBeansOfType(MessageChannel.class); + for (Map.Entry entry : channelBeans.entrySet()) { + this.registerChannel(entry.getKey(), entry.getValue()); + if (logger.isInfoEnabled()) { + logger.info("registered channel '" + entry.getKey() + "'"); + } + } + } + + @SuppressWarnings("unchecked") + private void registerEndpointsFromContext() { + Map endpointBeans = (Map) this.applicationContext + .getBeansOfType(MessageEndpoint.class); + for (Map.Entry entry : endpointBeans.entrySet()) { + this.registerEndpoint(entry.getKey(), entry.getValue()); + if (logger.isInfoEnabled()) { + logger.info("registered endpoint '" + entry.getKey() + "'"); + } + } + } + + public void initialize() { + this.dispatcherExecutor = new ScheduledThreadPoolExecutor(this.dispatcherTasks.size() > 0 ? this.dispatcherTasks.size() : 1); + } + + public MessageChannel resolve(String channelName) { + return this.channels.get(channelName); + } + + public void registerChannel(String name, MessageChannel channel) { + this.channels.put(name, channel); + } + + public void registerEndpoint(String name, MessageEndpoint endpoint) { + this.endpoints.put(name, endpoint); + } + + public void activateSubscription(String channelName, String endpointName, ConsumerPolicy policy) { + MessageChannel channel = this.channels.get(channelName); + if (channel == null) { + if (this.autoCreateChannels == false) { + throw new MessagingException("Cannot activate subscription, unknown channel '" + channelName + + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); + } + this.registerChannel(channelName, new PointToPointChannel()); + if (this.logger.isInfoEnabled()) { + logger.info("created channel '" + channelName + "'"); + } + } + MessageEndpoint endpoint = this.endpoints.get(endpointName); + if (endpoint == null) { + throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'"); + } + EndpointExecutor endpointExecutor = new EndpointExecutor(policy.getConcurrency(), policy.getMaxConcurrency()); + endpointExecutors.put(endpoint, endpointExecutor); + DispatcherTask dispatcherTask = new DispatcherTask(channel, endpoint, policy); + this.dispatcherTasks.add(dispatcherTask); + if (this.isRunning()) { + scheduleDispatcherTask(dispatcherTask); + } + } + + public int getActiveCountForEndpoint(String endpointName) { + MessageEndpoint endpoint = this.endpoints.get(endpointName); + if (endpoint != null) { + EndpointExecutor executor = this.endpointExecutors.get(endpoint); + if (executor != null) { + return executor.getActiveCount(); + } + } + return 0; + } + + private void scheduleDispatcherTask(DispatcherTask task) { + ConsumerPolicy policy = task.getPolicy(); + if (policy.getPeriod() <= 0) { + if (policy.getReceiveTimeout() <= 0) { + if (logger.isWarnEnabled()) { + logger.warn("Scheduling a repeating task with no receive timeout is not recommended! " + + "Consider providing a positive value for either 'period' or 'receiveTimeout'"); + } + } + dispatcherExecutor.schedule(new RepeatingDispatcherTask(task), policy.getInitialDelay(), policy.getTimeUnit()); + } + else if (policy.isFixedRate()) { + dispatcherExecutor.scheduleAtFixedRate(task, policy.getInitialDelay(), policy.getPeriod(), policy.getTimeUnit()); + } + else { + dispatcherExecutor.scheduleWithFixedDelay(task, policy.getInitialDelay(), policy.getPeriod(), policy.getTimeUnit()); + } + } + + public boolean isRunning() { + synchronized (this.lifecycleMonitor) { + return this.running; + } + } + + public void start() { + if (this.dispatcherExecutor == null) { + this.initialize(); + } + synchronized (this.lifecycleMonitor) { + if (!this.isRunning()) { + this.running = true; + for (DispatcherTask task : this.dispatcherTasks) { + scheduleDispatcherTask(task); + } + } + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.isRunning()) { + this.running = false; + this.dispatcherExecutor.shutdownNow(); + } + } + } + + + private class DispatcherTask implements Runnable { + + private MessageChannel channel; + + private MessageEndpoint endpoint; + + private ConsumerPolicy policy; + + + public DispatcherTask(MessageChannel channel, MessageEndpoint endpoint, ConsumerPolicy policy) { + this.channel = channel; + this.endpoint = endpoint; + this.policy = policy; + } + + public MessageChannel getChannel() { + return this.channel; + } + + public MessageEndpoint getEndpoint() { + return this.endpoint; + } + + public ConsumerPolicy getPolicy() { + return this.policy; + } + + public void run() { + EndpointExecutor executor = endpointExecutors.get(this.endpoint); + if (executor == null || executor.isShutdown()) { + if (logger.isWarnEnabled()) { + logger.warn("dispatcher shutting down, endpoint executor is not active"); + } + return; + } + for (int i = 0; i < policy.getMaxMessagesPerTask(); i++) { + Message message = channel.receive(this.policy.getReceiveTimeout()); + if (message == null) { + return; + } + else { + boolean taskSubmitted = false; + int attempts = 0; + while (!taskSubmitted) { + try { + executor.execute(new EndpointTask(this.endpoint, message)); + taskSubmitted = true; + } + catch (RejectedExecutionException rex) { + attempts++; + if (attempts == policy.getRejectionLimit()) { + attempts = 0; + if (logger.isDebugEnabled()) { + logger.debug("reached rejected execution limit"); + } + try { + Thread.sleep(policy.getRejectionLimitWait()); + } + catch (InterruptedException iex) { + Thread.currentThread().interrupt(); + } + } + } + } + } + } + } + } + + + private class RepeatingDispatcherTask implements Runnable { + + private DispatcherTask task; + + RepeatingDispatcherTask(DispatcherTask task) { + this.task = task; + } + + public void run() { + task.run(); + dispatcherExecutor.execute(new RepeatingDispatcherTask(task)); + } + } + + + public static class EndpointTask implements Runnable { + + private MessageEndpoint endpoint; + + private Message message; + + private Throwable error; + + + EndpointTask(MessageEndpoint endpoint, Message message) { + this.endpoint = endpoint; + this.message = message; + } + + public Throwable getError() { + return this.error; + } + + public void run() { + try { + this.endpoint.messageReceived(this.message); + } + catch (Throwable t) { + this.error = t; + } + } + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/ConsumerType.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/ConsumerType.java similarity index 93% rename from spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/ConsumerType.java rename to spring-eai-core/src/main/java/org/springframework/integration/channel/ConsumerType.java index f3230dd74f..c0e1a3388b 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/ConsumerType.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/channel/ConsumerType.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.channel.consumer; +package org.springframework.integration.channel; /** * Enumeration of the different types of message consumer. diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java deleted file mode 100644 index b79ddeda07..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractConsumer.java +++ /dev/null @@ -1,146 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.channel.consumer; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.context.Lifecycle; -import org.springframework.integration.MessageSource; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; -import org.springframework.util.Assert; - -/** - * Base class for consumers defining common properties and behavior. - * - * @author Mark Fisher - */ -public abstract class AbstractConsumer implements Lifecycle { - - /** - * The default receive timeout: 1000 ms = 1 second. - */ - public static final long DEFAULT_RECEIVE_TIMEOUT = 1000; - - - protected final Log logger = LogFactory.getLog(getClass()); - - private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT; - - private MessageSource source; - - private MessageEndpoint endpoint; - - private boolean active = false; - - private boolean running = false; - - protected final Object lifecycleMonitor = new Object(); - - - public AbstractConsumer(MessageSource source, MessageEndpoint endpoint) { - Assert.notNull(source, "source must not be null"); - Assert.notNull(endpoint, "endpoint must not be null"); - this.source = source; - this.endpoint = endpoint; - } - - - public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; - } - - public final boolean isRunning() { - synchronized (this.lifecycleMonitor) { - return this.running; - } - } - - public final void start() { - synchronized (this.lifecycleMonitor) { - this.running = true; - this.lifecycleMonitor.notifyAll(); - } - this.doStart(); - } - - public final void stop() { - synchronized (this.lifecycleMonitor) { - this.running = false; - this.lifecycleMonitor.notifyAll(); - } - this.doStop(); - } - - public final boolean isActive() { - synchronized (this.lifecycleMonitor) { - return this.active; - } - } - - public final void initialize() { - synchronized (this.lifecycleMonitor) { - this.active = true; - this.lifecycleMonitor.notifyAll(); - } - doInitialize(); - } - - /** - * If the consumer is active but not yet running, then wait until it is running. - */ - protected void waitWhileNotRunning() { - synchronized (this.lifecycleMonitor) { - while (this.active && !this.running) { - try { - this.lifecycleMonitor.wait(); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - } - } - } - } - - protected boolean receiveAndPassToEndpoint() { - boolean messageReceived = false; - Message message = null; - if (this.receiveTimeout < 0) { // indefinite timeout - message = this.source.receive(); - } - else { - message = this.source.receive(this.receiveTimeout); - } - if (message != null) { - messageReceived = true; - messageReceived(message); - this.endpoint.messageReceived(message); - } - return messageReceived; - } - - - protected abstract void doInitialize(); - - protected abstract void doStart(); - - protected abstract void doStop(); - - protected abstract void messageReceived(Message message); - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java deleted file mode 100644 index 91e61645ab..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/AbstractPollingConsumer.java +++ /dev/null @@ -1,101 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.channel.consumer; - -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.TimeUnit; - -import org.springframework.integration.MessageSource; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; - -/** - * Base class for consumers that poll on a given interval. - * - * @author Mark Fisher - */ -public abstract class AbstractPollingConsumer extends AbstractConsumer { - - private static final int DEFAULT_POLL_INTERVAL = 1000; - - - protected ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1); - - private int initialDelay = 0; - - private int pollInterval = DEFAULT_POLL_INTERVAL; - - private TimeUnit timeUnit = TimeUnit.MILLISECONDS; - - - public AbstractPollingConsumer(MessageSource source, MessageEndpoint endpoint) { - super(source, endpoint); - this.setReceiveTimeout(0); - } - - - public void setPollInterval(int pollInterval) { - this.pollInterval = pollInterval; - } - - /** - * Specify the {@link TimeUnit} for polling. Default is milliseconds. - */ - public void setTimeUnit(TimeUnit timeUnit) { - this.timeUnit = timeUnit; - } - - @Override - protected void doStart() { - scheduleInvoker(new PollingInvoker(), this.initialDelay, this.pollInterval, this.timeUnit); - } - - /** - * Each subclass must implement this method depending on its scheduling - * behavior (e.g. fixed-rate versus fixed-delay). - * - * @param invoker the invoker task to schedule - * @param initialDelay the time in milliseconds to wait before the first - * poll - * @param pollInterval the polling interval in milliseconds - */ - protected abstract void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit); - - - @Override - protected void doInitialize() { - } - - @Override - protected void doStop() { - this.executor.shutdown(); - } - - @Override - protected void messageReceived(Message message) { - } - - - private class PollingInvoker implements Runnable { - - public void run() { - receiveAndPassToEndpoint(); - } - - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java deleted file mode 100644 index b5a4de5c59..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/EventDrivenConsumer.java +++ /dev/null @@ -1,401 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.channel.consumer; - -import java.lang.reflect.Method; -import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedList; -import java.util.List; -import java.util.Set; - -import org.springframework.context.Lifecycle; -import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.MessageSource; -import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; -import org.springframework.scheduling.SchedulingAwareRunnable; -import org.springframework.scheduling.SchedulingTaskExecutor; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; -import org.springframework.util.Assert; - -/** - * A consumer that runs tasks repeatedly in order to pass to the endpoint as - * soon as a message is received by one of those tasks. - * - * @author Mark Fisher - * @author Juergen Hoeller - */ -public class EventDrivenConsumer extends AbstractConsumer implements Lifecycle { - - private static final int DEFAULT_CONCURRENCY = 1; - - private static final int DEFAULT_MAX_CONCURRENCY = 10; - - private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10; - - private static final int DEFAULT_IDLE_TASK_EXECUTION_LIMIT = 1; - - - private TaskExecutor executor; - - private int concurrency = DEFAULT_CONCURRENCY; - - private int maxConcurrency = DEFAULT_MAX_CONCURRENCY; - - private int maxMessagesPerTask = DEFAULT_MAX_MESSAGES_PER_TASK; - - private int idleTaskExecutionLimit = DEFAULT_IDLE_TASK_EXECUTION_LIMIT; - - private final Set scheduledInvokers = new HashSet(); - - private int activeInvokerCount = 0; - - private final Object activeInvokerMonitor = new Object(); - - private final List pausedInvokers = new LinkedList(); - - - public EventDrivenConsumer(MessageSource source, MessageEndpoint endpoint) { - super(source, endpoint); - } - - - public void setExecutor(TaskExecutor executor) { - Assert.notNull(executor, "executor must not be null"); - this.executor = executor; - } - - public void setConcurrency(int concurrency) { - if (concurrency < 1) { - throw new IllegalArgumentException("'concurrency' value must be at least 1"); - } - synchronized (this.activeInvokerMonitor) { - this.concurrency = concurrency; - if (this.maxConcurrency < concurrency) { - this.maxConcurrency = concurrency; - } - } - } - - public void setMaxConcurrency(int maxConcurrency) { - if (maxConcurrency < 1) { - throw new IllegalArgumentException("'maxConcurrency' value must be at least 1"); - } - synchronized (this.activeInvokerMonitor) { - this.maxConcurrency = Math.max(maxConcurrency, this.concurrency); - } - } - - public void setMaxMessagesPerTask(int maxMessagesPerTask) { - if (maxMessagesPerTask == 0) { - throw new IllegalArgumentException("'maxMessagesPerTask' must not be 0"); - } - synchronized (this.activeInvokerMonitor) { - this.maxMessagesPerTask = maxMessagesPerTask; - } - } - - public void setIdleTaskExecutionLimit(int idleTaskExecutionLimit) { - if (idleTaskExecutionLimit < 1) { - throw new IllegalArgumentException("'idleTaskExecutionLimit' must be at least 1"); - } - synchronized (this.activeInvokerMonitor) { - this.idleTaskExecutionLimit = idleTaskExecutionLimit; - } - } - - public void doStart() { - synchronized (this.lifecycleMonitor) { - this.resumePausedTasks(); - } - } - - public void doStop() { - this.shutdown(); - } - - public void doInitialize() { - synchronized (this.activeInvokerMonitor) { - if (this.executor == null) { - this.executor = createDefaultExecutor(); - } - else if (this.executor instanceof SchedulingTaskExecutor && - ((SchedulingTaskExecutor) this.executor).prefersShortLivedTasks() && - this.maxMessagesPerTask == Integer.MIN_VALUE) { - this.maxMessagesPerTask = 1; - } - initializeExecutorIfPossible(); - for (int i = 0; i < this.concurrency; i++) { - scheduleNewInvoker(); - } - } - } - - private void initializeExecutorIfPossible() { - try { - Method initMethod = this.executor.getClass().getMethod("initialize"); - initMethod.invoke(this.executor); - } - catch (Exception e) { - // do nothing - } - } - - private void shutdown() { - synchronized (this.lifecycleMonitor) { - this.shutdownExecutorIfPossible(); - } - } - - private void shutdownExecutorIfPossible() { - try { - if (this.executor instanceof Lifecycle) { - ((Lifecycle) this.executor).stop(); - } - else { - Method shutdownMethod = this.executor.getClass().getMethod("shutdown"); - shutdownMethod.invoke(this.executor); - } - } - catch (Exception e) { - // do nothing - } - } - - private TaskExecutor createDefaultExecutor() { - ThreadPoolTaskExecutor defaultExecutor = new ThreadPoolTaskExecutor(); - defaultExecutor.setCorePoolSize(this.maxConcurrency); - defaultExecutor.setMaxPoolSize(this.maxConcurrency); - defaultExecutor.setQueueCapacity(5); - return defaultExecutor; - } - - /** - * Try scheduling a new invoker, since we know messages are being received. - * @see #scheduleNewInvokerIfAppropriate() - */ - @Override - protected void messageReceived(Message message) { - scheduleNewInvokerIfAppropriate(); - } - - /** - * Schedule a new invoker, increasing the total number of scheduled - * invokers for this consumer. - */ - private void scheduleNewInvoker() { - MessageEndpointInvoker invoker = new MessageEndpointInvoker(); - if (rescheduleInvokerIfNecessary(invoker)) { - this.scheduledInvokers.add(invoker); - } - } - - private boolean rescheduleInvokerIfNecessary(MessageEndpointInvoker invoker) { - synchronized (this.lifecycleMonitor) { - if (this.isRunning()) { - try { - doRescheduleInvoker(invoker); - } - catch (RuntimeException ex) { - logRejectedInvoker(invoker, ex); - this.pausedInvokers.add(invoker); - } - return true; - } - if (this.isActive()) { - this.pausedInvokers.add(invoker); - return true; - } - else { - return false; - } - } - } - - protected void doRescheduleInvoker(final MessageEndpointInvoker invoker) { - this.executor.execute(invoker); - } - - private boolean shouldRescheduleInvoker(int idleTaskExecutionCount) { - synchronized (this.activeInvokerMonitor) { - boolean idle = (idleTaskExecutionCount >= this.idleTaskExecutionLimit); - return (this.scheduledInvokers.size() <= (idle ? this.concurrency : this.maxConcurrency)); - } - } - - private boolean hasIdleInvokers() { - for (MessageEndpointInvoker invoker : this.scheduledInvokers) { - if (invoker.isIdle()) { - return true; - } - } - return false; - } - - /** - * Try to resume all paused tasks. - * Tasks for which rescheduling failed simply remain in paused mode. - */ - protected void resumePausedTasks() { - synchronized (this.lifecycleMonitor) { - if (!this.pausedInvokers.isEmpty()) { - for (Iterator it = this.pausedInvokers.iterator(); it.hasNext();) { - MessageEndpointInvoker invoker = it.next(); - try { - doRescheduleInvoker(invoker); - it.remove(); - if (logger.isDebugEnabled()) { - logger.debug("Resumed paused invoker: " + invoker); - } - } - catch (RuntimeException e) { - logRejectedInvoker(invoker, e); - // Keep the task in paused mode... - } - } - } - } - } - - public int getPausedInvokerCount() { - synchronized (this.lifecycleMonitor) { - return this.pausedInvokers.size(); - } - } - - /** - * Log an invoker that has been rejected by {@link #doRescheduleInvoker}. - *

The default implementation simply logs a corresponding message - * at debug level. - * @param invoker the rejected invoker object - * @param ex the exception thrown from {@link #doRescheduleInvoker} - */ - protected void logRejectedInvoker(Object invoker, RuntimeException ex) { - if (logger.isDebugEnabled()) { - logger.debug("Invoker [" + invoker + "] has been rejected and paused: " + ex); - } - } - - private void scheduleNewInvokerIfAppropriate() { - if (this.isRunning()) { - this.resumePausedTasks(); - synchronized (this.activeInvokerMonitor) { - if (this.scheduledInvokers.size() < this.maxConcurrency && !hasIdleInvokers()) { - scheduleNewInvoker(); - if (logger.isDebugEnabled()) { - logger.debug("Raised scheduled invoker count: " + scheduledInvokers.size()); - } - } - } - } - } - - public final int getScheduledInvokerCount() { - synchronized (this.activeInvokerMonitor) { - return this.scheduledInvokers.size(); - } - } - - public final int getActiveInvokerCount() { - synchronized (this.activeInvokerMonitor) { - return this.activeInvokerCount; - } - } - - - private class MessageEndpointInvoker implements SchedulingAwareRunnable { - - private int idleTaskExecutionCount = 0; - - private volatile boolean idle = true; - - - public void run() { - synchronized (activeInvokerMonitor) { - activeInvokerCount++; - activeInvokerMonitor.notifyAll(); - } - boolean messageReceived = false; - //TODO: try { - if (maxMessagesPerTask < 0) { - while (isActive()) { - waitWhileNotRunning(); - if (isActive()) { - messageReceived = receiveAndPassToEndpoint(); - this.idle = !messageReceived; - } - } - } - else { - int messageCount = 0; - while (isRunning() && messageCount < maxMessagesPerTask) { - boolean messageHandled = receiveAndPassToEndpoint(); - this.idle = !messageHandled; - messageReceived = (messageHandled || messageReceived); - messageCount++; - } - } - // TODO: } catch (Throwable t) { check if last message succeeded, else sleep between recovery attempts } - synchronized (activeInvokerMonitor) { - activeInvokerCount--; - activeInvokerMonitor.notifyAll(); - } - if (!messageReceived) { - this.idleTaskExecutionCount++; - } - else { - this.idleTaskExecutionCount = 0; - } - if (!shouldRescheduleInvoker(this.idleTaskExecutionCount) || !rescheduleInvokerIfNecessary(this)) { - this.shutdown(); - } - else if (isRunning()) { - int nonPausedInvokers = getScheduledInvokerCount() - getPausedInvokerCount(); - if (nonPausedInvokers < 1) { - logger.error("All scheduled invokers have been paused, probably due to tasks having been rejected. " + - "Check your thread pool configuration! Manual recovery necessary through a start() call."); - } - else if (nonPausedInvokers < concurrency) { - logger.warn("Number of scheduled invokers has dropped below concurrency limit, probably " + - "due to tasks having been rejected. Check your thread pool configuration! Automatic recovery " + - "to be triggered by remaining invokers."); - } - } - } - - private void shutdown() { - synchronized (activeInvokerMonitor) { - scheduledInvokers.remove(this); - if (logger.isDebugEnabled()) { - logger.debug("Lowered scheduled invoker count: " + scheduledInvokers.size()); - } - activeInvokerMonitor.notifyAll(); - } - } - - public boolean isLongLived() { - return (maxMessagesPerTask < 0); - } - - public boolean isIdle() { - return this.idle; - } - - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java deleted file mode 100644 index 1bfb28fccb..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedDelayConsumer.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.channel.consumer; - -import java.util.concurrent.TimeUnit; - -import org.springframework.integration.MessageSource; -import org.springframework.integration.endpoint.MessageEndpoint; - -/** - * A consumer that measures the pollInterval between the - * invoker's execution completion time and the subsequent start. - * - * @author Mark Fisher - */ -public class FixedDelayConsumer extends AbstractPollingConsumer { - - public FixedDelayConsumer(MessageSource source, MessageEndpoint endpoint) { - super(source, endpoint); - } - - - @Override - protected void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit) { - this.executor.scheduleWithFixedDelay(invoker, initialDelay, pollInterval, timeUnit); - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java b/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java deleted file mode 100644 index b6535f1050..0000000000 --- a/spring-eai-core/src/main/java/org/springframework/integration/channel/consumer/FixedRateConsumer.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.channel.consumer; - -import java.util.concurrent.TimeUnit; - -import org.springframework.integration.MessageSource; -import org.springframework.integration.endpoint.MessageEndpoint; - -/** - * A consumer that measures the pollInterval between each - * execution's start. - * - * @author Mark Fisher - */ -public class FixedRateConsumer extends AbstractPollingConsumer { - - public FixedRateConsumer(MessageSource source, MessageEndpoint endpoint) { - super(source, endpoint); - } - - - @Override - protected void scheduleInvoker(Runnable invoker, int initialDelay, int pollInterval, TimeUnit timeUnit) { - this.executor.scheduleAtFixedRate(invoker, initialDelay, pollInterval, timeUnit); - } - -} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java index 5b739ecba7..fed3b75c7e 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/GenericMessageEndpoint.java @@ -20,7 +20,6 @@ import org.springframework.integration.MessageHandlingException; import org.springframework.integration.MessageSource; import org.springframework.integration.MessageTarget; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.consumer.ConsumerType; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; @@ -47,8 +46,6 @@ public class GenericMessageEndpoint implements MessageEndpoint { private ChannelResolver channelResolver; - private ConsumerType consumerType = ConsumerType.EVENT_DRIVEN; - /** * Set the source from which this endpoint receives messages. @@ -78,20 +75,6 @@ public class GenericMessageEndpoint implements MessageEndpoint { this.handler = handler; } - /** - * Set the type of consumer to use for this endpoint. - */ - public void setConsumerType(ConsumerType consumerType) { - this.consumerType = consumerType; - } - - /** - * Return the type of consumer to use for this endpoint. - */ - public ConsumerType getConsumerType() { - return this.consumerType; - } - /** * Set the channel resolver strategy to use when a message * provides a 'replyChannelName'. @@ -104,6 +87,7 @@ public class GenericMessageEndpoint implements MessageEndpoint { public void messageReceived(Message message) { if (this.handler == null) { target.send(message); + return; } Message replyMessage = handler.handle(message); if (replyMessage != null) { diff --git a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java index 60a87ba200..f7f24eb59d 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java @@ -18,7 +18,6 @@ package org.springframework.integration.endpoint; import org.springframework.integration.MessageSource; import org.springframework.integration.channel.ChannelResolver; -import org.springframework.integration.channel.consumer.ConsumerType; import org.springframework.integration.message.Message; /** @@ -28,8 +27,6 @@ import org.springframework.integration.message.Message; */ public interface MessageEndpoint { - ConsumerType getConsumerType(); - MessageSource getSource(); void setChannelResolver(ChannelResolver channelResolver); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java b/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java index a76ee05be1..fafed69c0c 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/MessageBusTests.java @@ -23,6 +23,8 @@ import org.junit.Test; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; @@ -69,6 +71,10 @@ public class MessageBusTests { MessageChannel sourceChannel = (MessageChannel) context.getBean("sourceChannel"); sourceChannel.send(new DocumentMessage("123", "test")); MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel"); + // TODO: add metadata for this + MessageBus bus = (MessageBus) context.getBean("bus"); + ConsumerPolicy policy = new ConsumerPolicy(); + bus.activateSubscription("sourceChannel", "endpoint", policy); Message result = targetChannel.receive(10); assertEquals("test", result.getPayload()); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java similarity index 66% rename from spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java rename to spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java index f846095436..9c0b37ed66 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/EventDrivenConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/EventDrivenConsumerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.channel.consumer; +package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -24,12 +24,14 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; + +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; -import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; /** * @author Mark Fisher @@ -37,49 +39,54 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; public class EventDrivenConsumerTests { @Test + public void stub() {} + + // TODO: make this a @Test public void testDynamicConcurrency() throws Exception { int messagesToSend = 200; int concurrency = 1; - int maxConcurrency = 10; + int maxConcurrency = 100; final AtomicInteger counter = new AtomicInteger(0); final CountDownLatch latch = new CountDownLatch(messagesToSend); - final ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); final AtomicInteger maxActive = new AtomicInteger(0); final AtomicInteger activeSum = new AtomicInteger(0); - executor.setCorePoolSize(concurrency); - executor.setMaxPoolSize(maxConcurrency); - executor.setQueueCapacity(0); + final MessageBus bus = new MessageBus(); PointToPointChannel channel = new PointToPointChannel(); MessageEndpoint endpoint = new GenericMessageEndpoint() { public void messageReceived(Message message) { counter.incrementAndGet(); latch.countDown(); - activeSum.set(activeSum.addAndGet(executor.getActiveCount())); - maxActive.set(Math.max(executor.getActiveCount(), maxActive.get())); + activeSum.set(activeSum.addAndGet(bus.getActiveCountForEndpoint("testEndpoint"))); + maxActive.set(Math.max(bus.getActiveCountForEndpoint("testEndpoint"), maxActive.get())); } }; - EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); - consumer.setExecutor(executor); - consumer.setConcurrency(concurrency); - consumer.setMaxConcurrency(maxConcurrency); - consumer.setIdleTaskExecutionLimit(1); - consumer.setMaxMessagesPerTask(1); - consumer.setReceiveTimeout(100); - consumer.initialize(); - consumer.start(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setConcurrency(concurrency); + policy.setMaxConcurrency(maxConcurrency); + policy.setMaxMessagesPerTask(1); + policy.setRejectionLimit(1); + policy.setPeriod(0); + policy.setReceiveTimeout(100); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); for (int i = 0; i < messagesToSend - 110; i++) { channel.send(new DocumentMessage(1, "fast-1." + (i+1))); } - int activeCountAfterFirstBurst = executor.getActiveCount(); + int activeCountAfterFirstBurst = bus.getActiveCountForEndpoint("testEndpoint"); + System.out.println("after-first: " + activeCountAfterFirstBurst); for (int i = 0; i < 10; i++) { channel.send(new DocumentMessage(1, "slow-1." + (i+1))); Thread.sleep(50); } - int activeCountAfterSlowDown = executor.getActiveCount(); + int activeCountAfterSlowDown = bus.getActiveCountForEndpoint("testEndpoint"); + System.out.println("after-slowdown: " + activeCountAfterSlowDown); for (int i = 0; i < 100; i++) { channel.send(new DocumentMessage(1, "fast-2." + (i+1))); } - int activeCountAfterLastBurst = executor.getActiveCount(); + int activeCountAfterLastBurst = bus.getActiveCountForEndpoint("testEndpoint"); + System.out.println("after-last: " + activeCountAfterLastBurst); latch.await(10, TimeUnit.SECONDS); int averageActive = activeSum.get() / messagesToSend; assertTrue(activeCountAfterSlowDown < activeCountAfterFirstBurst); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java similarity index 72% rename from spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java rename to spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java index 5cf4ad2f02..ff548049b6 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedDelayConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.channel.consumer; +package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -24,6 +24,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; @@ -47,10 +49,17 @@ public class FixedDelayConsumerTests { latch.countDown(); } }; - FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint); - consumer.setPollInterval(10); - consumer.initialize(); - consumer.start(); + MessageBus bus = new MessageBus(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setConcurrency(1); + policy.setMaxConcurrency(1); + policy.setMaxMessagesPerTask(1); + policy.setFixedRate(true); + policy.setPeriod(10); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } @@ -70,10 +79,17 @@ public class FixedDelayConsumerTests { latch.countDown(); } }; - FixedDelayConsumer consumer = new FixedDelayConsumer(channel, endpoint); - consumer.setPollInterval(10); - consumer.initialize(); - consumer.start(); + MessageBus bus = new MessageBus(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setConcurrency(1); + policy.setMaxConcurrency(1); + policy.setMaxMessagesPerTask(1); + policy.setFixedRate(true); + policy.setPeriod(10); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java similarity index 74% rename from spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java rename to spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java index 2394d84d37..9a9cac604a 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/channel/consumer/FixedRateConsumerTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.channel.consumer; +package org.springframework.integration.bus; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; @@ -24,6 +24,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; + +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.endpoint.GenericMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; @@ -47,10 +50,14 @@ public class FixedRateConsumerTests { latch.countDown(); } }; - FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint); - consumer.setPollInterval(10); - consumer.initialize(); - consumer.start(); + MessageBus bus = new MessageBus(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setFixedRate(true); + policy.setPeriod(10); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } @@ -70,10 +77,17 @@ public class FixedRateConsumerTests { latch.countDown(); } }; - FixedRateConsumer consumer = new FixedRateConsumer(channel, endpoint); - consumer.setPollInterval(10); - consumer.initialize(); - consumer.start(); + MessageBus bus = new MessageBus(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setConcurrency(1); + policy.setMaxConcurrency(1); + policy.setMaxMessagesPerTask(1); + policy.setFixedRate(true); + policy.setPeriod(10); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); for (int i = 0; i < messagesToSend; i++) { channel.send(new DocumentMessage(1, "test " + (i+1))); } diff --git a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java index f51a19759d..ec7116c6e3 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/endpoint/GenericMessageEndpointTests.java @@ -21,10 +21,11 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; +import org.springframework.integration.bus.ConsumerPolicy; +import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelResolver; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; -import org.springframework.integration.channel.consumer.EventDrivenConsumer; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.DocumentMessage; import org.springframework.integration.message.Message; @@ -47,9 +48,13 @@ public class GenericMessageEndpointTests { endpoint.setSource(channel); endpoint.setHandler(handler); endpoint.setTarget(replyChannel); - EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); - consumer.initialize(); - consumer.start(); + MessageBus bus = new MessageBus(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setPeriod(0); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); DocumentMessage testMessage = new DocumentMessage(1, "test"); channel.send(testMessage); Message reply = replyChannel.receive(10); @@ -78,9 +83,13 @@ public class GenericMessageEndpointTests { endpoint.setSource(channel); endpoint.setHandler(handler); endpoint.setChannelResolver(channelResolver); - EventDrivenConsumer consumer = new EventDrivenConsumer(channel, endpoint); - consumer.initialize(); - consumer.start(); + MessageBus bus = new MessageBus(); + bus.registerChannel("testChannel", channel); + bus.registerEndpoint("testEndpoint", endpoint); + ConsumerPolicy policy = new ConsumerPolicy(); + policy.setPeriod(0); + bus.activateSubscription("testChannel", "testEndpoint", policy); + bus.start(); DocumentMessage testMessage = new DocumentMessage(1, "test"); testMessage.getHeader().setReplyChannelName("replyChannel"); channel.send(testMessage); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml b/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml index 7048ee20fa..cade803fbf 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml +++ b/spring-eai-core/src/test/java/org/springframework/integration/messageBusTests.xml @@ -4,7 +4,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> - +