From ffc695584125906ab51caa51c4d30d9ba814523c Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 25 Dec 2007 23:20:00 +0000 Subject: [PATCH] Factored out MessageRetriever and MessageDispatcher strategies. --- .../bus/AbstractMessageDispatcher.java | 78 ++++++++++++++++ .../bus/ChannelPollingMessageRetriever.java | 63 +++++++++++++ .../integration/bus/ConsumerPolicy.java | 2 +- .../integration/bus/EndpointExecutor.java | 2 +- .../integration/bus/MessageBus.java | 65 +++----------- .../integration/bus/MessageDispatcher.java | 28 ++++++ .../integration/bus/MessageRetriever.java | 32 +++++++ .../bus/UnicastMessageDispatcher.java | 88 +++++++++++++++++++ ...eEndpointAnnotationPostProcessorTests.java | 1 + 9 files changed, 302 insertions(+), 57 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/bus/MessageRetriever.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java 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 new file mode 100644 index 0000000000..5ac976386d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java @@ -0,0 +1,78 @@ +/* + * 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.integration.message.Message; + +/** + * 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 endpointExecutors = new CopyOnWriteArrayList(); + + + public AbstractMessageDispatcher(MessageRetriever retriever) { + this.retriever = retriever; + } + + + public void addEndpointExecutor(EndpointExecutor executor) { + executor.start(); + this.endpointExecutors.add(executor); + } + + protected List getEndpointExecutors() { + return this.endpointExecutors; + } + + /** + * Receives messages and dispatches to the endpoints. Returns the number of + * messages processed. + */ + public int receiveAndDispatch() { + 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/ChannelPollingMessageRetriever.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java new file mode 100644 index 0000000000..2dc262b91c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java @@ -0,0 +1,63 @@ +/* + * 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.LinkedList; +import java.util.List; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; + +/** + * 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. In general, it is recommended to use a value of 1 for + * 'maxMessagesPerTask' whenever a non-zero timeout is provided. Otherwise the + * retriever may be holding on to available messages while waiting for + * additional messages. + * + * @author Mark Fisher + */ +public class ChannelPollingMessageRetriever implements MessageRetriever { + + private MessageChannel channel; + + private ConsumerPolicy policy; + + + public ChannelPollingMessageRetriever(MessageChannel channel, ConsumerPolicy policy) { + this.channel = channel; + this.policy = policy; + } + + + public Collection> retrieveMessages() { + List> messages = new LinkedList>(); + while (messages.size() < this.policy.getMaxMessagesPerTask()) { + Message message = this.channel.receive(this.policy.getReceiveTimeout()); + if (message == null) { + return messages; + } + messages.add(message); + } + return messages; + } + +} 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 index c4ba7668a3..ed91ac25d4 100644 --- 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 @@ -29,7 +29,7 @@ public class ConsumerPolicy { private static final int DEFAULT_MAX_CONCURRENCY = 10; - private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 10; + private static final int DEFAULT_MAX_MESSAGES_PER_TASK = 1; private static final int DEFAULT_REJECTION_LIMIT = 10; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java index fc08319418..1594a77478 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java @@ -101,7 +101,7 @@ public class EndpointExecutor implements Lifecycle { } } - public void executeTask(Message message) { + public void processMessage(Message message) { if (threadPoolExecutor == null) { throw new MessageHandlingException("executor is not running"); } 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 b79f5373b9..6afe4df2fe 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.RejectedExecutionException; import java.util.concurrent.ScheduledThreadPoolExecutor; import org.apache.commons.logging.Log; @@ -36,7 +35,6 @@ import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PointToPointChannel; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.endpoint.MessageEndpoint; -import org.springframework.integration.message.Message; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -178,7 +176,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } EndpointExecutor endpointExecutor = new EndpointExecutor(endpoint, policy.getConcurrency(), policy.getMaxConcurrency()); endpointExecutors.put(endpoint, endpointExecutor); - DispatcherTask dispatcherTask = new DispatcherTask(channel, endpoint, policy); + MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); + UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); + dispatcher.addEndpointExecutor(endpointExecutor); + DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); this.dispatcherTasks.add(dispatcherTask); if (this.logger.isInfoEnabled()) { logger.info("registered dispatcher task: channel='" + @@ -260,73 +261,27 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } - private class DispatcherTask implements Runnable { + private static class DispatcherTask implements Runnable { - private MessageChannel channel; - - private MessageEndpoint endpoint; + private MessageDispatcher dispatcher; private ConsumerPolicy policy; - public DispatcherTask(MessageChannel channel, MessageEndpoint endpoint, ConsumerPolicy policy) { - this.channel = channel; - this.endpoint = endpoint; + public DispatcherTask(MessageDispatcher dispatcher, ConsumerPolicy policy) { + this.dispatcher = dispatcher; 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.executeTask(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(); - } - } - } - } - } - } + dispatcher.receiveAndDispatch(); } + } 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/bus/MessageDispatcher.java new file mode 100644 index 0000000000..6d37db2f45 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java @@ -0,0 +1,28 @@ +/* + * 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; + +/** + * Strategy interface for dispatching messages. + * + * @author Mark Fisher + */ +public interface MessageDispatcher { + + int receiveAndDispatch(); + +} 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/bus/MessageRetriever.java new file mode 100644 index 0000000000..f728cdd3a9 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageRetriever.java @@ -0,0 +1,32 @@ +/* + * 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 org.springframework.integration.message.Message; + +/** + * Strategy interface for retrieving messages. + * + * @author Mark Fisher + */ +public interface MessageRetriever { + + Collection> retrieveMessages(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java new file mode 100644 index 0000000000..59c738d235 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java @@ -0,0 +1,88 @@ +/* + * 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.Iterator; +import java.util.concurrent.RejectedExecutionException; + +import org.springframework.integration.message.Message; + +/** + * A {@link MessageDispatcher} implementation that dispatches each retrieved + * {@link Message} to a single {@link EndpointExecutor}. + * + * @author Mark Fisher + */ +public class UnicastMessageDispatcher extends AbstractMessageDispatcher { + + private ConsumerPolicy policy; + + + public UnicastMessageDispatcher(MessageRetriever retriever, ConsumerPolicy policy) { + super(retriever); + this.policy = policy; + } + + + @Override + protected boolean dispatchMessage(Message message) { + int attempts = 0; + Iterator iter = this.getEndpointExecutors().iterator(); + if (!iter.hasNext()) { + if (logger.isWarnEnabled()) { + logger.warn("dispatcher has no active endpoint executors"); + } + return false; + } + while (iter.hasNext()) { + EndpointExecutor executor = iter.next(); + try { + if (executor == null || !executor.isRunning()) { + if (logger.isInfoEnabled()) { + logger.info("removing inactive endpoint executor"); + } + iter.remove(); + continue; + } + executor.processMessage(message); + return 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(); + } + } + } + catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn("error occurred during dispatch", e); + } + } + } + return false; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index f0d3b415d9..824ec05c62 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java @@ -39,6 +39,7 @@ public class MessageEndpointAnnotationPostProcessorTests { inputChannel.send(new GenericMessage(1, "world")); Message message = outputChannel.receive(); assertEquals("hello world", message.getPayload()); + context.stop(); } }