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

Added Target interface and split DefaultMessageEndpoint into TargetEndpoint and HandlerEndpoint. All "one-way" adapters now implement Target instead of MessageHandler, the ConcurrentHandler is now ConcurrentTarget, and the MessageDispatcher also operates on Targets rather than MessageHandlers.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-17 17:25:56 +00:00
parent e9673586c4
commit 0f6bd60ee0
69 changed files with 930 additions and 1008 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractTargetAdapter.java
View File

@@ -1,56 +0,0 @@
/*
* Copyright 2002-2008 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.adapter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.SimplePayloadMessageMapper;
import org.springframework.util.Assert;
/**
* Base class providing common behavior for target adapters.
*
* @author Mark Fisher
*/
public abstract class AbstractTargetAdapter<T> implements MessageHandler {
protected Log logger = LogFactory.getLog(this.getClass());
private MessageMapper<?,T> mapper = new SimplePayloadMessageMapper<T>();
public void setMessageMapper(MessageMapper<?,T> mapper) {
Assert.notNull(mapper, "'mapper' must not be null");
this.mapper = mapper;
}
protected MessageMapper<?,T> getMessageMapper() {
return this.mapper;
}
public final Message handle(Message message) {
this.sendToTarget(this.mapper.fromMessage(message));
return null;
}
protected abstract boolean sendToTarget(T object);
}

42
spring-integration-core/src/main/java/org/springframework/integration/adapter/DefaultTargetAdapter.java
View File

@@ -1,42 +0,0 @@
/*
* Copyright 2002-2007 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.adapter;
import org.springframework.integration.message.MessageMapper;
import org.springframework.util.Assert;
/**
* Target adapter implementation that delegates to a {@link MessageMapper}
* and then passes the resulting object to the provided {@link Target}.
*
* @author Mark Fisher
*/
public class DefaultTargetAdapter<T> extends AbstractTargetAdapter<T> {
private Target<T> target;
public DefaultTargetAdapter(Target<T> target) {
Assert.notNull(target, "'target' must not be null");
this.target = target;
}
public boolean sendToTarget(T object) {
return this.target.send(object);
}
}

77
spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingHandler.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* Copyright 2002-2008 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.adapter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.handler.HandlerMethodInvoker;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
import org.springframework.integration.message.MessageMapper;
import org.springframework.util.Assert;
/**
* A {@link MessageHandler} that invokes the specified method on the provided object.
*
* @author Mark Fisher
*/
public class MethodInvokingHandler implements MessageHandler, InitializingBean {
private volatile Object object;
private volatile String method;
private volatile MessageMapper messageMapper;
private volatile MessageCreator messageCreator;
protected HandlerMethodInvoker<?> invoker;
public void setObject(Object object) {
Assert.notNull(object, "'object' must not be null");
this.object = object;
}
public void setMethod(String method) {
Assert.notNull(method, "'method' must not be null");
this.method = method;
}
public void setMessageMapper(MessageMapper messageMapper) {
this.messageMapper = messageMapper;
}
public void setMessageCreator(MessageCreator messageCreator) {
this.messageCreator = messageCreator;
}
public void afterPropertiesSet() {
this.invoker = new HandlerMethodInvoker(this.object, this.method);
}
public Message<?> handle(Message<?> message) {
Object args = (this.messageMapper != null) ? this.messageMapper.mapMessage(message) : message.getPayload();
Object result = this.invoker.invokeMethod(args);
if (result == null) {
return null;
}
return (this.messageCreator != null) ? this.messageCreator.createMessage(result) : new GenericMessage<Object>(result);
}
}

61
spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingTarget.java
View File

@@ -16,61 +16,34 @@
package org.springframework.integration.adapter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.lang.reflect.Method;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.handler.HandlerMethodInvoker;
import org.springframework.util.Assert;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Target;
import org.springframework.integration.util.MethodValidator;
/**
* A messaging target that invokes the specified method on the provided object.
*
* @author Mark Fisher
*/
public class MethodInvokingTarget<T> implements Target<Object>, InitializingBean {
private Log logger = LogFactory.getLog(this.getClass());
private T object;
private String method;
private HandlerMethodInvoker<T> invoker;
private ArgumentListPreparer argumentListPreparer;
public void setObject(T object) {
Assert.notNull(object, "'object' must not be null");
this.object = object;
}
public void setMethod(String method) {
Assert.notNull(method, "'method' must not be null");
this.method = method;
}
public void setArgumentListPreparer(ArgumentListPreparer argumentListPreparer) {
this.argumentListPreparer = argumentListPreparer;
}
public class MethodInvokingTarget extends MethodInvokingHandler implements Target {
@Override
public void afterPropertiesSet() {
this.invoker = new HandlerMethodInvoker<T>(this.object, this.method);
super.afterPropertiesSet();
this.invoker.setMethodValidator(new MethodValidator() {
public void validate(Method method) throws Exception {
if (!method.getReturnType().equals(void.class)) {
throw new ConfigurationException("target method must have a void return");
}
}
});
}
public boolean send(Object object) {
Object args[] = null;
if (this.argumentListPreparer != null) {
args = this.argumentListPreparer.prepare(object);
}
else {
args = new Object[] { object };
}
Object result = this.invoker.invokeMethod(args);
if (result != null && logger.isWarnEnabled()) {
logger.warn("ignoring outbound channel adapter's return value");
}
public boolean send(Message<?> message) {
this.handle(message);
return true;
}

28
spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -23,9 +23,9 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.SimplePayloadMessageMapper;
import org.springframework.util.Assert;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
/**
* Interceptor that publishes a target method's return value to a channel.
@@ -34,11 +34,11 @@ import org.springframework.util.Assert;
*/
public class MessagePublishingInterceptor implements MethodInterceptor {
protected Log logger = LogFactory.getLog(getClass());
protected final Log logger = LogFactory.getLog(getClass());
private MessageMapper mapper = new SimplePayloadMessageMapper();
private volatile MessageCreator messageCreator;
private MessageChannel defaultChannel;
private volatile MessageChannel defaultChannel;
public void setDefaultChannel(MessageChannel defaultChannel) {
@@ -46,14 +46,13 @@ public class MessagePublishingInterceptor implements MethodInterceptor {
}
/**
* Specify the {@link MessageMapper} to use when creating a message from the
* return value Object. The default is a {@link SimplePayloadMessageMapper}.
* Specify the {@link MessageCreator} to use when creating a message from the
* return value Object.
*
* @param mapper the mapper to use
* @param messageCreator the MessageCreator to use
*/
public void setMessageMapper(MessageMapper mapper) {
Assert.notNull(mapper, "mapper must not be null");
this.mapper = mapper;
public void setMessageCreator(MessageCreator messageCreator) {
this.messageCreator = messageCreator;
}
/**
@@ -70,7 +69,8 @@ public class MessagePublishingInterceptor implements MethodInterceptor {
}
}
else {
channel.send(mapper.toMessage(retval));
Message<?> message = (this.messageCreator != null) ? this.messageCreator.createMessage(retval) : new GenericMessage<Object>(retval);
channel.send(message);
}
}
return retval;

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

@@ -44,11 +44,13 @@ import org.springframework.integration.dispatcher.SchedulingMessageDispatcher;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultEndpointRegistry;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
@@ -235,17 +237,25 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
public void registerHandler(String name, MessageHandler handler, Subscription subscription, ConcurrencyPolicy concurrencyPolicy) {
if (!this.initialized) {
this.initialize();
}
Assert.notNull(name, "'name' must not be null");
Assert.notNull(handler, "'handler' must not be null");
Assert.notNull(subscription, "'subscription' must not be null");
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
this.doRegisterEndpoint(name, endpoint, subscription, concurrencyPolicy);
}
public void registerTarget(String name, Target target, Subscription subscription) {
this.registerTarget(name, target, subscription, this.defaultConcurrencyPolicy);
}
public void registerTarget(String name, Target target, Subscription subscription, ConcurrencyPolicy concurrencyPolicy) {
Assert.notNull(target, "'target' must not be null");
TargetEndpoint endpoint = new TargetEndpoint(target);
this.doRegisterEndpoint(name, endpoint, subscription, concurrencyPolicy);
}
private void doRegisterEndpoint(String name, TargetEndpoint endpoint, Subscription subscription, ConcurrencyPolicy concurrencyPolicy) {
endpoint.setName(name);
endpoint.setSubscription(subscription);
endpoint.setConcurrencyPolicy(concurrencyPolicy);
endpoint.afterPropertiesSet();
this.registerEndpoint(name, endpoint);
}
@@ -257,8 +267,11 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
}
if (endpoint.getConcurrencyPolicy() == null && this.defaultConcurrencyPolicy != null
&& endpoint instanceof DefaultMessageEndpoint) {
((DefaultMessageEndpoint) endpoint).setConcurrencyPolicy(this.defaultConcurrencyPolicy);
&& endpoint instanceof TargetEndpoint) {
((TargetEndpoint) endpoint).setConcurrencyPolicy(this.defaultConcurrencyPolicy);
}
if (endpoint instanceof TargetEndpoint) {
((TargetEndpoint) endpoint).afterPropertiesSet();
}
this.endpointRegistry.registerEndpoint(name, endpoint);
if (this.isRunning()) {
@@ -277,7 +290,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
Collection<SchedulingMessageDispatcher> dispatchers = this.dispatchers.values();
boolean removed = false;
for (SchedulingMessageDispatcher dispatcher : dispatchers) {
removed = (removed || dispatcher.removeHandler(endpoint));
removed = (removed || dispatcher.removeTarget(endpoint));
}
if (removed) {
return endpoint;
@@ -329,9 +342,9 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
this.registerChannel(channelName, channel);
}
}
if (endpoint instanceof DefaultMessageEndpoint) {
DefaultMessageEndpoint dme = (DefaultMessageEndpoint) endpoint;
String outputChannelName = dme.getDefaultOutputChannelName();
if (endpoint instanceof HandlerEndpoint) {
HandlerEndpoint handlerEndpoint = (HandlerEndpoint) endpoint;
String outputChannelName = handlerEndpoint.getDefaultOutputChannelName();
if (outputChannelName != null && this.lookupChannel(outputChannelName) == null) {
if (!this.autoCreateChannels) {
throw new ConfigurationException("Unknown channel '" + outputChannelName +
@@ -340,8 +353,11 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
this.registerChannel(outputChannelName, new SimpleChannel());
}
if (!dme.hasErrorHandler() && this.getErrorChannel() != null && !this.getErrorChannel().equals(channel)) {
dme.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
}
if (endpoint instanceof TargetEndpoint) {
TargetEndpoint targetEndpoint = (TargetEndpoint) endpoint;
if (!targetEndpoint.hasErrorHandler() && this.getErrorChannel() != null && !this.getErrorChannel().equals(channel)) {
targetEndpoint.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
}
}
this.registerWithDispatcher(channel, endpoint, subscription.getSchedule());
@@ -369,14 +385,14 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
}
private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule) {
private void registerWithDispatcher(MessageChannel channel, Target target, Schedule schedule) {
if (schedule == null && (channel instanceof SynchronousChannel)) {
((SynchronousChannel) channel).addHandler(handler);
if (handler instanceof Lifecycle) {
((Lifecycle) handler).start();
((SynchronousChannel) channel).addTarget(target);
if (target instanceof Lifecycle) {
((Lifecycle) target).start();
}
if (handler instanceof DefaultMessageEndpoint) {
((DefaultMessageEndpoint) handler).setErrorHandler(new ErrorHandler() {
if (target instanceof TargetEndpoint) {
((TargetEndpoint) target).setErrorHandler(new ErrorHandler() {
public void handle(Throwable t) {
if (t instanceof MessagingException) {
throw (MessagingException) t;
@@ -394,7 +410,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
}
return;
}
dispatcher.addHandler(handler, schedule);
dispatcher.addTarget(target, schedule);
if (this.isRunning() && !dispatcher.isRunning()) {
dispatcher.start();
}

25
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -25,11 +25,10 @@ import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.DefaultTargetAdapter;
import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
@@ -78,7 +77,10 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
if (this.isInbound) {
adapterDef = new RootBeanDefinition(PollingSourceAdapter.class);
invokerDef = new RootBeanDefinition(MethodInvokingSource.class);
String invokerBeanName = this.configureAndRegisterInvoker(invokerDef, ref, method, parserContext);
invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
invokerDef.getPropertyValues().addPropertyValue("method", method);
String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName));
String period = element.getAttribute(PERIOD_ATTRIBUTE);
if (!StringUtils.hasText(period)) {
throw new ConfigurationException("'period' is required");
@@ -89,10 +91,9 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(schedule);
}
else {
adapterDef = new RootBeanDefinition(DefaultTargetAdapter.class);
invokerDef = new RootBeanDefinition(MethodInvokingTarget.class);
String invokerBeanName = this.configureAndRegisterInvoker(invokerDef, ref, method, parserContext);
adapterDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(invokerBeanName));
adapterDef = new RootBeanDefinition(MethodInvokingTarget.class);
adapterDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(ref));
adapterDef.getPropertyValues().addPropertyValue("method", method);
}
adapterDef.setSource(parserContext.extractSource(element));
String beanName = element.getAttribute(ID_ATTRIBUTE);
@@ -100,7 +101,7 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
beanName = parserContext.getReaderContext().generateBeanName(adapterDef);
}
if (!this.isInbound) {
RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class);
RootBeanDefinition endpointDef = new RootBeanDefinition(HandlerEndpoint.class);
RootBeanDefinition subscriptionDef = new RootBeanDefinition(Subscription.class);
subscriptionDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(channel));
String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
@@ -114,12 +115,4 @@ public class ChannelAdapterParser implements BeanDefinitionParser {
return adapterDef;
}
private String configureAndRegisterInvoker(RootBeanDefinition invokerDef, String objectRef, String methodName, ParserContext parserContext) {
invokerDef.getPropertyValues().addPropertyValue("object", new RuntimeBeanReference(objectRef));
invokerDef.getPropertyValues().addPropertyValue("method", methodName);
String invokerBeanName = parserContext.getReaderContext().generateBeanName(invokerDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(invokerDef, invokerBeanName));
return invokerBeanName;
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java
View File

@@ -32,7 +32,7 @@ import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.scheduling.PollingSchedule;
@@ -95,7 +95,7 @@ public class EndpointParser implements BeanDefinitionParser {
public BeanDefinition parse(Element element, ParserContext parserContext) {
RootBeanDefinition endpointDef = new RootBeanDefinition(DefaultMessageEndpoint.class);
RootBeanDefinition endpointDef = new RootBeanDefinition(HandlerEndpoint.class);
endpointDef.setSource(parserContext.extractSource(element));
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
String defaultOutputChannel = element.getAttribute(DEFAULT_OUTPUT_CHANNEL_ATTRIBUTE);

26
spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -34,7 +34,6 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.OrderComparator;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.adapter.DefaultTargetAdapter;
import org.springframework.integration.adapter.MethodInvokingSource;
import org.springframework.integration.adapter.MethodInvokingTarget;
import org.springframework.integration.adapter.PollingSourceAdapter;
@@ -51,7 +50,7 @@ import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
@@ -121,7 +120,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
if (handlerChain == null) {
throw new ConfigurationException("@MessageEndpoint has no handler method");
}
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handlerChain);
HandlerEndpoint endpoint = new HandlerEndpoint(handlerChain);
this.configureInput(bean, beanName, endpointAnnotation, endpoint);
if (StringUtils.hasText(defaultOutputChannelName)) {
endpoint.setDefaultOutputChannelName(defaultOutputChannelName);
@@ -143,7 +142,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
}
private void configureInput(final Object bean, final String beanName, MessageEndpoint annotation,
final DefaultMessageEndpoint endpoint) {
final HandlerEndpoint endpoint) {
String channelName = annotation.input();
if (StringUtils.hasText(channelName)) {
Subscription subscription = new Subscription(channelName);
@@ -175,7 +174,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
});
}
private void configureDefaultOutput(final Object bean, final String beanName, final DefaultMessageEndpoint endpoint) {
private void configureDefaultOutput(final Object bean, final String beanName, final HandlerEndpoint endpoint) {
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
boolean foundDefaultOutput = false;
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
@@ -184,23 +183,12 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
if (foundDefaultOutput) {
throw new ConfigurationException("only one @DefaultOutput allowed per endpoint");
}
MethodInvokingTarget<Object> target = new MethodInvokingTarget<Object>();
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(bean);
target.setMethod(method.getName());
target.afterPropertiesSet();
DefaultTargetAdapter<Object> adapter = new DefaultTargetAdapter<Object>(target);
MessageHandler handler = endpoint.getHandler();
if (handler == null) {
endpoint.setHandler(adapter);
}
else if (handler instanceof MessageHandlerChain) {
((MessageHandlerChain) handler).add(adapter);
}
else {
MessageHandlerChain chain = new MessageHandlerChain();
chain.add(handler);
chain.add(adapter);
}
((MessageHandlerChain) handler).add(target);
foundDefaultOutput = true;
return;
}
@@ -208,7 +196,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
});
}
private void configureCompletionStrategy(final Object bean, final DefaultMessageEndpoint endpoint) {
private void configureCompletionStrategy(final Object bean, final HandlerEndpoint endpoint) {
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);

58
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java
View File

@@ -29,8 +29,8 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.MessagingTask;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
import org.springframework.integration.scheduling.PollingSchedule;
@@ -58,7 +58,7 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher {
private volatile Schedule defaultSchedule = new PollingSchedule(5);
private final ConcurrentMap<Schedule, List<MessageHandler>> scheduledHandlers = new ConcurrentHashMap<Schedule, List<MessageHandler>>();
private final ConcurrentMap<Schedule, List<Target>> scheduledTargets = new ConcurrentHashMap<Schedule, List<Target>>();
private final AtomicLong totalMessagesProcessed = new AtomicLong();
@@ -81,41 +81,41 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher {
this.defaultSchedule = defaultSchedule;
}
public void addHandler(MessageHandler handler) {
this.addHandler(handler, null);
public void addTarget(Target target) {
this.addTarget(target, null);
}
public void addHandler(MessageHandler handler, Schedule schedule) {
Assert.notNull(handler, "'handler' must not be null");
public void addTarget(Target target, Schedule schedule) {
Assert.notNull(target, "'target' must not be null");
if (schedule == null) {
schedule = this.defaultSchedule;
}
else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) {
if (logger.isInfoEnabled()) {
logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " +
"Therefore all handlers are scheduled with its 'defaultSchedule', " +
"Therefore all targets are scheduled with its 'defaultSchedule', " +
"and the provided schedule will be ignored.");
}
schedule = this.defaultSchedule;
}
if (this.isRunning() && handler instanceof Lifecycle) {
((Lifecycle) handler).start();
if (this.isRunning() && target instanceof Lifecycle) {
((Lifecycle) target).start();
}
List<MessageHandler> handlers = this.scheduledHandlers.get(schedule);
if (handlers == null) {
handlers = this.scheduledHandlers.putIfAbsent(schedule, new CopyOnWriteArrayList<MessageHandler>());
List<Target> targets = this.scheduledTargets.get(schedule);
if (targets == null) {
targets = this.scheduledTargets.putIfAbsent(schedule, new CopyOnWriteArrayList<Target>());
}
this.scheduledHandlers.get(schedule).add(handler);
if (handlers == null && this.isRunning()) {
this.scheduledTargets.get(schedule).add(target);
if (targets == null && this.isRunning()) {
this.scheduleDispatcherTask(schedule);
}
}
public boolean removeHandler(MessageHandler handler) {
public boolean removeTarget(Target target) {
boolean removed = false;
Collection<List<MessageHandler>> handlerLists = this.scheduledHandlers.values();
for (List<MessageHandler> handlers : handlerLists) {
removed = (removed || handlers.remove(handler));
Collection<List<Target>> targetLists = this.scheduledTargets.values();
for (List<Target> targets : targetLists) {
removed = (removed || targets.remove(target));
}
return removed;
}
@@ -135,7 +135,7 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher {
if (!this.scheduler.isRunning()) {
this.scheduler.start();
}
for (Schedule schedule : this.scheduledHandlers.keySet()) {
for (Schedule schedule : this.scheduledTargets.keySet()) {
scheduleDispatcherTask(schedule);
}
this.running = true;
@@ -143,10 +143,10 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher {
}
private void scheduleDispatcherTask(Schedule schedule) {
List<MessageHandler> handlers = this.scheduledHandlers.get(schedule);
for (MessageHandler handler : handlers) {
if (handler instanceof Lifecycle) {
((Lifecycle) handler).start();
List<Target> targets = this.scheduledTargets.get(schedule);
for (Target target : targets) {
if (target instanceof Lifecycle) {
((Lifecycle) target).start();
}
}
this.scheduler.schedule(new DispatcherTask(schedule));
@@ -157,10 +157,10 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher {
return;
}
synchronized (this.lifecycleMonitor) {
for (List<MessageHandler> handlerList : scheduledHandlers.values()) {
for (MessageHandler handler : handlerList) {
if (handler instanceof Lifecycle) {
((Lifecycle) handler).stop();
for (List<Target> targetList : this.scheduledTargets.values()) {
for (Target target : targetList) {
if (target instanceof Lifecycle) {
((Lifecycle) target).stop();
}
}
}
@@ -193,8 +193,8 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher {
schedule = this.defaultSchedule;
}
MessageDistributor distributor = new DefaultMessageDistributor(this.channel.getDispatcherPolicy());
for (MessageHandler handler : this.scheduledHandlers.get(schedule)) {
distributor.addHandler(handler);
for (Target target : this.scheduledTargets.get(schedule)) {
distributor.addTarget(target);
}
return distributor;
}

34
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java
View File

@@ -25,11 +25,11 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.Target;
import org.springframework.integration.message.selector.MessageSelectorRejectedException;
import org.springframework.util.Assert;
@@ -42,7 +42,7 @@ public class DefaultMessageDistributor implements MessageDistributor {
private final Log logger = LogFactory.getLog(this.getClass());
private final List<MessageHandler> handlers = new CopyOnWriteArrayList<MessageHandler>();
private final List<Target> targets = new CopyOnWriteArrayList<Target>();
private final DispatcherPolicy dispatcherPolicy;
@@ -53,21 +53,21 @@ public class DefaultMessageDistributor implements MessageDistributor {
}
public void addHandler(MessageHandler handler) {
this.handlers.add(handler);
public void addTarget(Target target) {
this.targets.add(target);
}
public boolean removeHandler(MessageHandler handler) {
return this.handlers.remove(handler);
public boolean removeTarget(Target target) {
return this.targets.remove(target);
}
public boolean distribute(Message<?> message) {
int attempts = 0;
List<MessageHandler> targets = new ArrayList<MessageHandler>(this.handlers);
List<Target> targets = new ArrayList<Target>(this.targets);
while (attempts < this.dispatcherPolicy.getRejectionLimit()) {
if (attempts > 0) {
if (logger.isDebugEnabled()) {
logger.debug("handler(s) rejected message after " + attempts +
logger.debug("target(s) rejected message after " + attempts +
" attempt(s), will try again after 'retryInterval' of " +
this.dispatcherPolicy.getRetryInterval() + " milliseconds");
}
@@ -79,37 +79,37 @@ public class DefaultMessageDistributor implements MessageDistributor {
return false;
}
}
Iterator<MessageHandler> iter = targets.iterator();
Iterator<Target> iter = targets.iterator();
if (!iter.hasNext()) {
if (logger.isWarnEnabled()) {
logger.warn("no active handlers");
logger.warn("no active targets");
}
return false;
}
boolean rejected = false;
while (iter.hasNext()) {
MessageHandler handler = iter.next();
Target target = iter.next();
try {
handler.handle(message);
if (!this.dispatcherPolicy.isPublishSubscribe()) {
boolean sent = target.send(message);
if (!this.dispatcherPolicy.isPublishSubscribe() && sent) {
return true;
}
iter.remove();
}
catch (MessageSelectorRejectedException e) {
if (logger.isDebugEnabled()) {
logger.debug("selector rejected message, continuing with other handlers if available", e);
logger.debug("selector rejected message, continuing with other targets if available", e);
}
}
catch (MessageHandlerNotRunningException e) {
if (logger.isDebugEnabled()) {
logger.debug("handler not running, continuing with other handlers if available", e);
logger.debug("target not running, continuing with other targets if available", e);
}
}
catch (MessageHandlerRejectedExecutionException e) {
rejected = true;
if (logger.isDebugEnabled()) {
logger.debug("handler is busy, continuing with other handlers if available", e);
logger.debug("target is busy, continuing with other targets if available", e);
}
}
}
@@ -121,7 +121,7 @@ public class DefaultMessageDistributor implements MessageDistributor {
if (this.dispatcherPolicy.getShouldFailOnRejectionLimit()) {
throw new MessageDeliveryException(message, "Dispatcher reached rejection limit of "
+ this.dispatcherPolicy.getRejectionLimit()
+ ". Consider increasing the handler's concurrency and/or "
+ ". Consider increasing the target's concurrency and/or "
+ "the dispatcherPolicy's 'rejectionLimit'.");
}
return false;

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -16,7 +16,7 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Target;
/**
* Strategy interface for dispatching messages.
@@ -25,9 +25,9 @@ import org.springframework.integration.handler.MessageHandler;
*/
public interface MessageDispatcher {
void addHandler(MessageHandler handler);
void addTarget(Target target);
boolean removeHandler(MessageHandler handler);
boolean removeTarget(Target target);
int dispatch();

9
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java
View File

@@ -16,20 +16,19 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Target;
/**
* Strategy interface for distributing a {@link Message} to one or more
* {@link MessageHandler MessageHandlers}.
* Strategy interface for distributing a {@link Message} to one or more {@link Target targets}.
*
* @author Mark Fisher
*/
public interface MessageDistributor {
void addHandler(MessageHandler handler);
void addTarget(Target target);
boolean removeHandler(MessageHandler handler);
boolean removeTarget(Target target);
boolean distribute(Message<?> message);

8
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -17,11 +17,11 @@
package org.springframework.integration.dispatcher;
import org.springframework.context.Lifecycle;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.Schedule;
/**
* An extension to the {@link MessageDispatcher} strategy for handlers that may
* An extension to the {@link MessageDispatcher} strategy for targets that may
* be scheduled.
*
* @author Mark Fisher
@@ -30,6 +30,6 @@ public interface SchedulingMessageDispatcher extends MessageDispatcher, Lifecycl
void setDefaultSchedule(Schedule defaultSchedule);
void addHandler(MessageHandler handler, Schedule schedule);
void addTarget(Target target, Schedule schedule);
}

9
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
View File

@@ -27,6 +27,7 @@ import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.Target;
import org.springframework.integration.message.selector.MessageSelector;
/**
@@ -70,13 +71,13 @@ public class SynchronousChannel extends AbstractMessageChannel {
this.source = source;
}
public void addHandler(MessageHandler handler) {
this.distributor.addHandler(handler);
public void addTarget(Target target) {
this.distributor.addTarget(target);
this.handlerCount.incrementAndGet();
}
public boolean removeHandler(MessageHandler handler) {
if (this.distributor.removeHandler(handler)) {
public boolean removeTarget(Target target) {
if (this.distributor.removeTarget(target)) {
this.handlerCount.decrementAndGet();
return true;
}

41
spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentHandler.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/ConcurrentTarget.java
View File

@@ -25,36 +25,34 @@ import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.Target;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;
/**
* A {@link MessageHandler} implementation that encapsulates a
* {@link ThreadPoolTaskExecutor} and delegates to a wrapped handler for
* concurrent, asynchronous message handling.
* A {@link Target} implementation that encapsulates an Executor and delegates
* to a wrapped target for concurrent, asynchronous message handling.
*
* @author Mark Fisher
*/
public class ConcurrentHandler implements MessageHandler, DisposableBean {
public class ConcurrentTarget implements Target, DisposableBean {
private final Log logger = LogFactory.getLog(this.getClass());
private final MessageHandler handler;
private final Target target;
private final ExecutorService executor;
private volatile ErrorHandler errorHandler;
private volatile ReplyHandler replyHandler;
public ConcurrentHandler(MessageHandler handler, ExecutorService executor) {
Assert.notNull(handler, "'handler' must not be null");
public ConcurrentTarget(Target target, ExecutorService executor) {
Assert.notNull(target, "'target' must not be null");
Assert.notNull(executor, "'executor' must not be null");
this.handler = handler;
this.target = target;
this.executor = executor;
}
@@ -63,21 +61,17 @@ public class ConcurrentHandler implements MessageHandler, DisposableBean {
this.errorHandler = errorHandler;
}
public void setReplyHandler(ReplyHandler replyHandler) {
this.replyHandler = replyHandler;
}
public void destroy() {
this.executor.shutdownNow();
this.executor.shutdown();
}
public Message<?> handle(Message<?> message) {
public boolean send(Message<?> message) {
if (this.executor.isShutdown()) {
throw new MessageHandlerNotRunningException(message);
}
try {
this.executor.execute(new HandlerTask(message));
return null;
this.executor.execute(new TargetTask(message));
return true;
}
catch (RuntimeException e) {
throw new MessageHandlerRejectedExecutionException(message, e);
@@ -85,19 +79,18 @@ public class ConcurrentHandler implements MessageHandler, DisposableBean {
}
private class HandlerTask implements Runnable {
private class TargetTask implements Runnable {
private Message<?> message;
HandlerTask(Message<?> message) {
TargetTask(Message<?> message) {
this.message = message;
}
public void run() {
try {
Message<?> reply = handler.handle(this.message);
if (replyHandler != null) {
replyHandler.handle(reply, this.message.getHeader());
if (!target.send(this.message)) {
throw new MessageDeliveryException(message, "failed to send message to target");
}
}
catch (Throwable t) {

163
spring-integration-core/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java Normal file
View File

@@ -0,0 +1,163 @@
/*
* Copyright 2002-2008 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.endpoint;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.Target;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Implementation of the {@link MessageEndpoint} interface for invoking
* {@link MessageHandler MessageHandlers}.
*
* @author Mark Fisher
*/
public class HandlerEndpoint extends TargetEndpoint {
private volatile MessageHandler handler;
private volatile ReplyHandler replyHandler = new EndpointReplyHandler();
private volatile long replyTimeout = 1000;
private volatile String defaultOutputChannelName;
public HandlerEndpoint(MessageHandler handler) {
Assert.notNull(handler, "handler must not be null");
this.handler = handler;
}
public MessageHandler getHandler() {
return this.handler;
}
public void setReplyHandler(ReplyHandler replyHandler) {
Assert.notNull(replyHandler, "'replyHandler' must not be null");
this.replyHandler = replyHandler;
}
/**
* Set the timeout in milliseconds to be enforced when this endpoint sends a
* reply message. If the message is not sent successfully within the
* allotted time, then a MessageDeliveryException will be thrown.
* The default <code>replyTimeout</code> value is 1000 milliseconds.
*/
public void setReplyTimeout(long replyTimeout) {
this.replyTimeout = replyTimeout;
}
/**
* Set the name of the channel to which this endpoint should send reply
* messages by default.
*/
public void setDefaultOutputChannelName(String defaultOutputChannelName) {
this.defaultOutputChannelName = defaultOutputChannelName;
}
public String getDefaultOutputChannelName() {
return this.defaultOutputChannelName;
}
public void afterPropertiesSet() {
Assert.notNull(this.handler, "handler must not be null");
if (this.handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.handler).setChannelRegistry(this.getChannelRegistry());
}
super.setTarget(new HandlerInvokingTarget(this.handler, this.replyHandler));
super.afterPropertiesSet();
}
private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) {
Object returnAddress = originalMessageHeader.getReturnAddress();
if (returnAddress instanceof MessageChannel) {
return (MessageChannel) returnAddress;
}
ChannelRegistry registry = this.getChannelRegistry();
if (returnAddress instanceof String && registry != null) {
String channelName = (String) returnAddress;
if (StringUtils.hasText(channelName)) {
return registry.lookupChannel(channelName);
}
}
if (this.defaultOutputChannelName != null && registry != null) {
return registry.lookupChannel(this.defaultOutputChannelName);
}
return null;
}
private static class HandlerInvokingTarget implements Target {
private final MessageHandler handler;
private final ReplyHandler replyHandler;
public HandlerInvokingTarget(MessageHandler handler, ReplyHandler replyHandler) {
this.handler = handler;
this.replyHandler = replyHandler;
}
public boolean send(Message<?> message) {
Message<?> replyMessage = this.handler.handle(message);
if (replyMessage != null) {
if (replyMessage.getHeader().getCorrelationId() == null) {
replyMessage.getHeader().setCorrelationId(message.getId());
}
this.replyHandler.handle(replyMessage, message.getHeader());
}
return true;
}
}
private class EndpointReplyHandler implements ReplyHandler {
public void handle(Message<?> replyMessage, MessageHeader originalMessageHeader) {
if (replyMessage == null) {
return;
}
MessageChannel replyChannel = resolveReplyChannel(originalMessageHeader);
if (replyChannel == null) {
throw new MessageHandlingException(replyMessage, "Unable to determine reply channel for message. " +
"Provide a 'returnAddress' in the message header or a 'defaultOutputChannelName' on the message endpoint.");
}
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + HandlerEndpoint.this + "' replying to channel '" + replyChannel + "' with message: " + replyMessage);
}
if (!replyChannel.send(replyMessage, replyTimeout)) {
throw new MessageDeliveryException(replyMessage,
"unable to send reply message within alloted timeout of " + replyTimeout + " milliseconds");
}
}
}
}

8
spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageEndpoint.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -16,8 +16,10 @@
package org.springframework.integration.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.Subscription;
/**
@@ -25,7 +27,7 @@ import org.springframework.integration.scheduling.Subscription;
*
* @author Mark Fisher
*/
public interface MessageEndpoint extends MessageHandler, Lifecycle {
public interface MessageEndpoint extends Target, ChannelRegistryAware, InitializingBean, Lifecycle {
String getName();

159
spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java → spring-integration-core/src/main/java/org/springframework/integration/endpoint/TargetEndpoint.java
View File

@@ -29,37 +29,32 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerNotRunningException;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.Target;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.integration.message.selector.MessageSelectorRejectedException;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* Default implementation of the {@link MessageEndpoint} interface.
* Base class for {@link MessageEndpoint} implementations.
*
* @author Mark Fisher
*/
public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryAware, InitializingBean, BeanNameAware {
public class TargetEndpoint implements MessageEndpoint, BeanNameAware {
private final Log logger = LogFactory.getLog(this.getClass());
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile String name;
private volatile MessageHandler handler;
private volatile Target target;
private volatile Subscription subscription;
@@ -69,12 +64,6 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
private final List<MessageSelector> selectors = new CopyOnWriteArrayList<MessageSelector>();
private volatile ReplyHandler replyHandler = new EndpointReplyHandler();
private volatile long replyTimeout = 1000;
private volatile String defaultOutputChannelName;
private volatile ChannelRegistry channelRegistry;
private volatile boolean initialized;
@@ -82,9 +71,12 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
private volatile boolean running;
public DefaultMessageEndpoint(MessageHandler handler) {
Assert.notNull(handler, "handler must not be null");
this.handler = handler;
public TargetEndpoint() {
}
public TargetEndpoint(Target target) {
Assert.notNull(target, "target must not be null");
this.target = target;
}
@@ -100,15 +92,13 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
this.setName(beanName);
}
public MessageHandler getHandler() {
return this.handler;
public Target getTarget() {
return this.target;
}
/**
* Set the handler to be invoked for each consumed message.
*/
public void setHandler(MessageHandler handler) {
this.handler = handler;
public void setTarget(Target target) {
Assert.notNull(target, "target must not be null");
this.target = target;
}
public void setMessageSelectors(List<MessageSelector> selectors) {
@@ -145,34 +135,6 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
return (this.errorHandler != null);
}
public void setReplyHandler(ReplyHandler replyHandler) {
Assert.notNull(replyHandler, "'replyHandler' must not be null");
this.replyHandler = replyHandler;
}
/**
* Set the timeout in milliseconds to be enforced when this endpoint sends a
* reply message. If the message is not sent successfully within the
* allotted time, then it will be sent within a MessageDeliveryException to
* the error handler instead. The default <code>replyTimeout</code> value
* is 1000 milliseconds.
*/
public void setReplyTimeout(long replyTimeout) {
this.replyTimeout = replyTimeout;
}
public String getDefaultOutputChannelName() {
return this.defaultOutputChannelName;
}
/**
* Set the name of the channel to which this endpoint should send reply
* messages by default.
*/
public void setDefaultOutputChannelName(String defaultOutputChannelName) {
this.defaultOutputChannelName = defaultOutputChannelName;
}
/**
* Set the channel registry to use for looking up channels by name.
*/
@@ -180,23 +142,25 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
this.channelRegistry = channelRegistry;
}
protected ChannelRegistry getChannelRegistry() {
return this.channelRegistry;
}
public void afterPropertiesSet() {
if (this.handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.handler).setChannelRegistry(this.channelRegistry);
if (this.target instanceof ChannelRegistryAware) {
((ChannelRegistryAware) this.target).setChannelRegistry(this.channelRegistry);
}
if (this.concurrencyPolicy != null && !(this.handler instanceof ConcurrentHandler)) {
if (this.concurrencyPolicy != null && !(this.target instanceof ConcurrentTarget)) {
int capacity = this.concurrencyPolicy.getQueueCapacity();
BlockingQueue<Runnable> queue = (capacity < 1) ? new SynchronousQueue<Runnable>() : new ArrayBlockingQueue<Runnable>(capacity);
ExecutorService executor = new ThreadPoolExecutor(
this.concurrencyPolicy.getCoreSize(), this.concurrencyPolicy.getMaxSize(),
ExecutorService executor = new ThreadPoolExecutor(this.concurrencyPolicy.getCoreSize(), this.concurrencyPolicy.getMaxSize(),
this.concurrencyPolicy.getKeepAliveSeconds(), TimeUnit.SECONDS, queue);
this.handler = new ConcurrentHandler(this.handler, executor);
this.target = new ConcurrentTarget(this.target, executor);
}
if (this.handler instanceof ConcurrentHandler) {
if (this.target instanceof ConcurrentTarget) {
if (this.errorHandler != null) {
((ConcurrentHandler) this.handler).setErrorHandler(this.errorHandler);
((ConcurrentTarget) this.target).setErrorHandler(this.errorHandler);
}
((ConcurrentHandler) this.handler).setReplyHandler(this.replyHandler);
}
this.initialized = true;
}
@@ -209,7 +173,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
if (this.isRunning()) {
return;
}
if (!initialized) {
if (!this.initialized) {
this.afterPropertiesSet();
}
this.running = true;
@@ -219,10 +183,20 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
if (!this.isRunning()) {
return;
}
if (this.target instanceof DisposableBean) {
try {
((DisposableBean) this.target).destroy();
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("exception occurred when destroying target", e);
}
}
}
this.running = false;
}
public final Message<?> handle(Message<?> message) {
public final boolean send(Message<?> message) {
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + this + "' handling message: " + message);
}
@@ -235,69 +209,26 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
}
}
try {
Message<?> replyMessage = this.handler.handle(message);
if (replyMessage != null) {
if (replyMessage.getHeader().getCorrelationId() == null) {
replyMessage.getHeader().setCorrelationId(message.getId());
}
this.replyHandler.handle(replyMessage, message.getHeader());
}
return this.target.send(message);
}
catch (MessageHandlerRejectedExecutionException e) {
throw e;
}
catch (Throwable t) {
if (this.errorHandler == null) {
if (t instanceof MessageHandlingException) {
throw (MessageHandlingException) t;
}
throw new MessageHandlingException(message,
"error occurred in endpoint, and no 'errorHandler' available", t);
}
this.errorHandler.handle(t);
return false;
}
return null;
}
public String toString() {
return (this.name != null) ? this.name : super.toString();
}
private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) {
Object returnAddress = originalMessageHeader.getReturnAddress();
if (returnAddress instanceof MessageChannel) {
return (MessageChannel) returnAddress;
}
if (returnAddress instanceof String && this.channelRegistry != null) {
String channelName = (String) returnAddress;
if (StringUtils.hasText(channelName)) {
return this.channelRegistry.lookupChannel(channelName);
}
}
if (this.defaultOutputChannelName != null && this.channelRegistry != null) {
return this.channelRegistry.lookupChannel(this.defaultOutputChannelName);
}
return null;
}
private class EndpointReplyHandler implements ReplyHandler {
public void handle(Message<?> replyMessage, MessageHeader originalMessageHeader) {
if (replyMessage == null) {
return;
}
MessageChannel replyChannel = resolveReplyChannel(originalMessageHeader);
if (replyChannel == null) {
throw new MessageHandlingException(replyMessage,
"Unable to determine reply channel for message. Provide a 'returnAddress' in the message header " +
"or a 'defaultOutputChannelName' on the message endpoint.");
}
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + DefaultMessageEndpoint.this + "' replying to channel '" + replyChannel + "' with message: " + replyMessage);
}
if (!replyChannel.send(replyMessage, replyTimeout)) {
throw new MessageDeliveryException(replyMessage,
"unable to send reply message within alloted timeout of " + replyTimeout + " milliseconds");
}
}
}
}

3
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java
View File

@@ -24,7 +24,6 @@ import org.springframework.core.Ordered;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.MessageMapper;
import org.springframework.util.Assert;
/**
@@ -123,7 +122,7 @@ public abstract class AbstractMessageHandlerAdapter<T> implements MessageHandler
}
protected Message<?> createReplyMessage(Object payload, MessageHeader originalMessageHeader) {
return new GenericMessage(payload, originalMessageHeader);
return new GenericMessage<Object>(payload, originalMessageHeader);
}
/**

43
spring-integration-core/src/main/java/org/springframework/integration/message/AbstractMessageMapper.java
View File

@@ -1,43 +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.message;
import org.springframework.integration.util.RandomUuidGenerator;
import org.springframework.integration.util.IdGenerator;
import org.springframework.util.Assert;
/**
* Base class that provides the default {@link IdGenerator} as well as a setter
* for providing a custom id generator implementation.
*
* @author Mark Fisher
*/
public abstract class AbstractMessageMapper<M, O> implements MessageMapper<M, O> {
private IdGenerator idGenerator = new RandomUuidGenerator();
public void setIdGenerator(IdGenerator idGenerator) {
Assert.notNull(idGenerator, "'idGenerator' must not be null");
this.idGenerator = idGenerator;
}
protected IdGenerator getIdGenerator() {
return this.idGenerator;
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/adapter/ArgumentListPreparer.java → spring-integration-core/src/main/java/org/springframework/integration/message/MessageCreator.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -14,15 +14,15 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
package org.springframework.integration.message;
/**
* A strategy for preparing an argument list from a single source object.
* Strategy interface for creating a {@link Message} from an Object.
*
* @author Mark Fisher
*/
public interface ArgumentListPreparer {
public interface MessageCreator<O, P> {
Object[] prepare(Object source);
Message<P> createMessage(O object);
}

13
spring-integration-core/src/main/java/org/springframework/integration/message/MessageMapper.java
View File

@@ -17,20 +17,15 @@
package org.springframework.integration.message;
/**
* Strategy interface for mapping between messages and objects.
* Strategy interface for mapping from a {@link Message} to an Object.
*
* @author Mark Fisher
*/
public interface MessageMapper<M,O> {
public interface MessageMapper<P, O> {
/**
* Map to a {@link Message} from the given object.
* Map from the given {@link Message} to an Object.
*/
Message<M> toMessage(O source);
/**
* Map from the given {@link Message} to an object.
*/
O fromMessage(Message<M> message);
O mapMessage(Message<P> message);
}

41
spring-integration-core/src/main/java/org/springframework/integration/message/SimplePayloadMessageMapper.java
View File

@@ -1,41 +0,0 @@
/*
* Copyright 2002-2008 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.message;
/**
* A {@link MessageMapper} implementation that simply wraps and unwraps a
* payload object in a {@link Message}.
*
* @author Mark Fisher
*/
public class SimplePayloadMessageMapper<T> extends AbstractMessageMapper<T,T> {
/**
* Return the payload of the given Message.
*/
public T fromMessage(Message<T> message) {
return message.getPayload();
}
/**
* Return a {@link Message} with the given object as its payload.
*/
public Message<T> toMessage(T source) {
return new GenericMessage<T>(this.getIdGenerator().generateId(), source);
}
}

11
spring-integration-core/src/main/java/org/springframework/integration/adapter/Target.java → spring-integration-core/src/main/java/org/springframework/integration/message/Target.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -14,16 +14,15 @@
* limitations under the License.
*/
package org.springframework.integration.adapter;
package org.springframework.integration.message;
/**
* Interface for any external target that may receive data from an outgoing
* channel adapter.
* Interface for any target to which {@link Message Messages} can be sent.
*
* @author Mark Fisher
*/
public interface Target<T> {
public interface Target {
boolean send(T t);
boolean send(Message<?> message);
}

83
spring-integration-core/src/test/java/org/springframework/integration/adapter/DefaultTargetAdapterTests.java
View File

@@ -1,83 +0,0 @@
/*
* Copyright 2002-2008 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.adapter;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.scheduling.Subscription;
/**
* @author Mark Fisher
*/
public class DefaultTargetAdapterTests {
@Test
public void testAdapterSendsToChannel() throws Exception {
SynchronousQueue<String> queue = new SynchronousQueue<String>();
TestBean testBean = new TestBean(queue);
MethodInvokingTarget<TestBean> target = new MethodInvokingTarget<TestBean>();
target.setObject(testBean);
target.setMethod("foo");
target.afterPropertiesSet();
DefaultTargetAdapter adapter = new DefaultTargetAdapter(target);
SimpleChannel channel = new SimpleChannel();
Subscription subscription = new Subscription(channel);
Message<String> message = new GenericMessage<String>("123", "testing");
channel.send(message);
assertNull(queue.poll());
MessageBus bus = new MessageBus();
bus.registerChannel("channel", channel);
bus.registerHandler("targetAdapter", adapter, subscription);
bus.start();
String result = queue.poll(500, TimeUnit.MILLISECONDS);
assertNotNull(result);
assertEquals("testing", result);
bus.stop();
}
public static class TestBean {
private BlockingQueue<String> queue;
public TestBean(BlockingQueue<String> queue) {
this.queue = queue;
}
public void foo(String s) {
try {
this.queue.put(s);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}

77
spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java
View File

@@ -16,11 +16,24 @@
package org.springframework.integration.adapter;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.Subscription;
/**
* @author Mark Fisher
@@ -29,40 +42,82 @@ public class MethodInvokingTargetTests {
@Test
public void testValidMethod() {
MethodInvokingTarget<TestSink> target = new MethodInvokingTarget<TestSink>();
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(new TestSink());
target.setMethod("validMethod");
target.afterPropertiesSet();
boolean result = target.send("test");
boolean result = target.send(new GenericMessage<String>("test"));
assertTrue(result);
}
@Test(expected=MessagingException.class)
public void testInvalidMethodWithNoArgs() {
MethodInvokingTarget<TestSink> target = new MethodInvokingTarget<TestSink>();
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(new TestSink());
target.setMethod("invalidMethodWithNoArgs");
target.afterPropertiesSet();
target.send("test");
target.send(new StringMessage("test"));
}
@Test
public void testValidMethodWithIgnoredReturnValue() {
MethodInvokingTarget<TestSink> target = new MethodInvokingTarget<TestSink>();
@Test(expected=MessagingException.class)
public void testMethodWithReturnValue() {
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(new TestSink());
target.setMethod("validMethodWithIgnoredReturnValue");
target.setMethod("methodWithReturnValue");
target.afterPropertiesSet();
boolean result = target.send("test");
boolean result = target.send(new StringMessage("test"));
assertTrue(result);
}
@Test(expected=MessagingException.class)
public void testNoMatchingMethodName() {
MethodInvokingTarget<TestSink> target = new MethodInvokingTarget<TestSink>();
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(new TestSink());
target.setMethod("noSuchMethod");
target.afterPropertiesSet();
target.send("test");
target.send(new StringMessage("test"));
}
@Test
public void testSubscription() throws Exception {
SynchronousQueue<String> queue = new SynchronousQueue<String>();
TestBean testBean = new TestBean(queue);
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(testBean);
target.setMethod("foo");
target.afterPropertiesSet();
SimpleChannel channel = new SimpleChannel();
Subscription subscription = new Subscription(channel);
Message<String> message = new GenericMessage<String>("123", "testing");
channel.send(message);
assertNull(queue.poll());
MessageBus bus = new MessageBus();
bus.registerChannel("channel", channel);
bus.registerHandler("targetAdapter", target, subscription);
bus.start();
String result = queue.poll(500, TimeUnit.MILLISECONDS);
assertNotNull(result);
assertEquals("testing", result);
bus.stop();
}
public static class TestBean {
private BlockingQueue<String> queue;
public TestBean(BlockingQueue<String> queue) {
this.queue = queue;
}
public void foo(String s) {
try {
this.queue.put(s);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java
View File

@@ -38,8 +38,8 @@ public class TestSink {
public void invalidMethodWithNoArgs() {
}
public String validMethodWithIgnoredReturnValue(String s) {
return "ignored";
public String methodWithReturnValue(String s) {
return "value";
}
public void store(String s) {

14
spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml
View File

@@ -27,17 +27,13 @@
</constructor-arg>
</bean>
<bean id="targetAdapter" class="org.springframework.integration.adapter.DefaultTargetAdapter">
<constructor-arg>
<bean class="org.springframework.integration.adapter.MethodInvokingTarget">
<property name="object" ref="sink"/>
<property name="method" value="store"/>
</bean>
</constructor-arg>
<bean id="target" class="org.springframework.integration.adapter.MethodInvokingTarget">
<property name="object" ref="sink"/>
<property name="method" value="store"/>
</bean>
<bean id="targetEndpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
<constructor-arg ref="targetAdapter"/>
<bean id="targetEndpoint" class="org.springframework.integration.endpoint.TargetEndpoint">
<constructor-arg ref="target"/>
<property name="subscription">
<bean class="org.springframework.integration.scheduling.Subscription">
<constructor-arg ref="channel"/>

6
spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java
View File

@@ -27,7 +27,7 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor;
import org.springframework.integration.dispatcher.SynchronousChannel;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessagingException;
@@ -55,7 +55,7 @@ public class SynchronousChannelSubscriptionTests {
@Test
public void testSendAndReceiveForRegisteredEndpoint() {
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new TestHandler());
HandlerEndpoint endpoint = new HandlerEndpoint(new TestHandler());
endpoint.setSubscription(new Subscription("sourceChannel"));
endpoint.setDefaultOutputChannelName("targetChannel");
bus.registerEndpoint("testEndpoint", endpoint);
@@ -83,7 +83,7 @@ public class SynchronousChannelSubscriptionTests {
public void testExceptionThrownFromRegisteredEndpoint() {
SimpleChannel errorChannel = new SimpleChannel();
bus.setErrorChannel(errorChannel);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
throw new RuntimeException("intentional test failure");
}

2
spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml
View File

@@ -10,7 +10,7 @@
<bean id="targetChannel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
<bean id="endpoint" class="org.springframework.integration.endpoint.HandlerEndpoint">
<constructor-arg ref="handler"/>
<property name="subscription">
<bean class="org.springframework.integration.scheduling.Subscription">

4
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java
View File

@@ -25,7 +25,7 @@ import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.SequenceSizeCompletionStrategy;
@@ -83,7 +83,7 @@ public class AggregatorAnnotationTests {
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context,
final String endpointName) {
MessageBus messageBus = getMessageBus(context);
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus.lookupEndpoint(endpointName + "-endpoint");
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint(endpointName + "-endpoint");
MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor(
messageHandlerChain).getPropertyValue("handlers")).get(0);

34
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java
View File

@@ -32,12 +32,12 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagePriority;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
/**
@@ -72,10 +72,10 @@ public class ChannelParserTests {
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
AtomicInteger counter = new AtomicInteger();
CountDownLatch latch = new CountDownLatch(1);
TestHandler handler1 = new TestHandler(counter, latch);
TestHandler handler2 = new TestHandler(counter, latch);
dispatcher.addHandler(handler1);
dispatcher.addHandler(handler2);
TestTarget target1 = new TestTarget(counter, latch);
TestTarget target2 = new TestTarget(counter, latch);
dispatcher.addTarget(target1);
dispatcher.addTarget(target2);
dispatcher.start();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -92,10 +92,10 @@ public class ChannelParserTests {
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
AtomicInteger counter = new AtomicInteger();
CountDownLatch latch = new CountDownLatch(1);
TestHandler handler1 = new TestHandler(counter, latch);
TestHandler handler2 = new TestHandler(counter, latch);
dispatcher.addHandler(handler1);
dispatcher.addHandler(handler2);
TestTarget target1 = new TestTarget(counter, latch);
TestTarget target2 = new TestTarget(counter, latch);
dispatcher.addTarget(target1);
dispatcher.addTarget(target2);
dispatcher.start();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -112,10 +112,10 @@ public class ChannelParserTests {
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
AtomicInteger counter = new AtomicInteger();
CountDownLatch latch = new CountDownLatch(2);
TestHandler handler1 = new TestHandler(counter, latch);
TestHandler handler2 = new TestHandler(counter, latch);
dispatcher.addHandler(handler1);
dispatcher.addHandler(handler2);
TestTarget target1 = new TestTarget(counter, latch);
TestTarget target2 = new TestTarget(counter, latch);
dispatcher.addTarget(target1);
dispatcher.addTarget(target2);
dispatcher.start();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -270,21 +270,21 @@ public class ChannelParserTests {
}
private static class TestHandler implements MessageHandler {
private static class TestTarget implements Target {
private AtomicInteger counter;
private CountDownLatch latch;
TestHandler(AtomicInteger counter, CountDownLatch latch) {
TestTarget(AtomicInteger counter, CountDownLatch latch) {
this.counter = counter;
this.latch = latch;
}
public Message<?> handle(Message<?> message) {
public boolean send(Message<?> message) {
this.counter.incrementAndGet();
this.latch.countDown();
return null;
return true;
}
}

4
spring-integration-core/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java
View File

@@ -26,7 +26,7 @@ import org.springframework.beans.factory.BeanCreationException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.CompletionStrategyAdapter;
@@ -58,7 +58,7 @@ public class CompletionStrategyAnnotationTests {
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context,
final String endpointName) {
MessageBus messageBus = getMessageBus(context);
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus
.lookupEndpoint(endpointName + "-endpoint");
MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor(

25
spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java
View File

@@ -29,14 +29,13 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.Target;
import org.springframework.integration.message.selector.MessageSelectorRejectedException;
import org.springframework.integration.util.ErrorHandler;
/**
* @author Mark Fisher
@@ -98,7 +97,7 @@ public class EndpointParserTests {
public void testDefaultConcurrency() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"endpointConcurrencyTests.xml", this.getClass());
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) context.getBean("defaultConcurrencyEndpoint");
HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("defaultConcurrencyEndpoint");
ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy();
assertEquals(ConcurrencyPolicy.DEFAULT_CORE_SIZE, concurrencyPolicy.getCoreSize());
assertEquals(ConcurrencyPolicy.DEFAULT_MAX_SIZE, concurrencyPolicy.getMaxSize());
@@ -110,7 +109,7 @@ public class EndpointParserTests {
public void testConfiguredConcurrency() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"endpointConcurrencyTests.xml", this.getClass());
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) context.getBean("configuredConcurrencyEndpoint");
HandlerEndpoint endpoint = (HandlerEndpoint) context.getBean("configuredConcurrencyEndpoint");
ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy();
assertEquals(7, concurrencyPolicy.getCoreSize());
assertEquals(77, concurrencyPolicy.getMaxSize());
@@ -122,12 +121,12 @@ public class EndpointParserTests {
public void testEndpointWithSelectorAccepts() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"endpointWithSelectors.xml", this.getClass());
MessageHandler endpoint = (MessageHandler) context.getBean("endpoint");
Target endpoint = (Target) context.getBean("endpoint");
((Lifecycle) endpoint).start();
Message<?> message = new StringMessage("test");
MessageChannel replyChannel = new SimpleChannel();
message.getHeader().setReturnAddress(replyChannel);
endpoint.handle(message);
endpoint.send(message);
Message<?> reply = replyChannel.receive(500);
assertNotNull(reply);
assertEquals("foo", reply.getPayload());
@@ -137,20 +136,20 @@ public class EndpointParserTests {
public void testEndpointWithSelectorRejects() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"endpointWithSelectors.xml", this.getClass());
MessageHandler endpoint = (MessageHandler) context.getBean("endpoint");
Target endpoint = (Target) context.getBean("endpoint");
((Lifecycle) endpoint).start();
endpoint.handle(new GenericMessage<Integer>(123));
endpoint.send(new GenericMessage<Integer>(123));
}
@Test
public void testCustomErrorHandler() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"endpointWithErrorHandler.xml", this.getClass());
MessageHandler endpoint = (MessageHandler) context.getBean("endpoint");
Target endpoint = (Target) context.getBean("endpoint");
TestErrorHandler errorHandler = (TestErrorHandler) context.getBean("errorHandler");
assertNull(errorHandler.getLastError());
Message<?> message = new StringMessage("test");
endpoint.handle(message);
endpoint.send(message);
Throwable error = errorHandler.getLastError();
assertEquals(MessageHandlingException.class, error.getClass());
MessageHandlingException exception = (MessageHandlingException) error;
@@ -161,11 +160,11 @@ public class EndpointParserTests {
public void testCustomReplyHandler() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"endpointWithReplyHandler.xml", this.getClass());
MessageHandler endpoint = (MessageHandler) context.getBean("endpoint");
Target endpoint = (Target) context.getBean("endpoint");
TestReplyHandler replyHandler = (TestReplyHandler) context.getBean("replyHandler");
assertNull(replyHandler.getLastMessage());
Message<?> message = new StringMessage("test");
endpoint.handle(message);
endpoint.send(message);
Message<?> reply = replyHandler.getLastMessage();
assertNotNull(reply);
assertEquals("foo", reply.getPayload());

167
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDispatcherTests.java
View File

@@ -21,10 +21,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -32,11 +29,9 @@ import org.junit.Test;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.ConcurrentHandler;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.InterceptingMessageHandler;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerRejectedExecutionException;
import org.springframework.integration.handler.TestHandlers;
@@ -44,6 +39,7 @@ import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.Target;
import org.springframework.integration.message.selector.PayloadTypeSelector;
import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
@@ -66,8 +62,8 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel();
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()));
dispatcher.addTarget(createEndpoint(handler1, true));
dispatcher.addTarget(createEndpoint(handler2, true));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -84,8 +80,8 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()));
dispatcher.addTarget(createEndpoint(handler1, true));
dispatcher.addTarget(createEndpoint(handler2, true));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -102,14 +98,14 @@ public class DefaultMessageDispatcherTests {
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
SimpleChannel channel = new SimpleChannel();
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler1, createExecutor());
inactiveHandler.destroy();
dispatcher.addHandler(inactiveHandler);
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()));
MessageEndpoint inactiveEndpoint = createEndpoint(handler1, true);
dispatcher.addTarget(inactiveEndpoint);
dispatcher.addTarget(createEndpoint(handler2, true));
dispatcher.addTarget(createEndpoint(handler3, true));
dispatcher.start();
inactiveEndpoint.stop();
channel.send(new StringMessage(1, "test"));
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
assertEquals("inactive handler should not have received message", 0, counter1.get());
@@ -126,14 +122,14 @@ public class DefaultMessageDispatcherTests {
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
ConcurrentHandler inactiveHandler = new ConcurrentHandler(handler2, createExecutor());
inactiveHandler.destroy();
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(inactiveHandler);
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()));
MessageEndpoint inactiveEndpoint = createEndpoint(handler2, true);
dispatcher.addTarget(createEndpoint(handler1, true));
dispatcher.addTarget(inactiveEndpoint);
dispatcher.addTarget(createEndpoint(handler3, true));
dispatcher.start();
inactiveEndpoint.stop();
channel.send(new StringMessage(1, "test"));
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
assertEquals("inactive handler should not have received message", 0, counter2.get());
@@ -151,25 +147,22 @@ public class DefaultMessageDispatcherTests {
@Test
public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException {
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
final AtomicInteger counter3 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(2);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true));
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()));
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
public Message<?> handle(Message<?> message) {
dispatcher.addTarget(createEndpoint(handler1, true));
dispatcher.addTarget(new Target() {
public boolean send(Message<?> message) {
throw new MessageHandlerRejectedExecutionException(message);
}
});
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()));
dispatcher.addTarget(createEndpoint(handler3, true));
SimpleChannel errorChannel = new SimpleChannel();
scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
dispatcher.start();
@@ -194,14 +187,14 @@ public class DefaultMessageDispatcherTests {
channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(handler1);
dispatcher.addHandler(new MessageHandler() {
dispatcher.addTarget(createEndpoint(handler1, false));
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
dispatcher.addHandler(handler2);
}, false));
dispatcher.addTarget(createEndpoint(handler2, false));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -218,8 +211,8 @@ public class DefaultMessageDispatcherTests {
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
dispatcher.addHandler(handler1);
dispatcher.addHandler(handler2);
dispatcher.addTarget(createEndpoint(handler1, false));
dispatcher.addTarget(createEndpoint(handler2, false));
SimpleChannel errorChannel = new SimpleChannel();
scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
dispatcher.start();
@@ -238,30 +231,26 @@ public class DefaultMessageDispatcherTests {
final AtomicInteger rejectedCounter1 = new AtomicInteger();
final AtomicInteger rejectedCounter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(4);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
SimpleChannel channel = new SimpleChannel();
channel.getDispatcherPolicy().setRejectionLimit(2);
channel.getDispatcherPolicy().setRetryInterval(3);
channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) {
@Override
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
rejectedCounter1.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
}, false));
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
rejectedCounter2.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
}, false));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -280,9 +269,7 @@ public class DefaultMessageDispatcherTests {
final AtomicInteger rejectedCounter2 = new AtomicInteger();
final AtomicInteger rejectedCounter3 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(5);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch);
final MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
dispatcherPolicy.setRejectionLimit(2);
dispatcherPolicy.setRetryInterval(3);
@@ -290,33 +277,30 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) {
@Override
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
rejectedCounter1.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
}, false));
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
if (rejectedCounter2.get() == 1) {
return super.handle(message);
return handler2.handle(message);
}
rejectedCounter2.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
dispatcher.addHandler(new ConcurrentHandler(handler3, createExecutor()) {
@Override
}, false));
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
rejectedCounter3.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
}, false));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -335,8 +319,8 @@ public class DefaultMessageDispatcherTests {
final AtomicInteger rejectedCounter1 = new AtomicInteger();
final AtomicInteger rejectedCounter2 = new AtomicInteger();
final CountDownLatch latch = new CountDownLatch(8);
MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
final MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch);
final MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch);
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true);
dispatcherPolicy.setRejectionLimit(5);
dispatcherPolicy.setRetryInterval(3);
@@ -344,28 +328,26 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy);
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
dispatcher.addHandler(new ConcurrentHandler(handler1, createExecutor()) {
@Override
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
if (rejectedCounter1.get() == 2) {
return super.handle(message);
return handler1.handle(message);
}
rejectedCounter1.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
dispatcher.addHandler(new ConcurrentHandler(handler2, createExecutor()) {
@Override
}, false));
dispatcher.addTarget(createEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
if (rejectedCounter2.get() == 4) {
return super.handle(message);
return handler2.handle(message);
}
rejectedCounter2.incrementAndGet();
latch.countDown();
throw new MessageHandlerRejectedExecutionException(message);
}
});
}, false));
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -385,12 +367,12 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel();
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1);
DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2);
HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class));
endpoint2.addMessageSelector(new PayloadTypeSelector(String.class));
dispatcher.addHandler(endpoint1);
dispatcher.addHandler(endpoint2);
dispatcher.addTarget(endpoint1);
dispatcher.addTarget(endpoint2);
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -411,28 +393,30 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel();
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1);
DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2);
final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
final HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class));
endpoint2.addMessageSelector(new PayloadTypeSelector(Integer.class));
MessageHandler interceptor1 = new InterceptingMessageHandler(endpoint1) {
@Override
public Message<?> handle(Message<?> message, MessageHandler target) {
endpoint1.start();
endpoint2.start();
MessageHandler interceptor1 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
attemptedCounter1.incrementAndGet();
attemptedLatch.countDown();
return target.handle(message);
endpoint1.send(message);
return null;
}
};
MessageHandler interceptor2 = new InterceptingMessageHandler(endpoint2) {
@Override
public Message<?> handle(Message<?> message, MessageHandler target) {
MessageHandler interceptor2 = new MessageHandler() {
public Message<?> handle(Message<?> message) {
attemptedCounter2.incrementAndGet();
attemptedLatch.countDown();
return target.handle(message);
endpoint2.send(message);
return null;
}
};
dispatcher.addHandler(interceptor1);
dispatcher.addHandler(interceptor2);
dispatcher.addTarget(createEndpoint(interceptor1, false));
dispatcher.addTarget(createEndpoint(interceptor2, false));
dispatcher.start();
attemptedLatch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, attemptedLatch.getCount());
@@ -454,14 +438,14 @@ public class DefaultMessageDispatcherTests {
SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true));
channel.send(new StringMessage(1, "test"));
DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel, scheduler);
DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint(handler1);
HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1);
endpoint1.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint(handler2);
HandlerEndpoint endpoint2 = new HandlerEndpoint(handler2);
endpoint2.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
endpoint1.addMessageSelector(new PayloadTypeSelector(Integer.class));
endpoint2.addMessageSelector(new PayloadTypeSelector(String.class));
dispatcher.addHandler(endpoint1);
dispatcher.addHandler(endpoint2);
dispatcher.addTarget(endpoint1);
dispatcher.addTarget(endpoint2);
dispatcher.start();
latch.await(2000, TimeUnit.MILLISECONDS);
assertEquals("messages should have been dispatched within allotted time", 0, latch.getCount());
@@ -470,8 +454,13 @@ public class DefaultMessageDispatcherTests {
}
private static ExecutorService createExecutor() {
return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
private static MessageEndpoint createEndpoint(MessageHandler handler, boolean asynchronous) {
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
if (asynchronous) {
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
}
endpoint.afterPropertiesSet();
return endpoint;
}
}

22
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.
@@ -25,8 +25,11 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.springframework.integration.channel.DispatcherPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.TestHandlers;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.Target;
/**
* @author Mark Fisher
@@ -37,7 +40,7 @@ public class DefaultMessageDistributorTests {
public void testSingleMessage() throws InterruptedException {
MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy());
final CountDownLatch latch = new CountDownLatch(1);
distributor.addHandler(TestHandlers.countDownHandler(latch));
distributor.addTarget(createEndpoint(TestHandlers.countDownHandler(latch)));
distributor.distribute(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -49,8 +52,8 @@ public class DefaultMessageDistributorTests {
final CountDownLatch latch = new CountDownLatch(1);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
distributor.addHandler(TestHandlers.countingCountDownHandler(counter1, latch));
distributor.addHandler(TestHandlers.countingCountDownHandler(counter2, latch));
distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
distributor.distribute(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -63,8 +66,8 @@ public class DefaultMessageDistributorTests {
final CountDownLatch latch = new CountDownLatch(2);
final AtomicInteger counter1 = new AtomicInteger();
final AtomicInteger counter2 = new AtomicInteger();
distributor.addHandler(TestHandlers.countingCountDownHandler(counter1, latch));
distributor.addHandler(TestHandlers.countingCountDownHandler(counter2, latch));
distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
distributor.addTarget(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
distributor.distribute(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
@@ -72,4 +75,11 @@ public class DefaultMessageDistributorTests {
assertEquals(1, counter2.get());
}
private static Target createEndpoint(MessageHandler handler) {
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.start();
return endpoint;
}
}

16
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java
View File

@@ -28,10 +28,10 @@ import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.message.Target;
/**
* @author Mark Fisher
@@ -44,7 +44,7 @@ public class SynchronousChannelTests {
@Test
public void testSend() {
SynchronousChannel channel = new SynchronousChannel();
channel.addHandler(new ThreadNameSettingTestHandler());
channel.addTarget(new ThreadNameSettingTestTarget());
StringMessage message = new StringMessage("test");
assertTrue(channel.send(message));
String handlerThreadName = message.getHeader().getProperty(HANDLER_THREAD);
@@ -82,7 +82,7 @@ public class SynchronousChannelTests {
public void testSendInSeparateThread() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
final SynchronousChannel channel = new SynchronousChannel();
channel.addHandler(new ThreadNameSettingTestHandler(latch));
channel.addTarget(new ThreadNameSettingTestTarget(latch));
final StringMessage message = new StringMessage("test");
new Thread(new Runnable() {
public void run() {
@@ -152,25 +152,25 @@ public class SynchronousChannelTests {
}
private static class ThreadNameSettingTestHandler implements MessageHandler {
private static class ThreadNameSettingTestTarget implements Target {
private final CountDownLatch latch;
ThreadNameSettingTestHandler() {
ThreadNameSettingTestTarget() {
this(null);
}
ThreadNameSettingTestHandler(CountDownLatch latch) {
ThreadNameSettingTestTarget(CountDownLatch latch) {
this.latch = latch;
}
public Message<?> handle(Message<?> message) {
public boolean send(Message<?> message) {
message.getHeader().setProperty(HANDLER_THREAD, Thread.currentThread().getName());
if (this.latch != null) {
this.latch.countDown();
}
return null;
return true;
}
}

125
spring-integration-core/src/test/java/org/springframework/integration/endpoint/DefaultMessageEndpointTests.java → spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java
View File

@@ -23,9 +23,6 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -48,7 +45,7 @@ import org.springframework.integration.util.ErrorHandler;
/**
* @author Mark Fisher
*/
public class DefaultMessageEndpointTests {
public class HandlerEndpointTests {
@Test
public void testDefaultReplyChannel() throws Exception {
@@ -60,11 +57,11 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
endpoint.send(new StringMessage(1, "test"));
endpoint.stop();
Message<?> reply = replyChannel.receive(50);
assertNotNull(reply);
@@ -79,11 +76,11 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.start();
StringMessage testMessage = new StringMessage(1, "test");
testMessage.getHeader().setReturnAddress(replyChannel);
endpoint.handle(testMessage);
endpoint.send(testMessage);
endpoint.stop();
Message<?> reply = replyChannel.receive(50);
assertNotNull(reply);
@@ -100,12 +97,12 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.start();
StringMessage testMessage = new StringMessage(1, "test");
testMessage.getHeader().setReturnAddress("replyChannel");
endpoint.handle(testMessage);
endpoint.send(testMessage);
endpoint.stop();
Message<?> reply = replyChannel.receive(50);
assertNotNull(reply);
@@ -123,19 +120,19 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.start();
StringMessage testMessage = new StringMessage("test");
testMessage.getHeader().setReturnAddress(replyChannel1);
endpoint.handle(testMessage);
endpoint.send(testMessage);
Message<?> reply1 = replyChannel1.receive(50);
assertNotNull(reply1);
assertEquals("hello test", reply1.getPayload());
Message<?> reply2 = replyChannel2.receive(0);
assertNull(reply2);
testMessage.getHeader().setReturnAddress("replyChannel2");
endpoint.handle(testMessage);
endpoint.send(testMessage);
reply1 = replyChannel1.receive(0);
assertNull(reply1);
reply2 = replyChannel2.receive(0);
@@ -147,7 +144,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testCustomErrorHandler() throws InterruptedException {
final CountDownLatch latch = new CountDownLatch(2);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.rejectingCountDownHandler(latch));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.rejectingCountDownHandler(latch));
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
endpoint.setErrorHandler(new ErrorHandler() {
public void handle(Throwable t) {
@@ -155,7 +152,7 @@ public class DefaultMessageEndpointTests {
}
});
endpoint.start();
endpoint.handle(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals("both handler and errorHandler should have been invoked", 0, latch.getCount());
}
@@ -172,13 +169,14 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new ConcurrentHandler(handler, createExecutor()));
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
endpoint.setChannelRegistry(channelRegistry);
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
endpoint.stop();
endpoint.send(new StringMessage(1, "test"));
latch.await(500, TimeUnit.MILLISECONDS);
endpoint.stop();
assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
Message<?> reply = replyChannel.receive(100);
assertNotNull(reply);
@@ -197,11 +195,11 @@ public class DefaultMessageEndpointTests {
return null;
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
endpoint.send(new StringMessage(1, "test"));
endpoint.stop();
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
@@ -221,13 +219,14 @@ public class DefaultMessageEndpointTests {
return null;
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new ConcurrentHandler(handler, createExecutor()));
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
endpoint.setChannelRegistry(channelRegistry);
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
endpoint.stop();
endpoint.send(new StringMessage(1, "test"));
latch.await(500, TimeUnit.MILLISECONDS);
endpoint.stop();
assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
Message<?> reply = replyChannel.receive(0);
assertNull(reply);
@@ -245,16 +244,17 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new ConcurrentHandler(handler, createExecutor()));
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(1, 1));
endpoint.setChannelRegistry(channelRegistry);
endpoint.start();
StringMessage message = new StringMessage(1, "test");
message.getHeader().setReturnAddress("replyChannel");
endpoint.handle(message);
endpoint.stop();
endpoint.send(message);
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
Message<?> reply = replyChannel.receive(100);
endpoint.stop();
assertNotNull(reply);
assertEquals("hello test", reply.getPayload());
}
@@ -271,14 +271,14 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14));
endpoint.setDefaultOutputChannelName("replyChannel");
endpoint.start();
endpoint.handle(new StringMessage(1, "test"));
endpoint.stop();
endpoint.send(new StringMessage(1, "test"));
latch.await(500, TimeUnit.MILLISECONDS);
endpoint.stop();
assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
Message<?> reply = replyChannel.receive(100);
assertNotNull(reply);
@@ -297,15 +297,15 @@ public class DefaultMessageEndpointTests {
return new StringMessage("123", "hello " + message.getPayload());
}
};
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(handler);
HandlerEndpoint endpoint = new HandlerEndpoint(handler);
endpoint.setChannelRegistry(channelRegistry);
endpoint.setConcurrencyPolicy(new ConcurrencyPolicy(3, 14));
endpoint.start();
StringMessage message = new StringMessage(1, "test");
message.getHeader().setReturnAddress("replyChannel");
endpoint.handle(message);
endpoint.stop();
endpoint.send(message);
latch.await(500, TimeUnit.MILLISECONDS);
endpoint.stop();
assertEquals("handler should have been invoked within allotted time", 0, latch.getCount());
Message<?> reply = replyChannel.receive(100);
assertNotNull(reply);
@@ -314,20 +314,20 @@ public class DefaultMessageEndpointTests {
@Test(expected=MessageHandlerNotRunningException.class)
public void testEndpointDoesNotHandleMessagesWhenNotYetStarted() {
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.nullHandler());
endpoint.handle(new StringMessage("test"));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.nullHandler());
endpoint.send(new StringMessage("test"));
}
@Test
public void testEndpointDoesNotHandleMessagesAfterBeingStopped() {
AtomicInteger counter = new AtomicInteger();
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.countingHandler(counter));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter));
boolean exceptionThrown = false;
try {
endpoint.start();
endpoint.handle(new StringMessage("test1"));
endpoint.send(new StringMessage("test1"));
endpoint.stop();
endpoint.handle(new StringMessage("test2"));
endpoint.send(new StringMessage("test2"));
}
catch (MessageHandlerNotRunningException e) {
exceptionThrown = true;
@@ -338,27 +338,27 @@ public class DefaultMessageEndpointTests {
@Test(expected=MessageSelectorRejectedException.class)
public void testEndpointWithSelectorRejecting() {
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.nullHandler());
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.nullHandler());
endpoint.addMessageSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
return false;
}
});
endpoint.start();
endpoint.handle(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
}
@Test
public void testEndpointWithSelectorAccepting() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.countDownHandler(latch));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countDownHandler(latch));
endpoint.addMessageSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
return true;
}
});
endpoint.start();
endpoint.handle(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
latch.await(100, TimeUnit.MILLISECONDS);
assertEquals("handler should have been invoked", 0, latch.getCount());
endpoint.stop();
@@ -367,7 +367,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testEndpointWithMultipleSelectorsAndFirstRejects() {
final AtomicInteger counter = new AtomicInteger();
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.countingHandler(counter));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter));
boolean exceptionThrown = false;
endpoint.addMessageSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -383,7 +383,7 @@ public class DefaultMessageEndpointTests {
});
endpoint.start();
try {
endpoint.handle(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
}
catch (MessageSelectorRejectedException e) {
exceptionThrown = true;
@@ -396,7 +396,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testEndpointWithMultipleSelectorsAndFirstAccepts() {
final AtomicInteger counter = new AtomicInteger();
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.countingHandler(counter));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter));
boolean exceptionThrown = false;
endpoint.addMessageSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
@@ -412,7 +412,7 @@ public class DefaultMessageEndpointTests {
});
endpoint.start();
try {
endpoint.handle(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
}
catch (MessageSelectorRejectedException e) {
exceptionThrown = true;
@@ -425,7 +425,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testEndpointWithMultipleSelectorsAndBothAccept() {
final AtomicInteger counter = new AtomicInteger();
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(TestHandlers.countingHandler(counter));
HandlerEndpoint endpoint = new HandlerEndpoint(TestHandlers.countingHandler(counter));
endpoint.addMessageSelector(new MessageSelector() {
public boolean accept(Message<?> message) {
counter.incrementAndGet();
@@ -439,7 +439,7 @@ public class DefaultMessageEndpointTests {
}
});
endpoint.start();
endpoint.handle(new StringMessage("test"));
endpoint.send(new StringMessage("test"));
assertEquals("both selectors and handler should have been invoked", 3, counter.get());
endpoint.stop();
}
@@ -449,7 +449,7 @@ public class DefaultMessageEndpointTests {
SimpleChannel output = new SimpleChannel(1);
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel("output", output);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
@@ -460,9 +460,9 @@ public class DefaultMessageEndpointTests {
endpoint.setErrorHandler(errorHandler);
endpoint.setReplyTimeout(0);
endpoint.start();
endpoint.handle(new StringMessage("test1"));
endpoint.send(new StringMessage("test1"));
assertNull(errorHandler.getLastError());
endpoint.handle(new StringMessage("test2"));
endpoint.send(new StringMessage("test2"));
Throwable error = errorHandler.getLastError();
assertNotNull(error);
assertEquals(MessageDeliveryException.class, error.getClass());
@@ -472,7 +472,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testReturnAddressChannelTimeoutSendsToErrorHandler() {
SimpleChannel replyChannel = new SimpleChannel(1);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
@@ -483,11 +483,11 @@ public class DefaultMessageEndpointTests {
endpoint.start();
Message<?> message1 = new StringMessage("test1");
message1.getHeader().setReturnAddress(replyChannel);
endpoint.handle(message1);
endpoint.send(message1);
assertNull(errorHandler.getLastError());
Message<?> message2 = new StringMessage("test2");
message2.getHeader().setReturnAddress(replyChannel);
endpoint.handle(message2);
endpoint.send(message2);
Throwable error = errorHandler.getLastError();
assertNotNull(error);
assertEquals(MessageDeliveryException.class, error.getClass());
@@ -499,7 +499,7 @@ public class DefaultMessageEndpointTests {
SimpleChannel replyChannel = new SimpleChannel(1);
ChannelRegistry channelRegistry = new DefaultChannelRegistry();
channelRegistry.registerChannel("replyChannel", replyChannel);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
@@ -511,11 +511,11 @@ public class DefaultMessageEndpointTests {
endpoint.start();
Message<?> message1 = new StringMessage("test1");
message1.getHeader().setReturnAddress("replyChannel");
endpoint.handle(message1);
endpoint.send(message1);
assertNull(errorHandler.getLastError());
Message<?> message2 = new StringMessage("test2");
message2.getHeader().setReturnAddress("replyChannel");
endpoint.handle(message2);
endpoint.send(message2);
Throwable error = errorHandler.getLastError();
assertNotNull(error);
assertEquals(MessageDeliveryException.class, error.getClass());
@@ -525,7 +525,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testCorrelationId() {
SimpleChannel replyChannel = new SimpleChannel(1);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
return message;
}
@@ -533,7 +533,7 @@ public class DefaultMessageEndpointTests {
endpoint.start();
Message<?> message = new StringMessage("test");
message.getHeader().setReturnAddress(replyChannel);
endpoint.handle(message);
endpoint.send(message);
Message<?> reply = replyChannel.receive(500);
assertEquals(message.getId(), reply.getHeader().getCorrelationId());
}
@@ -541,7 +541,7 @@ public class DefaultMessageEndpointTests {
@Test
public void testCorrelationIdSetByHandlerTakesPrecedence() {
SimpleChannel replyChannel = new SimpleChannel(1);
DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint(new MessageHandler() {
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
message.getHeader().setCorrelationId("ABC-123");
return message;
@@ -550,7 +550,7 @@ public class DefaultMessageEndpointTests {
endpoint.start();
Message<?> message = new StringMessage("test");
message.getHeader().setReturnAddress(replyChannel);
endpoint.handle(message);
endpoint.send(message);
Message<?> reply = replyChannel.receive(500);
Object correlationId = reply.getHeader().getCorrelationId();
assertFalse(message.getId().equals(correlationId));
@@ -558,11 +558,6 @@ public class DefaultMessageEndpointTests {
}
private static ExecutorService createExecutor() {
return new ThreadPoolExecutor(1, 1, 60, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
}
private static class TestErrorHandler implements ErrorHandler {
private volatile Throwable lastError;

4
spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java
View File

@@ -42,7 +42,7 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
@@ -130,7 +130,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
postProcessor.afterPropertiesSet();
ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
DefaultMessageEndpoint endpoint = (DefaultMessageEndpoint) messageBus.lookupEndpoint("testBean-endpoint");
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean-endpoint");
ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy();
assertEquals(17, concurrencyPolicy.getCoreSize());
assertEquals(42, concurrencyPolicy.getMaxSize());