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

Replaced UnicastMessageDispatcher with DefaultMessageDispatcher. Provides a 'broadcast' boolean option for publishing to all MessageReceivingExecutors. Default is 'false' for point-to-point messaging to one receiver even among multiple candidates.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-09 20:50:22 +00:00
parent fada237b51
commit ca8ad17908
11 changed files with 513 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java
View File

@@ -21,6 +21,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.bus.ConsumerPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMapper;
@@ -32,7 +33,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class AbstractSourceAdapter<T> implements SourceAdapter, InitializingBean {
public abstract class AbstractSourceAdapter<T> implements SourceAdapter, InitializingBean {
protected Log logger = LogFactory.getLog(this.getClass());
@@ -40,6 +41,8 @@ public class AbstractSourceAdapter<T> implements SourceAdapter, InitializingBean
private MessageMapper<?,T> mapper = new SimplePayloadMessageMapper<T>();
private ConsumerPolicy consumerPolicy;
private long sendTimeout = -1;
@@ -61,6 +64,15 @@ public class AbstractSourceAdapter<T> implements SourceAdapter, InitializingBean
return this.mapper;
}
public void setConsumerPolicy(ConsumerPolicy consumerPolicy) {
Assert.notNull(consumerPolicy, "'consumerPolicy' must not be null");
this.consumerPolicy = consumerPolicy;
}
public ConsumerPolicy getConsumerPolicy() {
return this.consumerPolicy;
}
public final void afterPropertiesSet() {
if (this.channel == null) {
throw new MessagingConfigurationException("'channel' is required");

13
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
View File

@@ -38,31 +38,26 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
private PollableSource<T> source;
private ConsumerPolicy policy = ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD);
public PollingSourceAdapter(PollableSource<T> source) {
Assert.notNull(source, "'source' must not be null");
this.source = source;
this.setConsumerPolicy(ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD));
}
public void setPeriod(int period) {
Assert.isTrue(period > 0, "'period' must be a positive value");
this.policy.setPeriod(period);
this.getConsumerPolicy().setPeriod(period);
}
public void setMaxMessagesPerTask(int maxMessagesPerTask) {
Assert.isTrue(maxMessagesPerTask > 0, "'maxMessagesPerTask' must be a positive value");
this.policy.setMaxMessagesPerTask(maxMessagesPerTask);
}
public ConsumerPolicy getConsumerPolicy() {
return this.policy;
this.getConsumerPolicy().setMaxMessagesPerTask(maxMessagesPerTask);
}
public int dispatch() {
int messagesProcessed = 0;
int limit = this.policy.getMaxMessagesPerTask();
int limit = this.getConsumerPolicy().getMaxMessagesPerTask();
Collection<T> results = this.source.poll(limit);
if (results != null) {
if (results.size() > limit) {

3
spring-integration-core/src/main/java/org/springframework/integration/adapter/SourceAdapter.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.adapter;
import org.springframework.integration.bus.ConsumerPolicy;
import org.springframework.integration.channel.MessageChannel;
/**
@@ -25,6 +26,8 @@ import org.springframework.integration.channel.MessageChannel;
*/
public interface SourceAdapter {
ConsumerPolicy getConsumerPolicy();
void setChannel(MessageChannel channel);
}

5
spring-integration-core/src/main/java/org/springframework/integration/adapter/jms/JmsMessageDrivenSourceAdapter.java
View File

@@ -83,11 +83,6 @@ public class JmsMessageDrivenSourceAdapter extends AbstractSourceAdapter<Object>
this.taskExecutor = taskExecutor;
}
public void setPolicy(ConsumerPolicy policy) {
Assert.notNull(policy, "'policy' must not be null");
this.policy = policy;
}
@Override
public void initialize() {
if (this.container == null) {

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

@@ -24,6 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* Abstract base class for message dispatchers. Delegates to a
@@ -46,6 +47,7 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher {
public void addExecutor(MessageReceivingExecutor executor) {
Assert.notNull(executor, "'executor' must not be null");
executor.start();
this.executors.add(executor);
}

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

@@ -0,0 +1,136 @@
/*
* 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.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.RejectedExecutionException;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.message.Message;
import org.springframework.util.Assert;
/**
* The default implementation of {@link MessageDispatcher}. If
* {@link #broadcast} is set to <code>false</code> (the default), each message
* will be sent to a single {@link MessageReceivingExecutor}. Otherwise, each
* retrieved {@link Message} will be sent to all executors.
*
* @author Mark Fisher
*/
public class DefaultMessageDispatcher extends AbstractMessageDispatcher {
private boolean broadcast = false;
private int rejectionLimit = 5;
private long retryInterval = 1000;
private boolean shouldFailOnRejectionLimit = true;
public DefaultMessageDispatcher(MessageRetriever retriever) {
super(retriever);
}
public void setBroadcast(boolean broadcast) {
this.broadcast = broadcast;
}
public void setRejectionLimit(int rejectionLimit) {
Assert.isTrue(rejectionLimit > 0, "'rejectionLimit' must be at least 1");
this.rejectionLimit = rejectionLimit;
}
public void setRetryInterval(long retryInterval) {
Assert.isTrue(retryInterval > 0, "'retryInterval' must not be negative");
this.retryInterval = retryInterval;
}
/**
* Specify whether an exception should be thrown when this dispatcher's
* {@link #rejectionLimit} is reached. The default value is 'true'.
*/
public void setShouldFailOnRejectionLimit(boolean shouldFailOnRejectionLimit) {
this.shouldFailOnRejectionLimit = shouldFailOnRejectionLimit;
}
@Override
protected boolean dispatchMessage(Message<?> message) {
int attempts = 0;
List<MessageReceivingExecutor> targets = new ArrayList<MessageReceivingExecutor>(this.getExecutors());
while (attempts < this.rejectionLimit) {
if (attempts > 0) {
if (logger.isDebugEnabled()) {
logger.debug("executor(s) rejected message after " + attempts
+ " attempt(s), will try again after 'retryInterval' of " + this.retryInterval
+ " milliseconds");
}
try {
Thread.sleep(this.retryInterval);
}
catch (InterruptedException iex) {
Thread.currentThread().interrupt();
return false;
}
}
Iterator<MessageReceivingExecutor> iter = targets.iterator();
if (!iter.hasNext()) {
if (logger.isWarnEnabled()) {
logger.warn("dispatcher has no active executors");
}
return false;
}
boolean encounteredRejection = false;
while (iter.hasNext()) {
MessageReceivingExecutor executor = iter.next();
if (executor == null || !executor.isRunning()) {
if (logger.isInfoEnabled()) {
logger.info("skipping inactive executor");
}
iter.remove();
continue;
}
try {
executor.processMessage(message);
if (!this.broadcast) {
return true;
}
iter.remove();
if (!iter.hasNext() && !encounteredRejection) {
return true;
}
}
catch (RejectedExecutionException rex) {
encounteredRejection = true;
if (logger.isDebugEnabled()) {
logger.debug("executor rejected task, continuing with other executors if available", rex);
}
}
}
attempts++;
}
if (this.shouldFailOnRejectionLimit) {
throw new MessageDeliveryException("Dispatcher reached rejection limit of " + this.rejectionLimit
+ ". Consider increasing the executor's concurrency and/or raising the 'rejectionLimit'.");
}
return false;
}
}

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

@@ -208,9 +208,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
}
public void registerSourceAdapter(String name, SourceAdapter adapter) {
ConsumerPolicy policy = adapter.getConsumerPolicy();
if (adapter instanceof MessageDispatcher) {
MessageDispatcher dispatcher = (MessageDispatcher) adapter;
ConsumerPolicy policy = dispatcher.getConsumerPolicy();
DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy);
this.addDispatcherTask(dispatcherTask);
}
@@ -230,7 +230,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
MessageChannel channel = adapter.getChannel();
ConsumerPolicy policy = adapter.getConsumerPolicy();
MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy);
UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy);
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever);
dispatcher.setRejectionLimit(policy.getRejectionLimit());
dispatcher.setRetryInterval(policy.getRetryInterval());
MessageReceivingExecutor executor = new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency());
dispatcher.addExecutor(executor);
this.addLifecycleComponent(name + "-executor", executor);
@@ -245,6 +247,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
String channelName = subscription.getChannel();
String endpointName = subscription.getEndpoint();
ConsumerPolicy policy = subscription.getPolicy();
MessageEndpoint<?> endpoint = this.endpoints.get(endpointName);
if (endpoint == null) {
throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'");
}
MessageChannel channel = this.lookupChannel(channelName);
if (channel == null) {
if (this.autoCreateChannels == false) {
@@ -257,10 +263,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
channel = new SimpleChannel();
this.registerChannel(channelName, channel);
}
MessageEndpoint<?> endpoint = this.endpoints.get(endpointName);
if (endpoint == null) {
throw new MessagingException("Cannot activate subscription, unknown endpoint '" + endpointName + "'");
}
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '" + channelName +
"' for endpoint '" + endpointName + "'");
@@ -269,7 +271,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
this.receiverExecutors.put(endpoint, executor);
this.lifecycleComponents.put(endpointName + "-executor", executor);
MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy);
UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy);
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever);
dispatcher.setRejectionLimit(policy.getRejectionLimit());
dispatcher.setRetryInterval(policy.getRetryInterval());
dispatcher.addExecutor(executor);
DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy);
if (this.isRunning()) {

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

@@ -23,8 +23,6 @@ package org.springframework.integration.bus;
*/
public interface MessageDispatcher {
ConsumerPolicy getConsumerPolicy();
int dispatch();
}

95
spring-integration-core/src/main/java/org/springframework/integration/bus/UnicastMessageDispatcher.java
View File

@@ -1,95 +0,0 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.bus;
import java.util.Iterator;
import java.util.concurrent.RejectedExecutionException;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.message.Message;
/**
* A {@link MessageDispatcher} implementation that dispatches each retrieved
* {@link Message} to a single {@link MessageReceivingExecutor}.
*
* @author Mark Fisher
*/
public class UnicastMessageDispatcher extends AbstractMessageDispatcher {
private ConsumerPolicy policy;
public UnicastMessageDispatcher(MessageRetriever retriever, ConsumerPolicy policy) {
super(retriever);
this.policy = policy;
}
public ConsumerPolicy getConsumerPolicy() {
return this.policy;
}
@Override
protected boolean dispatchMessage(Message<?> message) {
int attempts = 0;
while (attempts < policy.getRejectionLimit()) {
if (attempts > 0) {
if (logger.isDebugEnabled()) {
logger.debug("executor(s) rejected message after " + attempts
+ " attempt(s), will try again after 'retryInterval' of " + this.policy.getRetryInterval()
+ " milliseconds");
}
try {
Thread.sleep(policy.getRetryInterval());
}
catch (InterruptedException iex) {
Thread.currentThread().interrupt();
return false;
}
}
Iterator<MessageReceivingExecutor> iter = this.getExecutors().iterator();
if (!iter.hasNext()) {
if (logger.isWarnEnabled()) {
logger.warn("dispatcher has no active executors");
}
return false;
}
while (iter.hasNext()) {
MessageReceivingExecutor executor = iter.next();
if (executor == null || !executor.isRunning()) {
if (logger.isInfoEnabled()) {
logger.info("skipping inactive executor");
}
continue;
}
try {
executor.processMessage(message);
return true;
}
catch (RejectedExecutionException rex) {
if (logger.isDebugEnabled()) {
logger.debug("executor rejected task, continuing with other executors if available", rex);
}
}
}
attempts++;
}
throw new MessageDeliveryException("Dispatcher reached rejection limit of " +
this.policy.getRejectionLimit() + ". Consider increasing the concurrency and/or raising the limit.");
}
}