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

Refactored MessageConsumer with onMessage to MessageHandler with handleMessage.

Browse Source
This commit is contained in:
Mark Fisher
2008-11-03 14:17:53 +00:00
parent 90dcb7e88b
commit 39a8486ac8
101 changed files with 745 additions and 741 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java
View File

@@ -24,7 +24,7 @@ import org.springframework.util.Assert;
/**
* A base class for aggregating a group of Messages into a single Message.
* Extends {@link AbstractMessageBarrierConsumer} and waits for a
* Extends {@link AbstractMessageBarrierHandler} and waits for a
* <em>complete</em> group of {@link Message Messages} to arrive. Subclasses
* must provide the implementation of the {@link #aggregateMessages(List)}
* method to combine the group of Messages into a single {@link Message}.
@@ -34,13 +34,13 @@ import org.springframework.util.Assert;
* custom implementation of the {@link CompletionStrategy} may be provided.
*
* <p>All considerations regarding <code>timeout</code> and grouping by
* <code>correlationId</code> from {@link AbstractMessageBarrierConsumer}
* <code>correlationId</code> from {@link AbstractMessageBarrierHandler}
* apply here as well.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class AbstractMessageAggregator extends AbstractMessageBarrierConsumer {
public abstract class AbstractMessageAggregator extends AbstractMessageBarrierHandler {
private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();

16
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
View File

@@ -30,10 +30,10 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.consumer.AbstractMessageConsumer;
import org.springframework.integration.consumer.AbstractMessageHandler;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.TaskScheduler;
@@ -43,13 +43,13 @@ import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
/**
* Base class for {@link MessageBarrier}-based Message Consumers. A
* {@link MessageConsumer} implementation that waits for a group of
* Base class for {@link MessageBarrier}-based Message Handlers. A
* {@link MessageHandler} implementation that waits for a group of
* {@link Message Messages} to arrive and processes them together. Uses a
* {@link MessageBarrier} to store messages and to decide how the messages
* should be released.
* <p>
* Each {@link Message} that is received by this consumer will be associated
* Each {@link Message} that is received by this handler will be associated
* with a group based upon the '<code>correlationId</code>' property of its
* header. If no such property is available, a {@link MessageHandlingException}
* will be thrown.
@@ -64,7 +64,7 @@ import org.springframework.util.ObjectUtils;
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class AbstractMessageBarrierConsumer extends AbstractMessageConsumer implements TaskSchedulerAware, InitializingBean {
public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandler implements TaskSchedulerAware, InitializingBean {
public final static long DEFAULT_SEND_TIMEOUT = 1000;
@@ -101,7 +101,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractMessageCons
private ScheduledFuture<?> reaperFutureTask;
public AbstractMessageBarrierConsumer() {
public AbstractMessageBarrierHandler() {
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
@@ -184,7 +184,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractMessageCons
}
@Override
protected final void onMessageInternal(Message<?> message) {
protected final void handleMessageInternal(Message<?> message) {
if (!this.initialized) {
this.afterPropertiesSet();
}

6
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -22,7 +22,7 @@ import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageHeaders;
/**
* An {@link AbstractMessageBarrierConsumer} that waits for a group of
* An {@link AbstractMessageBarrierHandler} that waits for a group of
* {@link Message Messages} to arrive and re-sends them in order, sorted
* by their <code>sequenceNumber</code>.
* <p>
@@ -30,12 +30,12 @@ import org.springframework.integration.core.MessageHeaders;
* wait for the whole sequence to arrive before re-sending them.
* <p>
* All considerations regarding <code>timeout</code> and grouping by
* '<code>correlationId</code>' from {@link AbstractMessageBarrierConsumer}
* '<code>correlationId</code>' from {@link AbstractMessageBarrierHandler}
* apply here as well.
*
* @author Marius Bogoevici
*/
public class Resequencer extends AbstractMessageBarrierConsumer {
public class Resequencer extends AbstractMessageBarrierHandler {
private volatile boolean releasePartialSequences = true;

12
org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java
View File

@@ -19,12 +19,12 @@ package org.springframework.integration.channel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
* Base implementation of {@link MessageChannel} that invokes the subscribed
* {@link MessageConsumer consumer(s)} by delegating to a {@link MessageDispatcher}.
* {@link MessageHandler handler(s)} by delegating to a {@link MessageDispatcher}.
*
* @author Mark Fisher
*/
@@ -43,12 +43,12 @@ public class AbstractSubscribableChannel<T extends MessageDispatcher> extends Ab
return this.dispatcher;
}
public boolean subscribe(MessageConsumer consumer) {
return this.dispatcher.addConsumer(consumer);
public boolean subscribe(MessageHandler handler) {
return this.dispatcher.addHandler(handler);
}
public boolean unsubscribe(MessageConsumer consumer) {
return this.dispatcher.removeConsumer(consumer);
public boolean unsubscribe(MessageHandler handle) {
return this.dispatcher.removeHandler(handle);
}
@Override

10
org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannel.java
View File

@@ -17,7 +17,7 @@
package org.springframework.integration.channel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
/**
* Interface for any MessageChannel implementation that accepts subscribers.
@@ -27,13 +27,13 @@ import org.springframework.integration.message.MessageConsumer;
public interface SubscribableChannel extends MessageChannel {
/**
* Register a {@link MessageConsumer} as a subscriber to this channel.
* Register a {@link MessageHandler} as a subscriber to this channel.
*/
boolean subscribe(MessageConsumer consumer);
boolean subscribe(MessageHandler handler);
/**
* Remove a {@link MessageConsumer} from the subscribers of this channel.
* Remove a {@link MessageHandler} from the subscribers of this channel.
*/
boolean unsubscribe(MessageConsumer consumer);
boolean unsubscribe(MessageHandler handler);
}

14
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractConsumerFactoryBean.java
View File

@@ -17,9 +17,9 @@
package org.springframework.integration.config;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.consumer.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.consumer.AbstractReplyProducingMessageHandler;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
@@ -29,7 +29,7 @@ import org.springframework.util.Assert;
*/
public abstract class AbstractConsumerFactoryBean implements FactoryBean {
private volatile MessageConsumer consumer;
private volatile MessageHandler consumer;
private volatile Object targetObject;
@@ -59,8 +59,8 @@ public abstract class AbstractConsumerFactoryBean implements FactoryBean {
this.initializeConsumer();
Assert.notNull(this.consumer, "failed to create MessageConsumer");
if (this.outputChannel != null
&& this.consumer instanceof AbstractReplyProducingMessageConsumer) {
((AbstractReplyProducingMessageConsumer) this.consumer).setOutputChannel(this.outputChannel);
&& this.consumer instanceof AbstractReplyProducingMessageHandler) {
((AbstractReplyProducingMessageHandler) this.consumer).setOutputChannel(this.outputChannel);
}
}
return this.consumer;
@@ -70,7 +70,7 @@ public abstract class AbstractConsumerFactoryBean implements FactoryBean {
if (this.consumer != null) {
return this.consumer.getClass();
}
return MessageConsumer.class;
return MessageHandler.class;
}
public boolean isSingleton() {
@@ -90,6 +90,6 @@ public abstract class AbstractConsumerFactoryBean implements FactoryBean {
/**
* Subclasses must implement this method to create the MessageConsumer.
*/
protected abstract MessageConsumer createConsumer(Object targetObject, String targetMethodName);
protected abstract MessageHandler createConsumer(Object targetObject, String targetMethodName);
}

15
org.springframework.integration/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -29,7 +29,7 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumerEndpoint;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
@@ -41,7 +41,7 @@ import org.springframework.util.Assert;
*/
public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware, InitializingBean {
private final MessageConsumer consumer;
private final MessageHandler handler;
private volatile String beanName;
@@ -68,9 +68,9 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
private final Object initializationMonitor = new Object();
public ConsumerEndpointFactoryBean(MessageConsumer consumer) {
Assert.notNull(consumer, "consumer must not be null");
this.consumer = consumer;
public ConsumerEndpointFactoryBean(MessageHandler handler) {
Assert.notNull(handler, "handler must not be null");
this.handler = handler;
}
@@ -146,15 +146,14 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAwar
if (channel instanceof SubscribableChannel) {
Assert.isNull(trigger, "A trigger should not be specified for endpoint '" + this.beanName
+ "', since '" + this.inputChannelName + "' is a SubscribableChannel (not pollable).");
this.endpoint = new SubscribingConsumerEndpoint(
this.consumer, (SubscribableChannel) channel);
this.endpoint = new SubscribingConsumerEndpoint((SubscribableChannel) channel, this.handler);
}
else if (channel instanceof PollableChannel) {
if (this.trigger == null) {
this.trigger = new IntervalTrigger(0);
}
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
this.consumer, (PollableChannel) channel);
(PollableChannel) channel, this.handler);
pollingEndpoint.setTrigger(this.trigger);
pollingEndpoint.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
pollingEndpoint.setReceiveTimeout(this.receiveTimeout);

4
org.springframework.integration/src/main/java/org/springframework/integration/config/RouterFactoryBean.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.config;
import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.util.Assert;
@@ -45,7 +45,7 @@ public class RouterFactoryBean extends AbstractConsumerFactoryBean {
}
@Override
protected MessageConsumer createConsumer(Object targetObject, String targetMethodName) {
protected MessageHandler createConsumer(Object targetObject, String targetMethodName) {
Assert.notNull(targetObject, "target object must not be null");
AbstractMessageRouter router = this.createRouter(targetObject, targetMethodName);
if (this.defaultOutputChannel != null) {

4
org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.config;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.splitter.AbstractMessageSplitter;
import org.springframework.integration.splitter.DefaultMessageSplitter;
import org.springframework.integration.splitter.MethodInvokingSplitter;
@@ -31,7 +31,7 @@ import org.springframework.util.StringUtils;
public class SplitterFactoryBean extends AbstractConsumerFactoryBean {
@Override
protected MessageConsumer createConsumer(Object targetObject, String targetMethodName) {
protected MessageHandler createConsumer(Object targetObject, String targetMethodName) {
if (targetObject == null) {
Assert.isTrue(!StringUtils.hasText(targetMethodName),
"'method' should only be provided when 'ref' is also provided");

22
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -33,7 +33,7 @@ import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumerEndpoint;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -60,9 +60,9 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
public Object postProcess(Object bean, String beanName, Method method, T annotation) {
MessageConsumer consumer = this.createConsumer(bean, method, annotation);
MessageHandler handler = this.createHandler(bean, method, annotation);
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
MessageEndpoint endpoint = this.createEndpoint(consumer, annotation, pollerAnnotation);
MessageEndpoint endpoint = this.createEndpoint(handler, annotation, pollerAnnotation);
if (endpoint != null) {
if (endpoint instanceof InitializingBean) {
try {
@@ -75,14 +75,14 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
return endpoint;
}
return consumer;
return handler;
}
protected boolean shouldCreateEndpoint(T annotation) {
return (StringUtils.hasText((String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE)));
}
private MessageEndpoint createEndpoint(MessageConsumer consumer, T annotation, Poller pollerAnnotation) {
private MessageEndpoint createEndpoint(MessageHandler handler, T annotation, Poller pollerAnnotation) {
MessageEndpoint endpoint = null;
String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(inputChannelName)) {
@@ -90,7 +90,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
Assert.notNull(inputChannel, "failed to resolve inputChannel '" + inputChannelName + "'");
if (inputChannel instanceof PollableChannel) {
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
consumer, (PollableChannel) inputChannel);
(PollableChannel) inputChannel, handler);
if (pollerAnnotation != null) {
AnnotationConfigUtils.configurePollingEndpointWithPollerAnnotation(
pollingEndpoint, pollerAnnotation, this.beanFactoryAccessor.getBeanFactory());
@@ -100,21 +100,21 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
else if (inputChannel instanceof SubscribableChannel) {
Assert.isTrue(pollerAnnotation == null,
"The @Poller annotation should only be provided for a PollableChannel");
endpoint = new SubscribingConsumerEndpoint(consumer, (SubscribableChannel) inputChannel);
endpoint = new SubscribingConsumerEndpoint((SubscribableChannel) inputChannel, handler);
}
else {
throw new IllegalArgumentException("unsupported channel type: [" + inputChannel.getClass() + "]");
}
if (consumer instanceof BeanFactoryAware) {
((BeanFactoryAware) consumer).setBeanFactory(this.beanFactoryAccessor.getBeanFactory());
if (handler instanceof BeanFactoryAware) {
((BeanFactoryAware) handler).setBeanFactory(this.beanFactoryAccessor.getBeanFactory());
}
}
return endpoint;
}
/**
* Subclasses must implement this method to create the MessageConsumer.
* Subclasses must implement this method to create the MessageHandler.
*/
protected abstract MessageConsumer createConsumer(Object bean, Method method, T annotation);
protected abstract MessageHandler createHandler(Object bean, Method method, T annotation);
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -27,7 +27,7 @@ import org.springframework.integration.aggregator.MethodInvokingAggregator;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -45,7 +45,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
@Override
protected MessageConsumer createConsumer(Object bean, Method method, Aggregator annotation) {
protected MessageHandler createHandler(Object bean, Method method, Aggregator annotation) {
MethodInvokingAggregator aggregator = new MethodInvokingAggregator(bean, method);
this.configureCompletionStrategy(bean, aggregator);
String discardChannelName = annotation.discardChannel();

12
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -28,7 +28,7 @@ import org.springframework.integration.channel.ChannelResolver;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.consumer.MethodInvokingConsumer;
import org.springframework.integration.consumer.MethodInvokingMessageHandler;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.PollingConsumerEndpoint;
@@ -71,8 +71,8 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
endpoint = this.createInboundChannelAdapter(source, channel, pollerAnnotation);
}
else if (method.getParameterTypes().length > 0 && !hasReturnValue(method)) {
MethodInvokingConsumer consumer = new MethodInvokingConsumer(bean, method);
endpoint = this.createOutboundChannelAdapter(consumer, channel, pollerAnnotation);
MethodInvokingMessageHandler handler = new MethodInvokingMessageHandler(bean, method);
endpoint = this.createOutboundChannelAdapter(channel, handler, pollerAnnotation);
}
else {
throw new IllegalArgumentException("The @ChannelAdapter can only be applied to methods"
@@ -110,17 +110,17 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
return adapter;
}
private MessageEndpoint createOutboundChannelAdapter(MethodInvokingConsumer consumer, MessageChannel channel, Poller pollerAnnotation) {
private MessageEndpoint createOutboundChannelAdapter(MessageChannel channel, MethodInvokingMessageHandler handler, Poller pollerAnnotation) {
if (channel instanceof PollableChannel) {
Trigger trigger = (pollerAnnotation != null)
? AnnotationConfigUtils.parseTriggerFromPollerAnnotation(pollerAnnotation)
: new IntervalTrigger(0);
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(consumer, (PollableChannel) channel);
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint((PollableChannel) channel, handler);
endpoint.setTrigger(trigger);
return endpoint;
}
if (channel instanceof SubscribableChannel) {
return new SubscribingConsumerEndpoint(consumer, (SubscribableChannel) channel);
return new SubscribingConsumerEndpoint((SubscribableChannel) channel, handler);
}
return null;
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java
View File

@@ -21,7 +21,7 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.router.MethodInvokingRouter;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -39,7 +39,7 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP
@Override
protected MessageConsumer createConsumer(Object bean, Method method, Router annotation) {
protected MessageHandler createHandler(Object bean, Method method, Router annotation) {
MethodInvokingRouter router = new MethodInvokingRouter(bean, method);
router.setChannelResolver(this.channelResolver);
String defaultOutputChannelName = annotation.defaultOutputChannel();

8
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java
View File

@@ -20,8 +20,8 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.consumer.ServiceActivatingConsumer;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.consumer.ServiceActivatingHandler;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.StringUtils;
/**
@@ -37,8 +37,8 @@ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnot
@Override
protected MessageConsumer createConsumer(Object bean, Method method, ServiceActivator annotation) {
ServiceActivatingConsumer serviceActivator = new ServiceActivatingConsumer(bean, method);
protected MessageHandler createHandler(Object bean, Method method, ServiceActivator annotation) {
ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(bean, method);
String outputChannelName = annotation.outputChannel();
if (StringUtils.hasText(outputChannelName)) {
serviceActivator.setOutputChannel(this.channelResolver.resolveChannelName(outputChannelName));

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java
View File

@@ -20,7 +20,7 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.splitter.MethodInvokingSplitter;
import org.springframework.util.StringUtils;
@@ -37,7 +37,7 @@ public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPos
@Override
protected MessageConsumer createConsumer(Object bean, Method method, Splitter annotation) {
protected MessageHandler createHandler(Object bean, Method method, Splitter annotation) {
MethodInvokingSplitter splitter = new MethodInvokingSplitter(bean, method);
String outputChannelName = annotation.outputChannel();
if (StringUtils.hasText(outputChannelName)) {

12
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java
View File

@@ -20,9 +20,9 @@ import java.lang.reflect.Method;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.transformer.MethodInvokingTransformer;
import org.springframework.integration.transformer.MessageTransformingConsumer;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.util.StringUtils;
/**
@@ -38,14 +38,14 @@ public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotation
@Override
protected MessageConsumer createConsumer(Object bean, Method method, Transformer annotation) {
protected MessageHandler createHandler(Object bean, Method method, Transformer annotation) {
MethodInvokingTransformer transformer = new MethodInvokingTransformer(bean, method);
MessageTransformingConsumer consumer = new MessageTransformingConsumer(transformer);
MessageTransformingHandler handler = new MessageTransformingHandler(transformer);
String outputChannelName = annotation.outputChannel();
if (StringUtils.hasText(outputChannelName)) {
consumer.setOutputChannel(this.channelResolver.resolveChannelName(outputChannelName));
handler.setOutputChannel(this.channelResolver.resolveChannelName(outputChannelName));
}
return consumer;
return handler;
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractTransformerParser.java
View File

@@ -22,7 +22,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.transformer.Transformer;
import org.springframework.integration.transformer.MessageTransformingConsumer;
import org.springframework.integration.transformer.MessageTransformingHandler;
/**
* @author Mark Fisher
@@ -31,7 +31,7 @@ public abstract class AbstractTransformerParser extends AbstractConsumerEndpoint
@Override
protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageTransformingConsumer.class);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageTransformingHandler.class);
BeanDefinitionBuilder transformerBuilder =
BeanDefinitionBuilder.genericBeanDefinition(this.getTransformerClass());
this.parseTransformer(element, parserContext, transformerBuilder);

4
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/MethodInvokingOutboundChannelAdapterParser.java
View File

@@ -22,7 +22,7 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.consumer.MethodInvokingConsumer;
import org.springframework.integration.consumer.MethodInvokingMessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -46,7 +46,7 @@ public class MethodInvokingOutboundChannelAdapterParser extends AbstractOutbound
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder invokerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingConsumer.class);
BeanDefinitionBuilder invokerBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageHandler.class);
invokerBuilder.addConstructorArgReference(element.getAttribute("ref"));
invokerBuilder.addConstructorArgValue(element.getAttribute("method"));
return invokerBuilder.getBeanDefinition();

4
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java
View File

@@ -20,7 +20,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.consumer.ServiceActivatingConsumer;
import org.springframework.integration.consumer.ServiceActivatingHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -33,7 +33,7 @@ public class ServiceActivatorParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ServiceActivatingConsumer.class);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ServiceActivatingHandler.class);
String ref = element.getAttribute(REF_ATTRIBUTE);
Assert.hasText(ref, "The '" + REF_ATTRIBUTE + "' attribute is required.");
builder.addConstructorArgReference(ref);

4
org.springframework.integration/src/main/java/org/springframework/integration/config/xml/TransformerParser.java
View File

@@ -21,7 +21,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.transformer.MethodInvokingTransformer;
import org.springframework.integration.transformer.MessageTransformingConsumer;
import org.springframework.integration.transformer.MessageTransformingHandler;
/**
* Parser for the &lt;transformer/&gt; element.
@@ -32,7 +32,7 @@ public class TransformerParser extends AbstractConsumerEndpointParser {
@Override
protected BeanDefinitionBuilder parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageTransformingConsumer.class);
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageTransformingHandler.class);
builder.addConstructorArgReference(this.parseAdapter(element, parserContext, MethodInvokingTransformer.class));
return builder;
}

16
org.springframework.integration/src/main/java/org/springframework/integration/consumer/AbstractMessageConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/consumer/AbstractMessageHandler.java
View File

@@ -21,41 +21,41 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
/**
* Base class for MessageConsumer implementations that provides basic
* Base class for MessageHandler implementations that provides basic
* validation and error handling capabilities. Asserts that the incoming
* Message is not null and that it does not contain a null payload. Converts
* checked exceptions into runtime {@link MessagingException}s.
*
* @author Mark Fisher
*/
public abstract class AbstractMessageConsumer implements MessageConsumer {
public abstract class AbstractMessageHandler implements MessageHandler {
protected final Log logger = LogFactory.getLog(this.getClass());
public final void onMessage(Message<?> message) {
public final void handleMessage(Message<?> message) {
Assert.notNull(message == null, "Message must not be null");
Assert.notNull(message.getPayload(), "Message payload must not be null");
if (this.logger.isDebugEnabled()) {
this.logger.debug("consumer '" + this + "' received message: " + message);
this.logger.debug(this + " received message: " + message);
}
try {
this.onMessageInternal(message);
this.handleMessageInternal(message);
}
catch (Exception e) {
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
throw new MessageHandlingException(message,
"error occurred in consumer [" + this + "]", e);
"error occurred in message handler [" + this + "]", e);
}
}
protected abstract void onMessageInternal(Message<?> message) throws Exception;
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
}

16
org.springframework.integration/src/main/java/org/springframework/integration/consumer/AbstractReplyProducingMessageConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/consumer/AbstractReplyProducingMessageHandler.java
View File

@@ -32,11 +32,11 @@ import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**
* Base class for MessageConsumers that are capable of producing replies.
* Base class for MessageHandlers that are capable of producing replies.
*
* @author Mark Fisher
*/
public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer implements BeanFactoryAware {
public abstract class AbstractReplyProducingMessageHandler extends AbstractMessageHandler implements BeanFactoryAware {
public static final long DEFAULT_SEND_TIMEOUT = 1000;
@@ -52,7 +52,7 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
private final MessageChannelTemplate channelTemplate;
public AbstractReplyProducingMessageConsumer() {
public AbstractReplyProducingMessageHandler() {
this.channelTemplate = new MessageChannelTemplate();
this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
}
@@ -93,15 +93,15 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
}
@Override
protected final void onMessageInternal(Message<?> message) {
protected final void handleMessageInternal(Message<?> message) {
if (!this.supports(message)) {
throw new MessageRejectedException(message, "unsupported message");
}
ReplyMessageHolder replyMessageHolder = new ReplyMessageHolder();
this.onMessage(message, replyMessageHolder);
this.handleRequestMessage(message, replyMessageHolder);
if (replyMessageHolder.isEmpty()) {
if (this.requiresReply) {
throw new MessageHandlingException(message, "consumer '" + this
throw new MessageHandlingException(message, "handler '" + this
+ "' requires a reply, but no reply was received");
}
return;
@@ -121,12 +121,12 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess
}
}
protected abstract void onMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder);
protected abstract void handleRequestMessage(Message<?> requestMessage, ReplyMessageHolder replyMessageHolder);
protected boolean supports(Message<?> message) {
if (this.selector != null && !this.selector.accept(message)) {
if (logger.isDebugEnabled()) {
logger.debug("selector for consumer '" + this + "' rejected message: " + message);
logger.debug("selector for handler '" + this + "' rejected message: " + message);
}
return false;
}

18
org.springframework.integration/src/main/java/org/springframework/integration/consumer/MethodInvokingConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/consumer/MethodInvokingMessageHandler.java
View File

@@ -20,33 +20,33 @@ import java.lang.reflect.Method;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessagingException;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageMappingMethodInvoker;
import org.springframework.util.Assert;
/**
* A {@link MessageConsumer} that invokes the specified method on the provided object.
* A {@link MessageHandler} that invokes the specified method on the provided object.
*
* @author Mark Fisher
*/
public class MethodInvokingConsumer extends MessageMappingMethodInvoker implements MessageConsumer {
public class MethodInvokingMessageHandler extends MessageMappingMethodInvoker implements MessageHandler {
public MethodInvokingConsumer(Object object, Method method) {
public MethodInvokingMessageHandler(Object object, Method method) {
super(object, method);
Assert.isTrue(method.getReturnType().equals(void.class),
"MethodInvokingConsumer requires a void-returning method");
"MethodInvokingMessageHandler requires a void-returning method");
}
public MethodInvokingConsumer(Object object, String methodName) {
public MethodInvokingMessageHandler(Object object, String methodName) {
super(object, methodName);
}
public void onMessage(Message<?> message) {
public void handleMessage(Message<?> message) {
Object result = this.invokeMethod(message);
if (result != null) {
throw new MessagingException(message, "the MethodInvokingConsumer method must have a void return, "
+ "but '" + this + "' received a value: [" + result + "]");
throw new MessagingException(message, "the MethodInvokingMessageHandler method must "
+ "have a void return, but '" + this + "' received a value: [" + result + "]");
}
}

10
org.springframework.integration/src/main/java/org/springframework/integration/consumer/ServiceActivatingConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/consumer/ServiceActivatingHandler.java
View File

@@ -31,14 +31,14 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class ServiceActivatingConsumer extends AbstractReplyProducingMessageConsumer implements InitializingBean {
public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandler implements InitializingBean {
private final MethodResolver methodResolver = new DefaultMethodResolver(ServiceActivator.class);
private final MethodInvoker invoker;
public ServiceActivatingConsumer(final Object object) {
public ServiceActivatingHandler(final Object object) {
Assert.notNull(object, "object must not be null");
Method method = this.methodResolver.findMethod(object);
Assert.notNull(method, "unable to resolve ServiceActivator method on target class ["
@@ -46,11 +46,11 @@ public class ServiceActivatingConsumer extends AbstractReplyProducingMessageCons
this.invoker = new MessageMappingMethodInvoker(object, method);
}
public ServiceActivatingConsumer(Object object, Method method) {
public ServiceActivatingHandler(Object object, Method method) {
this.invoker = new MessageMappingMethodInvoker(object, method);
}
public ServiceActivatingConsumer(Object object, String methodName) {
public ServiceActivatingHandler(Object object, String methodName) {
this.invoker = new MessageMappingMethodInvoker(object, methodName);
}
@@ -62,7 +62,7 @@ public class ServiceActivatingConsumer extends AbstractReplyProducingMessageCons
}
@Override
protected void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
try {
Object result = this.invoker.invokeMethod(message);
if (result != null) {

22
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -24,7 +24,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageRejectedException;
/**
@@ -36,21 +36,21 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
protected final Log logger = LogFactory.getLog(this.getClass());
protected final Set<MessageConsumer> consumers = new CopyOnWriteArraySet<MessageConsumer>();
protected final Set<MessageHandler> handlers = new CopyOnWriteArraySet<MessageHandler>();
private volatile TaskExecutor taskExecutor;
public boolean addConsumer(MessageConsumer consumer) {
return this.consumers.add(consumer);
public boolean addHandler(MessageHandler handler) {
return this.handlers.add(handler);
}
public boolean removeConsumer(MessageConsumer consumer) {
return this.consumers.remove(consumer);
public boolean removeHandler(MessageHandler handler) {
return this.handlers.remove(handler);
}
/**
* Specify a {@link TaskExecutor} for invoking the consumers.
* Specify a {@link TaskExecutor} for invoking the handlers.
* If none is provided, the invocation will occur in the thread
* that runs this polling dispatcher.
*/
@@ -63,21 +63,21 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
}
public String toString() {
return this.getClass().getSimpleName() + " with consumers: " + this.consumers;
return this.getClass().getSimpleName() + " with handlers: " + this.handlers;
}
/**
* Convenience method available for subclasses. Returns 'true' unless a
* "Selective Consumer" throws a {@link MessageRejectedException}.
*/
protected boolean sendMessageToConsumer(Message<?> message, MessageConsumer consumer) {
protected boolean sendMessageToHandler(Message<?> message, MessageHandler handler) {
try {
consumer.onMessage(message);
handler.handleMessage(message);
return true;
}
catch (MessageRejectedException e) {
if (logger.isDebugEnabled()) {
logger.debug("Consumer '" + consumer + "' rejected Message, continuing with other consumers if available.", e);
logger.debug("Handler '" + handler + "' rejected Message, continuing with other handlers if available.", e);
}
return false;
}

18
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -19,13 +19,13 @@ package org.springframework.integration.dispatcher;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
/**
* A broadcasting dispatcher implementation. It makes a best effort to
* send the message to each of its consumers. If it fails to send to any
* one consumer, it will log a warn-level message but continue to send
* to the other consumers.
* send the message to each of its handlers. If it fails to send to any
* one handler, it will log a warn-level message but continue to send
* to the other handlers.
*
* @author Mark Fisher
*/
@@ -36,7 +36,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
/**
* Specify whether to apply sequence numbers to the messages
* prior to sending to the endpoints. By default, sequence
* prior to sending to the handlers. By default, sequence
* numbers will <em>not</em> be applied
*/
public void setApplySequence(boolean applySequence) {
@@ -45,8 +45,8 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
public boolean dispatch(Message<?> message) {
int sequenceNumber = 1;
int sequenceSize = this.consumers.size();
for (final MessageConsumer consumer : this.consumers) {
int sequenceSize = this.handlers.size();
for (final MessageHandler handler : this.handlers) {
final Message<?> messageToSend = (!this.applySequence) ? message
: MessageBuilder.fromMessage(message)
.setSequenceNumber(sequenceNumber++)
@@ -56,12 +56,12 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
if (executor != null) {
executor.execute(new Runnable() {
public void run() {
BroadcastingDispatcher.this.sendMessageToConsumer(messageToSend, consumer);
BroadcastingDispatcher.this.sendMessageToHandler(messageToSend, handler);
}
});
}
else {
this.sendMessageToConsumer(messageToSend, consumer);
this.sendMessageToHandler(messageToSend, handler);
}
}
return true;

8
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -17,18 +17,18 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
/**
* Strategy interface for dispatching messages to consumers.
* Strategy interface for dispatching messages to handlers.
*
* @author Mark Fisher
*/
public interface MessageDispatcher {
boolean addConsumer(MessageConsumer consumer);
boolean addHandler(MessageHandler handler);
boolean removeConsumer(MessageConsumer consumer);
boolean removeHandler(MessageHandler handler);
boolean dispatch(Message<?> message);

16
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -17,17 +17,17 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageRejectedException;
/**
* Basic implementation of {@link MessageDispatcher} that will attempt
* to send a {@link Message} to one of its consumers. As soon as <em>one</em>
* of the consumers accepts the Message, the dispatcher will return 'true'.
* to send a {@link Message} to one of its handlers. As soon as <em>one</em>
* of the handlers accepts the Message, the dispatcher will return 'true'.
* <p>
* If the dispatcher has no consumers, a {@link MessageDeliveryException}
* will be thrown. If all consumers reject the Message, the dispatcher will
* If the dispatcher has no handlers, a {@link MessageDeliveryException}
* will be thrown. If all handlers reject the Message, the dispatcher will
* throw a MessageRejectedException.
*
* @author Mark Fisher
@@ -35,14 +35,14 @@ import org.springframework.integration.message.MessageRejectedException;
public class SimpleDispatcher extends AbstractDispatcher {
public boolean dispatch(Message<?> message) {
if (this.consumers.size() == 0) {
if (this.handlers.size() == 0) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}
int count = 0;
int rejectedExceptionCount = 0;
for (MessageConsumer consumer : this.consumers) {
for (MessageHandler handler : this.handlers) {
count++;
if (this.sendMessageToConsumer(message, consumer)) {
if (this.sendMessageToHandler(message, handler)) {
return true;
}
rejectedExceptionCount++;

14
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/PollingConsumerEndpoint.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.endpoint;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
@@ -26,18 +26,18 @@ import org.springframework.util.Assert;
*/
public class PollingConsumerEndpoint extends AbstractPollingEndpoint {
private final MessageConsumer consumer;
private final PollableChannel inputChannel;
private final MessageHandler handler;
private volatile long receiveTimeout = 1000;
public PollingConsumerEndpoint(MessageConsumer consumer, PollableChannel inputChannel) {
Assert.notNull(consumer, "consumer must not be null");
public PollingConsumerEndpoint(PollableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
this.consumer = consumer;
Assert.notNull(handler, "handler must not be null");
this.inputChannel = inputChannel;
this.handler = handler;
}
@@ -53,7 +53,7 @@ public class PollingConsumerEndpoint extends AbstractPollingEndpoint {
if (message == null) {
return false;
}
this.consumer.onMessage(message);
this.handler.handleMessage(message);
return true;
}

16
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SubscribingConsumerEndpoint.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.endpoint;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.util.Assert;
/**
@@ -26,20 +26,20 @@ import org.springframework.util.Assert;
*/
public class SubscribingConsumerEndpoint extends AbstractEndpoint implements Lifecycle {
private final MessageConsumer consumer;
private final SubscribableChannel inputChannel;
private final MessageHandler handler;
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
public SubscribingConsumerEndpoint(MessageConsumer consumer, SubscribableChannel inputChannel) {
Assert.notNull(consumer, "consumer must not be null");
public SubscribingConsumerEndpoint(SubscribableChannel inputChannel, MessageHandler handler) {
Assert.notNull(inputChannel, "inputChannel must not be null");
this.consumer = consumer;
Assert.notNull(handler, "handler must not be null");
this.inputChannel = inputChannel;
this.handler = handler;
}
@@ -52,7 +52,7 @@ public class SubscribingConsumerEndpoint extends AbstractEndpoint implements Lif
public void start() {
synchronized (this.lifecycleMonitor) {
if (!this.running) {
this.inputChannel.subscribe(this.consumer);
this.inputChannel.subscribe(this.handler);
this.running = true;
}
}
@@ -61,7 +61,7 @@ public class SubscribingConsumerEndpoint extends AbstractEndpoint implements Lif
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.running) {
this.inputChannel.unsubscribe(this.consumer);
this.inputChannel.unsubscribe(this.handler);
this.running = false;
}
}

8
org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java
View File

@@ -16,20 +16,20 @@
package org.springframework.integration.filter;
import org.springframework.integration.consumer.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.consumer.AbstractReplyProducingMessageHandler;
import org.springframework.integration.consumer.ReplyMessageHolder;
import org.springframework.integration.core.Message;
import org.springframework.integration.selector.MessageSelector;
import org.springframework.util.Assert;
/**
* Message Consumer that delegates to a {@link MessageSelector}. If and only if
* Message Handler that delegates to a {@link MessageSelector}. If and only if
* the selector {@link MessageSelector#accept(Message) accepts} the Message, it
* will be passed to this filter's output channel.
*
* @author Mark Fisher
*/
public class MessageFilter extends AbstractReplyProducingMessageConsumer {
public class MessageFilter extends AbstractReplyProducingMessageHandler {
private MessageSelector selector;
@@ -41,7 +41,7 @@ public class MessageFilter extends AbstractReplyProducingMessageConsumer {
@Override
protected void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
if (this.selector.accept(message)) {
replyHolder.set(message);
}

12
org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
View File

@@ -20,7 +20,7 @@ import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
import org.springframework.integration.consumer.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.consumer.AbstractReplyProducingMessageHandler;
import org.springframework.integration.consumer.ReplyMessageHolder;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
@@ -30,7 +30,7 @@ import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.endpoint.PollingConsumerEndpoint;
import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.TaskSchedulerAware;
@@ -178,19 +178,19 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
return;
}
MessageEndpoint correlator = null;
MessageConsumer consumer = new AbstractReplyProducingMessageConsumer() {
MessageHandler handler = new AbstractReplyProducingMessageHandler() {
@Override
protected void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
replyHolder.set(message);
}
};
if (this.replyChannel instanceof SubscribableChannel) {
correlator = new SubscribingConsumerEndpoint(
consumer, (SubscribableChannel) this.replyChannel);
(SubscribableChannel) this.replyChannel, handler);
}
else if (this.replyChannel instanceof PollableChannel) {
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(
consumer, (PollableChannel) this.replyChannel);
(PollableChannel) this.replyChannel, handler);
endpoint.setTaskScheduler(this.taskScheduler);
endpoint.afterPropertiesSet();
correlator = endpoint;

6
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHandler.java
View File

@@ -19,12 +19,12 @@ package org.springframework.integration.message;
import org.springframework.integration.core.Message;
/**
* Base interface for any component that consumes Messages.
* Base interface for any component that handles Messages.
*
* @author Mark Fisher
*/
public interface MessageConsumer {
public interface MessageHandler {
void onMessage(Message<?> message);
void handleMessage(Message<?> message);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.router;
import java.util.Collection;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.consumer.AbstractMessageConsumer;
import org.springframework.integration.consumer.AbstractMessageHandler;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageDeliveryException;
@@ -29,7 +29,7 @@ import org.springframework.integration.message.MessageDeliveryException;
*
* @author Mark Fisher
*/
public abstract class AbstractMessageRouter extends AbstractMessageConsumer {
public abstract class AbstractMessageRouter extends AbstractMessageHandler {
private volatile MessageChannel defaultOutputChannel;
@@ -67,7 +67,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageConsumer {
}
@Override
protected void onMessageInternal(Message<?> message) {
protected void handleMessageInternal(Message<?> message) {
boolean sent = false;
Collection<MessageChannel> results = this.determineTargetChannels(message);
if (results != null) {

8
org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java
View File

@@ -18,19 +18,19 @@ package org.springframework.integration.splitter;
import java.util.Collection;
import org.springframework.integration.consumer.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.consumer.AbstractReplyProducingMessageHandler;
import org.springframework.integration.consumer.ReplyMessageHolder;
import org.springframework.integration.core.Message;
/**
* Base class for Message-splitting consumers.
* Base class for Message-splitting handlers.
*
* @author Mark Fisher
*/
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageConsumer {
public abstract class AbstractMessageSplitter extends AbstractReplyProducingMessageHandler {
@Override
protected final void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected final void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
Object result = this.splitMessage(message);
if (result == null) {
return;

16
org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingConsumer.java → org.springframework.integration/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java
View File

@@ -16,35 +16,35 @@
package org.springframework.integration.transformer;
import org.springframework.integration.consumer.AbstractReplyProducingMessageConsumer;
import org.springframework.integration.consumer.AbstractReplyProducingMessageHandler;
import org.springframework.integration.consumer.ReplyMessageHolder;
import org.springframework.integration.core.Message;
import org.springframework.util.Assert;
/**
* A reply-producing {@link org.springframework.integration.message.MessageConsumer}
* that delegates to a {@link Transformer} instance to modify the received
* {@link Message} and send the result to its output channel.
* A reply-producing {@link MessageHandler} that delegates to a
* {@link Transformer} instance to modify the received {@link Message}
* and sends the result to its output channel.
*
* @author Mark Fisher
*/
public class MessageTransformingConsumer extends AbstractReplyProducingMessageConsumer {
public class MessageTransformingHandler extends AbstractReplyProducingMessageHandler {
private final Transformer transformer;
/**
* Create a {@link MessageTransformingConsumer} instance that delegates to
* Create a {@link MessageTransformingHandler} instance that delegates to
* the provided {@link Transformer}.
*/
public MessageTransformingConsumer(Transformer transformer) {
public MessageTransformingHandler(Transformer transformer) {
Assert.notNull(transformer, "transformer must not be null");
this.transformer = transformer;
}
@Override
protected void onMessage(Message<?> message, ReplyMessageHolder replyHolder) {
protected void handleRequestMessage(Message<?> message, ReplyMessageHolder replyHolder) {
try {
Message<?> result = transformer.transform(message);
if (result != null) {