Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Factored out MessageRetriever and MessageDispatcher strategies.

Browse Source
This commit is contained in:
Mark Fisher
2007-12-25 23:20:00 +00:00
parent 2b26ac53f5
commit ffc6955841
9 changed files with 302 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java Normal file
View File

@@ -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<EndpointExecutor> endpointExecutors = new CopyOnWriteArrayList<EndpointExecutor>();
public AbstractMessageDispatcher(MessageRetriever retriever) {
this.retriever = retriever;
}
public void addEndpointExecutor(EndpointExecutor executor) {
executor.start();
this.endpointExecutors.add(executor);
}
protected List<EndpointExecutor> getEndpointExecutors() {
return this.endpointExecutors;
}
/**
* Receives messages and dispatches to the endpoints. Returns the number of
* messages processed.
*/
public int receiveAndDispatch() {
int messagesProcessed = 0;
Collection<Message<?>> 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);
}

63
spring-integration-core/src/main/java/org/springframework/integration/bus/ChannelPollingMessageRetriever.java Normal file
View File

@@ -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 '<em>maxMessagesPerTask</em>'
* property of the provided {@link ConsumerPolicy}, and the timeout for each
* receive call is determined by the policy's '<em>receiveTimeout</em>'
* 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<Message<?>> retrieveMessages() {
List<Message<?>> messages = new LinkedList<Message<?>>();
while (messages.size() < this.policy.getMaxMessagesPerTask()) {
Message<?> message = this.channel.receive(this.policy.getReceiveTimeout());
if (message == null) {
return messages;
}
messages.add(message);
}
return messages;
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/bus/ConsumerPolicy.java
View File

@@ -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;

2
spring-integration-core/src/main/java/org/springframework/integration/bus/EndpointExecutor.java
View File

@@ -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");
}

65
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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();
}
}

28
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java Normal file
View File

@@ -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();
}

32
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageRetriever.java Normal file
View File

@@ -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<Message<?>> retrieveMessages();
}

88
spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java Normal file
View File

@@ -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<EndpointExecutor> 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;
}
}