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

Refactored MessageEndpointAnnotationPostProcessor to MessagingAnnotationPostProcessor which delegates to HandlerAnnotationPostProcessor, SourceAnnotationPostProcessor, and TargetAnnotationPostProcessor (work related to INT-194 and INT-195).

Browse Source
This commit is contained in:
Mark Fisher
2008-05-30 22:18:07 +00:00
parent bd80868ca5
commit 05524a330a
47 changed files with 1002 additions and 585 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
org.springframework.integration/src/main/java/org/springframework/integration/annotation/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.
@@ -41,6 +41,4 @@ public @interface MessageEndpoint {
String output() default "";
int pollPeriod() default 0;
}

41
org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* 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.annotation;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import org.springframework.integration.message.Message;
/**
* Indicates that a method is capable of producing messages. The method must
* accept no parameters and return either a {@link Message} or an Object to
* be passed as the message payload. The enclosing class may also be annotated
* with {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
*/
@java.lang.annotation.Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface MessageSource {
}

9
org.springframework.integration/src/main/java/org/springframework/integration/annotation/DefaultOutput.java → org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java
View File

@@ -25,9 +25,10 @@ import java.lang.annotation.RetentionPolicy;
import org.springframework.integration.message.Message;
/**
* Indicates that a method is capable of sending messages. The method must
* accept a single parameter that is either a {@link Message} or an Object to
* be passed as a message payload. The enclosing class should be annotated with
* Indicates that a method is capable of consuming messages. The method must
* accept a single parameter that is either a {@link Message} or an Object of
* the expected message payload type. The method itself should define a void
* return, and the enclosing class may also be annotated with
* {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
@@ -36,6 +37,6 @@ import org.springframework.integration.message.Message;
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface DefaultOutput {
public @interface MessageTarget {
}

10
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.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.
@@ -29,15 +29,11 @@ import org.springframework.integration.scheduling.PollingSchedule;
/**
* Annotation that can be specified at class-level alongside a
* {@link MessageEndpoint @MessageEndpoint} annotation in order to provide the
* scheduling information for that endpoint. Alternatively, as a method-level
* annotation, this indicates that a method is capable of providing messages.
* The method must not accept any parameters but can return either a single
* object or collection. The enclosing class should be annotated with
* {@link MessageEndpoint @MessageEndpoint}.
* scheduling information for that endpoint.
*
* @author Mark Fisher
*/
@Target({ElementType.TYPE, ElementType.METHOD})
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented

1
org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -32,6 +32,7 @@ import org.springframework.integration.channel.config.PriorityChannelParser;
import org.springframework.integration.channel.config.QueueChannelParser;
import org.springframework.integration.channel.config.RendezvousChannelParser;
import org.springframework.integration.channel.config.ThreadLocalChannelParser;
import org.springframework.integration.config.annotation.AnnotationDrivenParser;
import org.springframework.integration.gateway.config.GatewayParser;
import org.springframework.integration.router.config.RouterParser;
import org.springframework.integration.router.config.SplitterParser;

298
org.springframework.integration/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java
View File

@@ -1,298 +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.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
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.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.Concurrency;
import org.springframework.integration.annotation.DefaultOutput;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.MethodInvokingTarget;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.CompletionStrategyAdapter;
import org.springframework.integration.router.config.AggregatorMessageHandlerCreator;
import org.springframework.integration.router.config.RouterMessageHandlerCreator;
import org.springframework.integration.router.config.SplitterMessageHandlerCreator;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* A {@link BeanPostProcessor} implementation that generates endpoints for
* classes annotated with {@link MessageEndpoint @MessageEndpoint}.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean {
private final Log logger = LogFactory.getLog(this.getClass());
private final Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators = new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private final MessageBus messageBus;
public MessageEndpointAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "'messageBus' must not be null");
this.messageBus = messageBus;
}
public void setCustomHandlerCreators(Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) {
this.handlerCreators.put(entry.getKey(), entry.getValue());
}
}
public void afterPropertiesSet() {
this.handlerCreators.put(Handler.class, new DefaultMessageHandlerCreator());
this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator());
this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator());
this.handlerCreators.put(Aggregator.class, new AggregatorMessageHandlerCreator(messageBus));
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Class<?> beanClass = this.getBeanClass(bean);
MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class);
if (endpointAnnotation == null) {
return bean;
}
if (bean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus);
}
String outputChannelName = endpointAnnotation.output();
MessageHandlerChain handlerChain = this.createHandlerChain(bean, outputChannelName);
if (handlerChain == null) {
throw new ConfigurationException("@MessageEndpoint has no handler method");
}
HandlerEndpoint endpoint = new HandlerEndpoint(handlerChain);
Polled polledAnnotation = AnnotationUtils.findAnnotation(beanClass, Polled.class);
this.configureInput(bean, beanName, endpointAnnotation, polledAnnotation, endpoint);
if (StringUtils.hasText(outputChannelName)) {
endpoint.setOutputChannelName(outputChannelName);
}
else {
this.configureOutput(bean, beanName, endpoint);
}
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(beanClass, Concurrency.class);
if (concurrencyAnnotation != null) {
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(),
concurrencyAnnotation.maxSize());
concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
endpoint.setConcurrencyPolicy(concurrencyPolicy);
}
this.configureCompletionStrategy(bean, endpoint);
this.messageBus.registerEndpoint(beanName + "-endpoint", endpoint);
return bean;
}
private void configureInput(final Object bean, final String beanName, MessageEndpoint annotation,
Polled polledAnnotation, final HandlerEndpoint endpoint) {
String channelName = annotation.input();
if (StringUtils.hasText(channelName)) {
PollingSchedule schedule = null;
if (polledAnnotation != null) {
schedule = new PollingSchedule(polledAnnotation.period());
schedule.setInitialDelay(polledAnnotation.initialDelay());
schedule.setFixedRate(polledAnnotation.fixedRate());
schedule.setTimeUnit(polledAnnotation.timeUnit());
}
Subscription subscription = new Subscription(channelName, schedule);
endpoint.setSubscription(subscription);
}
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class);
if (annotation != null) {
Polled polledAnnotation = (Polled) annotation;
int period = polledAnnotation.period();
long initialDelay = polledAnnotation.initialDelay();
boolean fixedRate = polledAnnotation.fixedRate();
MethodInvokingSource source = new MethodInvokingSource();
source.setObject(bean);
source.setMethod(method.getName());
DirectChannel channel = new DirectChannel();
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
SourceEndpoint sourceEndpoint = new SourceEndpoint(source, channel, schedule);
String channelName = beanName + "-inputChannel";
messageBus.registerChannel(channelName, channel);
messageBus.registerEndpoint(beanName + "-sourceEndpoint", sourceEndpoint);
Subscription subscription = new Subscription(channel);
endpoint.setSubscription(subscription);
}
}
});
}
private void configureOutput(final Object bean, final String beanName, final HandlerEndpoint endpoint) {
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
boolean foundOutput = false;
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, DefaultOutput.class);
if (annotation != null) {
if (foundOutput) {
throw new ConfigurationException("only one @DefaultOutput allowed per endpoint");
}
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(bean);
target.setMethodName(method.getName());
target.afterPropertiesSet();
MessageHandler handler = endpoint.getHandler();
((MessageHandlerChain) handler).add(target);
foundOutput = true;
return;
}
}
});
}
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);
if (annotation != null) {
final MessageHandler endpointHandler = endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = null;
if (endpointHandler != null) {
if (endpointHandler instanceof MessageHandlerChain) {
for (MessageHandler handlerInChain : ((MessageHandlerChain) endpointHandler).getHandlers()) {
if (handlerInChain instanceof AggregatingMessageHandler) {
aggregatingMessageHandler = (AggregatingMessageHandler) handlerInChain;
break;
}
}
}
else if (endpointHandler instanceof AggregatingMessageHandler) {
aggregatingMessageHandler = (AggregatingMessageHandler) endpointHandler;
}
}
if (aggregatingMessageHandler == null) {
throw new ConfigurationException(
"@CompletionStrategy supported only when @Aggregator is present");
}
else {
aggregatingMessageHandler.setCompletionStrategy(new CompletionStrategyAdapter(bean, method));
}
}
}
});
}
@SuppressWarnings("unchecked")
private MessageHandlerChain createHandlerChain(final Object bean, final String outputChannelName) {
final List<MessageHandler> handlers = new ArrayList<MessageHandler>();
ReflectionUtils.doWithMethods(this.getBeanClass(bean), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
for (Annotation annotation : annotations) {
if (isHandlerAnnotation(annotation)) {
Map<String, Object> attributes = AnnotationUtils.getAnnotationAttributes(annotation);
attributes.put(AbstractMessageHandlerAdapter.OUTPUT_CHANNEL_NAME_KEY, outputChannelName);
MessageHandlerCreator handlerCreator = handlerCreators.get(annotation.annotationType());
if (handlerCreator == null) {
if (logger.isWarnEnabled()) {
logger.warn("No handler creator has been registered for handler annotation '"
+ annotation.annotationType() + "'");
}
}
else {
MessageHandler handler = handlerCreator.createHandler(bean, method, attributes);
if (handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) handler).setChannelRegistry(messageBus);
}
if (handler instanceof InitializingBean) {
try {
((InitializingBean) handler).afterPropertiesSet();
}
catch (Exception e) {
throw new ConfigurationException("failed to create handler", e);
}
}
if (handler != null) {
handlers.add(handler);
}
}
}
}
}
});
if (handlers.size() > 0) {
MessageHandlerChain handlerChain = new MessageHandlerChain();
Collections.sort(handlers, new OrderComparator());
for (MessageHandler handler : handlers) {
handlerChain.add(handler);
}
return handlerChain;
}
return null;
}
private Class<?> getBeanClass(Object bean) {
return AopUtils.getTargetClass(bean);
}
private boolean isHandlerAnnotation(Annotation annotation) {
return annotation.annotationType().equals(Handler.class)
|| annotation.annotationType().isAnnotationPresent(Handler.class)
|| this.handlerCreators.keySet().contains(annotation.annotationType());
}
}

123
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java Normal file
View File

@@ -0,0 +1,123 @@
/*
* 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.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DelegatingIntroductionInterceptor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* Base class for post-processing annotated methods.
*
* @author Mark Fisher
*/
public abstract class AbstractAnnotationMethodPostProcessor<T> implements AnnotationMethodPostProcessor {
protected final Log logger = LogFactory.getLog(this.getClass());
private final Class<? extends Annotation> annotationType;
private final MessageBus messageBus;
private final ClassLoader beanClassLoader;
public AbstractAnnotationMethodPostProcessor(Class<? extends Annotation> annotationType, MessageBus messageBus, ClassLoader beanClassLoader) {
Assert.notNull(annotationType, "Annotation type must not be null.");
Assert.notNull(messageBus, "MessageBus must not be null.");
this.annotationType = annotationType;
this.messageBus = messageBus;
this.beanClassLoader = (beanClassLoader != null) ? beanClassLoader : ClassUtils.getDefaultClassLoader();
}
protected MessageBus getMessageBus() {
return this.messageBus;
}
public Object postProcess(final Object bean, final String beanName, final Class<?> originalBeanClass) {
final List<T> results = new ArrayList<T>();
ReflectionUtils.doWithMethods(originalBeanClass, new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = getAnnotation(method);
if (annotation != null) {
T result = processMethod(bean, method, annotation);
if (result != null) {
results.add(result);
}
}
}
});
T postProcessedBean = (results.size() > 0) ? this.processResults(results) : null;
if (postProcessedBean == null) {
return bean;
}
ProxyFactory proxyFactory = new ProxyFactory(bean);
proxyFactory.addAdvice(new DelegatingIntroductionInterceptor(postProcessedBean));
return proxyFactory.getProxy(this.beanClassLoader);
}
private Annotation getAnnotation(Method method) {
Annotation[] annotations = AnnotationUtils.getAnnotations(method);
for (Annotation annotation : annotations) {
if (annotation.annotationType().equals(this.annotationType)
|| annotation.annotationType().isAnnotationPresent(this.annotationType)) {
return annotation;
}
}
return null;
}
protected Subscription createSubscription(final Object bean, final String beanName, MessageEndpoint annotation, Polled polledAnnotation) {
String channelName = annotation.input();
if (StringUtils.hasText(channelName)) {
PollingSchedule schedule = null;
if (polledAnnotation != null) {
schedule = new PollingSchedule(polledAnnotation.period());
schedule.setInitialDelay(polledAnnotation.initialDelay());
schedule.setFixedRate(polledAnnotation.fixedRate());
schedule.setTimeUnit(polledAnnotation.timeUnit());
}
Subscription subscription = new Subscription(channelName, schedule);
return subscription;
}
return null;
}
protected abstract T processMethod(Object bean, Method method, Annotation annotation);
protected abstract T processResults(List<T> results);
}

25
org.springframework.integration/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java → org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.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,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import org.w3c.dom.Element;
@@ -24,6 +24,7 @@ import org.springframework.beans.factory.parsing.BeanComponentDefinition;
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.config.MessageBusParser;
/**
* Parser for the <em>annotation-driven</em> element of the integration
@@ -39,18 +40,18 @@ public class AnnotationDrivenParser implements BeanDefinitionParser {
private static final String SUBSCRIBER_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.SubscriberAnnotationPostProcessor";
private static final String MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.MessageEndpointAnnotationPostProcessor";
private static final String MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME =
"internal.MessagingAnnotationPostProcessor";
public BeanDefinition parse(Element element, ParserContext parserContext) {
this.createPublisherPostProcessor(parserContext);
this.createSubscriberPostProcessor(parserContext);
this.createMessageEndpointPostProcessor(parserContext);
this.registerPublisherPostProcessor(parserContext);
this.registerSubscriberPostProcessor(parserContext);
this.registerMessagingAnnotationPostProcessor(parserContext);
return null;
}
private void createPublisherPostProcessor(ParserContext parserContext) {
private void registerPublisherPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(PublisherAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("channelRegistry",
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
@@ -59,7 +60,7 @@ public class AnnotationDrivenParser implements BeanDefinitionParser {
parserContext.registerBeanComponent(bcd);
}
private void createSubscriberPostProcessor(ParserContext parserContext) {
private void registerSubscriberPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(SubscriberAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("messageBus",
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
@@ -68,12 +69,12 @@ public class AnnotationDrivenParser implements BeanDefinitionParser {
parserContext.registerBeanComponent(bcd);
}
private void createMessageEndpointPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(MessageEndpointAnnotationPostProcessor.class);
private void registerMessagingAnnotationPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(MessagingAnnotationPostProcessor.class);
bd.getConstructorArgumentValues().addGenericArgumentValue(
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
BeanComponentDefinition bcd = new BeanComponentDefinition(
bd, MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
bd, MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME);
parserContext.registerBeanComponent(bcd);
}

33
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* 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.config.annotation;
import org.springframework.integration.endpoint.MessageEndpoint;
/**
* Strategy interface for post-processing annotated methods.
*
* @author Mark Fisher
*/
public interface AnnotationMethodPostProcessor {
Object postProcess(Object bean, String beanName, Class<?> originalBeanClass);
MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation);
}

139
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,139 @@
/*
* 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.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.Concurrency;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.annotation.Router;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.config.DefaultMessageHandlerCreator;
import org.springframework.integration.handler.config.MessageHandlerCreator;
import org.springframework.integration.router.config.AggregatorMessageHandlerCreator;
import org.springframework.integration.router.config.RouterMessageHandlerCreator;
import org.springframework.integration.router.config.SplitterMessageHandlerCreator;
import org.springframework.integration.scheduling.Subscription;
import org.springframework.util.StringUtils;
/**
* Post-processor for the {@link Handler @Handler} annotation.
*
* @author Mark Fisher
*/
public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<MessageHandler> {
private final Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private final MessageHandlerCreator defaultHandlerCreator = new DefaultMessageHandlerCreator();
public HandlerAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) {
super(Handler.class, messageBus, beanClassLoader);
this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator());
this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator());
this.handlerCreators.put(Aggregator.class, new AggregatorMessageHandlerCreator(messageBus));
}
public void setCustomHandlerCreators(Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) {
this.handlerCreators.put(entry.getKey(), entry.getValue());
}
}
protected MessageHandler processMethod(Object bean, Method method, Annotation annotation) {
MessageHandlerCreator handlerCreator = this.handlerCreators.get(annotation.annotationType());
if (handlerCreator == null) {
handlerCreator = this.defaultHandlerCreator;
if (logger.isDebugEnabled()) {
logger.debug("No handler creator has been registered for handler annotation '"
+ annotation.annotationType() + "', using DefaultMessageHandlerCreator.");
}
}
MessageHandler handler = handlerCreator.createHandler(bean, method, AnnotationUtils.getAnnotationAttributes(annotation));
if (handler != null) {
if (handler instanceof ChannelRegistryAware) {
((ChannelRegistryAware) handler).setChannelRegistry(this.getMessageBus());
}
if (handler instanceof InitializingBean) {
try {
((InitializingBean) handler).afterPropertiesSet();
}
catch (Exception e) {
throw new ConfigurationException("failed to initialize handler", e);
}
}
}
return handler;
}
protected MessageHandler processResults(List<MessageHandler> results) {
MessageHandlerChain handlerChain = new MessageHandlerChain();
for (MessageHandler handler : results) {
handlerChain.add(handler);
}
if (handlerChain.getHandlers().size() == 0) {
return null;
}
if (handlerChain.getHandlers().size() == 1) {
return handlerChain.getHandlers().get(0);
}
return handlerChain;
}
public MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) {
HandlerEndpoint endpoint = new HandlerEndpoint((MessageHandler) bean);
String outputChannelName = endpointAnnotation.output();
if (StringUtils.hasText(outputChannelName)) {
endpoint.setOutputChannelName(outputChannelName);
}
Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class);
Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation);
if (subscription != null) {
endpoint.setSubscription(subscription);
}
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class);
if (concurrencyAnnotation != null) {
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(
concurrencyAnnotation.coreSize(), concurrencyAnnotation.maxSize());
concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
endpoint.setConcurrencyPolicy(concurrencyPolicy);
}
return endpoint;
}
}

108
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,108 @@
/*
* 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.config.annotation;
import java.util.HashMap;
import java.util.Map;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Source;
import org.springframework.integration.message.Target;
import org.springframework.util.Assert;
/**
* A {@link BeanPostProcessor} implementation that processes method-level
* messaging annotations such as @Handler, @MessageSource, and @MessageTarget.
* It also generates endpoints for classes annotated with the class-level
* {@link MessageEndpoint @MessageEndpoint} annotation.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public class MessagingAnnotationPostProcessor implements BeanPostProcessor, InitializingBean, BeanClassLoaderAware {
private final MessageBus messageBus;
private volatile ClassLoader beanClassLoader;
private Map<Class<?>, AnnotationMethodPostProcessor> postProcessors = new HashMap<Class<?>, AnnotationMethodPostProcessor>();
public MessagingAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "MessageBus must not be null.");
this.messageBus = messageBus;
}
public void setBeanClassLoader(ClassLoader beanClassLoader) {
this.beanClassLoader = beanClassLoader;
}
protected MessageBus getMessageBus() {
return this.messageBus;
}
public void afterPropertiesSet() {
this.postProcessors.put(MessageHandler.class, new HandlerAnnotationPostProcessor(this.messageBus, this.beanClassLoader));
this.postProcessors.put(Source.class, new SourceAnnotationPostProcessor(this.messageBus, this.beanClassLoader));
this.postProcessors.put(Target.class, new TargetAnnotationPostProcessor(this.messageBus, this.beanClassLoader));
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
Object originalBean = bean;
Class<?> beanClass = this.getBeanClass(bean);
MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class);
for (Map.Entry<Class<?>, AnnotationMethodPostProcessor> entry : this.postProcessors.entrySet()) {
AnnotationMethodPostProcessor postProcessor = entry.getValue();
bean = postProcessor.postProcess(bean, beanName, beanClass);
if (endpointAnnotation != null && entry.getKey().isAssignableFrom(bean.getClass())) {
org.springframework.integration.endpoint.MessageEndpoint endpoint =
postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation);
if (endpoint != null) {
this.messageBus.registerEndpoint(beanName + "." + entry.getKey().getSimpleName() + ".endpoint", endpoint);
}
}
}
if (bean instanceof ChannelRegistryAware) {
((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus);
}
if (endpointAnnotation != null && bean.equals(originalBean)) {
throw new ConfigurationException("Class [" + beanClass.getName()
+ "] is annotated with @MessageEndpoint but contains no source, target, or handler method annotations.");
}
return bean;
}
protected Class<?> getBeanClass(Object bean) {
return AopUtils.getTargetClass(bean);
}
}

10
org.springframework.integration/src/main/java/org/springframework/integration/config/PublisherAnnotationPostProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PublisherAnnotationPostProcessor.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
@@ -38,9 +38,9 @@ import org.springframework.util.Assert;
*/
public class PublisherAnnotationPostProcessor implements BeanPostProcessor, BeanClassLoaderAware {
private Class<? extends Annotation> publisherAnnotationType = Publisher.class;
private volatile Class<? extends Annotation> publisherAnnotationType = Publisher.class;
private String channelNameAttribute = "channel";
private volatile String channelNameAttribute = "channel";
private ChannelRegistry channelRegistry;
@@ -86,8 +86,8 @@ public class PublisherAnnotationPostProcessor implements BeanPostProcessor, Bean
if (targetClass == null) {
return bean;
}
if (advisor == null) {
createAdvisor();
if (this.advisor == null) {
this.createAdvisor();
}
if (AopUtils.canApply(this.advisor, targetClass)) {
if (bean instanceof Advised) {

82
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,82 @@
/*
* 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.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.List;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.MessageSource;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.message.Source;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.util.StringUtils;
/**
* Post-processor for classes annotated with {@link MessageSource @MessageSource}.
*
* @author Mark Fisher
*/
public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<Source<?>> {
public SourceAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) {
super(MessageSource.class, messageBus, beanClassLoader);
}
protected Source<?> processMethod(Object bean, Method method, Annotation annotation) {
MethodInvokingSource source = new MethodInvokingSource();
source.setObject(bean);
source.setMethod(method.getName());
return source;
}
protected Source<?> processResults(List<Source<?>> results) {
if (results.size() > 1) {
throw new ConfigurationException("At most one @MessageSource annotation is allowed per class.");
}
return (results.size() == 1) ? results.get(0) : null;
}
public MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) {
Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class);
int period = polledAnnotation.period();
long initialDelay = polledAnnotation.initialDelay();
boolean fixedRate = polledAnnotation.fixedRate();
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
String outputChannelName = endpointAnnotation.output();
MessageChannel outputChannel = (StringUtils.hasText(outputChannelName)) ?
this.getMessageBus().lookupChannel(outputChannelName) : null;
if (outputChannel == null) {
outputChannel = new DirectChannel();
this.getMessageBus().registerChannel(beanName + ".output", outputChannel);
}
return new SourceEndpoint((Source<?>) bean, outputChannel, schedule);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/SubscriberAnnotationPostProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;

79
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java Normal file
View File

@@ -0,0 +1,79 @@
/*
* 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.config.annotation;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.List;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Concurrency;
import org.springframework.integration.annotation.MessageTarget;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.TargetEndpoint;
import org.springframework.integration.handler.MethodInvokingTarget;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.Subscription;
/**
* Post-processor for classes annotated with {@link MessageTarget @MessageTarget}.
*
* @author Mark Fisher
*/
public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<Target> {
public TargetAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) {
super(MessageTarget.class, messageBus, beanClassLoader);
}
public Target processMethod(Object bean, Method method, Annotation annotation) {
MethodInvokingTarget target = new MethodInvokingTarget();
target.setObject(bean);
target.setMethod(method);
return target;
}
protected Target processResults(List<Target> results) {
if (results.size() > 1) {
throw new ConfigurationException("At most one @MessageTarget annotation is allowed per class.");
}
return (results.size() == 1) ? results.get(0) : null;
}
public MessageEndpoint createEndpoint(Object bean, String beanName, Class<?> originalBeanClass,
org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) {
TargetEndpoint endpoint = new TargetEndpoint((Target) bean);
Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class);
Subscription subscription = this.createSubscription(bean, beanName, endpointAnnotation, polledAnnotation);
endpoint.setSubscription(subscription);
Concurrency concurrencyAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Concurrency.class);
if (concurrencyAnnotation != null) {
ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy(concurrencyAnnotation.coreSize(),
concurrencyAnnotation.maxSize());
concurrencyPolicy.setKeepAliveSeconds(concurrencyAnnotation.keepAliveSeconds());
concurrencyPolicy.setQueueCapacity(concurrencyAnnotation.queueCapacity());
endpoint.setConcurrencyPolicy(concurrencyPolicy);
}
return endpoint;
}
}

19
org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.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,14 +16,19 @@
package org.springframework.integration.router.config;
import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.config.AbstractMessageHandlerCreator;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.AggregatorAdapter;
import org.springframework.integration.router.CompletionStrategyAdapter;
import org.springframework.util.ReflectionUtils;
/**
* Creates an {@link AggregatorAdapter AggregatorAdapter} for methods that aggregate messages.
@@ -82,7 +87,19 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat
messageHandler.setTrackedCorrelationIdCapacity(
(Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY));
}
this.configureCompletionStrategy(object, messageHandler);
return messageHandler;
}
private void configureCompletionStrategy(final Object object, final AggregatingMessageHandler handler) {
ReflectionUtils.doWithMethods(object.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CompletionStrategy.class);
if (annotation != null) {
handler.setCompletionStrategy(new CompletionStrategyAdapter(object, method));
}
}
});
}
}

9
org.springframework.integration/src/main/java/org/springframework/integration/router/config/SplitterMessageHandlerCreator.java
View File

@@ -19,6 +19,9 @@ package org.springframework.integration.router.config;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.aop.support.AopUtils;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.config.AbstractMessageHandlerCreator;
@@ -33,6 +36,12 @@ public class SplitterMessageHandlerCreator extends AbstractMessageHandlerCreator
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
String outputChannelName = (String) attributes.get(AbstractMessageHandlerAdapter.OUTPUT_CHANNEL_NAME_KEY);
if (outputChannelName == null) {
MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(AopUtils.getTargetClass(object), MessageEndpoint.class);
if (endpointAnnotation != null) {
outputChannelName = endpointAnnotation.output();
}
}
return new SplitterMessageHandlerAdapter(object, method, outputChannelName);
}

2
org.springframework.integration/src/test/java/org/springframework/integration/aop/PublisherAnnotationPostProcessorTests.java
View File

@@ -37,7 +37,7 @@ public class PublisherAnnotationPostProcessorTests {
ITestBean testBean = (ITestBean) context.getBean("testBean");
testBean.test();
MessageChannel channel = (MessageChannel) context.getBean("testChannel");
Message result = channel.receive();
Message<?> result = channel.receive();
assertEquals("test", result.getPayload());
}

2
org.springframework.integration/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml
View File

@@ -10,7 +10,7 @@
<bean id="testBean" class="org.springframework.integration.aop.PublisherAnnotationTestBean"/>
<bean class="org.springframework.integration.config.PublisherAnnotationPostProcessor">
<bean class="org.springframework.integration.config.annotation.PublisherAnnotationPostProcessor">
<property name="channelRegistry" ref="messageBus"/>
</bean>

8
org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -26,7 +26,7 @@ import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor;
import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
@@ -69,7 +69,7 @@ public class DirectChannelSubscriptionTests {
@Test
public void testSendAndReceiveForAnnotatedEndpoint() {
MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(bus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(bus);
postProcessor.afterPropertiesSet();
TestEndpoint endpoint = new TestEndpoint();
postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint");
@@ -83,7 +83,7 @@ public class DirectChannelSubscriptionTests {
@Test(expected=MessagingException.class)
public void testExceptionThrownFromRegisteredEndpoint() {
QueueChannel errorChannel = new QueueChannel();
bus.setErrorChannel(errorChannel);
bus.setErrorChannel(errorChannel);
HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() {
public Message<?> handle(Message<?> message) {
throw new RuntimeException("intentional test failure");
@@ -100,7 +100,7 @@ public class DirectChannelSubscriptionTests {
public void testExceptionThrownFromAnnotatedEndpoint() {
QueueChannel errorChannel = new QueueChannel();
bus.setErrorChannel(errorChannel);
MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(bus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(bus);
postProcessor.afterPropertiesSet();
FailingTestEndpoint endpoint = new FailingTestEndpoint();
postProcessor.postProcessAfterInitialization(endpoint, "testEndpoint");

79
org.springframework.integration/src/test/java/org/springframework/integration/config/CompletionStrategyAnnotationTests.java
View File

@@ -1,79 +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.config;
import java.lang.reflect.Method;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
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.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.CompletionStrategyAdapter;
/**
* @author Marius Bogoevici
*/
public class CompletionStrategyAnnotationTests {
@Test
public void testAnnotationWithDefaultSettings() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotationAndCustomCompletionStrategy";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof CompletionStrategyAdapter);
DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor(
aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy")).getPropertyValue("invoker"));
Assert.assertSame(context.getBean(endpointName), invokerAccessor.getPropertyValue("object"));
Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method");
Assert.assertEquals("completionChecker", completionCheckerMethod.getName());
}
@Test(expected=BeanCreationException.class)
public void testInvalidAnnotation() {
new ClassPathXmlApplicationContext(new String[] {
"classpath:/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml" });
}
@SuppressWarnings("unchecked")
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context,
final String endpointName) {
MessageBus messageBus = getMessageBus(context);
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus
.lookupEndpoint(endpointName + "-endpoint");
MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor(
messageHandlerChain).getPropertyValue("handlers")).get(0);
DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler);
return aggregatingMessageHandlerAccessor;
}
private MessageBus getMessageBus(ApplicationContext context) {
MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
return messageBus;
}
}

1
org.springframework.integration/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java
View File

@@ -35,6 +35,7 @@ import org.springframework.integration.annotation.Subscriber;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.annotation.SubscriberAnnotationPostProcessor;
import org.springframework.integration.message.StringMessage;
/**

71
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorAnnotationTests.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -14,31 +14,38 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.util.List;
import java.lang.reflect.Method;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.aop.support.DelegatingIntroductionInterceptor;
import org.springframework.beans.DirectFieldAccessor;
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.config.MessageBusParser;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.router.AggregatingMessageHandler;
import org.springframework.integration.router.CompletionStrategyAdapter;
import org.springframework.integration.router.SequenceSizeCompletionStrategy;
/**
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class AggregatorAnnotationTests {
@Test
public void testAnnotationWithDefaultSettings() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" });
new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotation";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
@@ -59,7 +66,7 @@ public class AggregatorAnnotationTests {
@Test
public void testAnnotationWithCustomSettings() {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/testAnnotatedAggregator.xml" });
new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithCustomizedAnnotation";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
@@ -79,23 +86,49 @@ public class AggregatorAnnotationTests {
aggregatingMessageHandlerAccessor.getPropertyValue("trackedCorrelationIdCapacity"));
}
@SuppressWarnings("unchecked")
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context,
final String endpointName) {
MessageBus messageBus = getMessageBus(context);
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint(endpointName + "-endpoint");
MessageHandlerChain messageHandlerChain = (MessageHandlerChain) endpoint.getHandler();
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) ((List) new DirectFieldAccessor(
messageHandlerChain).getPropertyValue("handlers")).get(0);
DirectFieldAccessor aggregatingMessageHandlerAccessor = new DirectFieldAccessor(aggregatingMessageHandler);
return aggregatingMessageHandlerAccessor;
@Test
public void testAnnotationWithCustomCompletionStrategy() throws Exception {
ApplicationContext context = new ClassPathXmlApplicationContext(
new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" });
final String endpointName = "endpointWithDefaultAnnotationAndCustomCompletionStrategy";
DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context,
endpointName);
Assert.assertTrue(aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy") instanceof CompletionStrategyAdapter);
DirectFieldAccessor invokerAccessor = new DirectFieldAccessor(new DirectFieldAccessor(
aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy")).getPropertyValue("invoker"));
Assert.assertSame(((Advised) context.getBean(endpointName)).getTargetSource().getTarget(), invokerAccessor.getPropertyValue("object"));
Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method");
Assert.assertEquals("completionChecker", completionCheckerMethod.getName());
}
@Test(expected=BeanCreationException.class)
public void testInvalidCompletionStrategyAnnotation() {
new ClassPathXmlApplicationContext(new String[] {
"classpath:/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml" });
}
@SuppressWarnings("unchecked")
private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context, final String endpointName) {
MessageBus messageBus = this.getMessageBus(context);
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint(endpointName + ".MessageHandler.endpoint");
MessageHandler handler = endpoint.getHandler();
try {
if (AopUtils.isAopProxy(handler)) {
DelegatingIntroductionInterceptor interceptor = (DelegatingIntroductionInterceptor)
((Advised) handler).getAdvisors()[0].getAdvice();
Object delegate = new DirectFieldAccessor(interceptor).getPropertyValue("delegate");
return new DirectFieldAccessor(delegate);
}
}
catch (Exception e) {
// will return the accessor for the handler
}
return new DirectFieldAccessor(endpoint.getHandler());
}
private MessageBus getMessageBus(ApplicationContext context) {
MessageBus messageBus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
return messageBus;
return (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
}
}

195
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -14,25 +14,32 @@
* limitations under the License.
*/
package org.springframework.integration.endpoint.annotation;
package org.springframework.integration.config.annotation;
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.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.annotation.Concurrency;
import org.springframework.integration.annotation.DefaultOutput;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.MessageSource;
import org.springframework.integration.annotation.MessageTarget;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.annotation.Splitter;
import org.springframework.integration.bus.MessageBus;
@@ -40,9 +47,9 @@ import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor;
import org.springframework.integration.endpoint.ConcurrencyPolicy;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.PollingSchedule;
@@ -51,7 +58,47 @@ import org.springframework.integration.scheduling.Schedule;
/**
* @author Mark Fisher
*/
public class MessageEndpointAnnotationPostProcessorTests {
public class MessagingAnnotationPostProcessorTests {
@Test
public void testHandlerAnnotation() {
MessageBus messageBus = new MessageBus();
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
HandlerAnnotatedBean bean = new HandlerAnnotatedBean();
Object result = postProcessor.postProcessAfterInitialization(bean, "testBean");
assertTrue(result instanceof MessageHandler);
}
@Test
public void testCustomHandlerAnnotation() {
MessageBus messageBus = new MessageBus();
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
CustomHandlerAnnotatedBean bean = new CustomHandlerAnnotatedBean();
Object result = postProcessor.postProcessAfterInitialization(bean, "testBean");
assertTrue(result instanceof MessageHandler);
}
@Test
public void testSimpleHandlerWithContext() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"handlerAnnotationPostProcessorTests.xml", this.getClass());
MessageHandler handler = (MessageHandler) context.getBean("simpleHandler");
Message<?> reply = handler.handle(new StringMessage("world"));
assertEquals("hello world", reply.getPayload());
}
@Test
public void testSimpleHandlerEndpointWithContext() {
ApplicationContext context = new ClassPathXmlApplicationContext(
"handlerAnnotationPostProcessorTests.xml", this.getClass());
MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
inputChannel.send(new StringMessage("foo"));
Message<?> reply = outputChannel.receive(1000);
assertEquals("hello foo", reply.getPayload());
}
@Test
public void testSimpleHandler() throws InterruptedException {
@@ -104,31 +151,14 @@ public class MessageEndpointAnnotationPostProcessorTests {
}
@Test
public void testPolledAnnotation() throws InterruptedException {
public void testTargetAnnotation() throws InterruptedException {
MessageBus messageBus = new MessageBus();
QueueChannel testChannel = new QueueChannel();
messageBus.registerChannel("testChannel", testChannel);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
PolledAnnotationTestBean testBean = new PolledAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
messageBus.start();
Message<?> message = testChannel.receive(1000);
assertEquals("test", message.getPayload());
messageBus.stop();
}
@Test
public void testDefaultOutputAnnotation() throws InterruptedException {
MessageBus messageBus = new MessageBus();
QueueChannel testChannel = new QueueChannel();
messageBus.registerChannel("testChannel", testChannel);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
CountDownLatch latch = new CountDownLatch(1);
DefaultOutputAnnotationTestBean testBean = new DefaultOutputAnnotationTestBean(latch);
TargetAnnotationTestBean testBean = new TargetAnnotationTestBean(latch);
postProcessor.postProcessAfterInitialization(testBean, "testBean");
messageBus.start();
testChannel.send(new StringMessage("foo"));
@@ -141,12 +171,11 @@ public class MessageEndpointAnnotationPostProcessorTests {
@Test
public void testConcurrencyAnnotationWithValues() {
MessageBus messageBus = new MessageBus();
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
ConcurrencyAnnotationTestBean testBean = new ConcurrencyAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean-endpoint");
HandlerEndpoint endpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint");
ConcurrencyPolicy concurrencyPolicy = endpoint.getConcurrencyPolicy();
assertEquals(17, concurrencyPolicy.getCoreSize());
assertEquals(42, concurrencyPolicy.getMaxSize());
@@ -156,14 +185,13 @@ public class MessageEndpointAnnotationPostProcessorTests {
@Test(expected=IllegalArgumentException.class)
public void testPostProcessorWithNullMessageBus() {
new MessageEndpointAnnotationPostProcessor(null);
new MessagingAnnotationPostProcessor(null);
}
@Test
public void testChannelRegistryAwareBean() {
MessageBus messageBus = new MessageBus();
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
ChannelRegistryAwareTestBean testBean = new ChannelRegistryAwareTestBean();
assertNull(testBean.getChannelRegistry());
@@ -177,8 +205,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
public void testProxiedMessageEndpointAnnotation() {
MessageBus messageBus = new MessageBus();
messageBus.setAutoCreateChannels(true);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpoint());
Object proxy = proxyFactory.getProxy();
@@ -195,8 +222,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
public void testMessageEndpointAnnotationInherited() {
MessageBus messageBus = new MessageBus();
messageBus.setAutoCreateChannels(true);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointSubclass(), "subclass");
messageBus.start();
@@ -211,8 +237,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
public void testMessageEndpointAnnotationInheritedWithProxy() {
MessageBus messageBus = new MessageBus();
messageBus.setAutoCreateChannels(true);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointSubclass());
Object proxy = proxyFactory.getProxy();
@@ -232,8 +257,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageChannel outputChannel = new QueueChannel();
messageBus.registerChannel("inputChannel", inputChannel);
messageBus.registerChannel("outputChannel", outputChannel);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl");
messageBus.start();
@@ -246,8 +270,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
public void testMessageEndpointAnnotationInheritedFromInterfaceWithAutoCreatedChannels() {
MessageBus messageBus = new MessageBus();
messageBus.setAutoCreateChannels(true);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl");
messageBus.start();
@@ -265,8 +288,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageChannel outputChannel = new QueueChannel();
messageBus.registerChannel("inputChannel", inputChannel);
messageBus.registerChannel("outputChannel", outputChannel);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointImplementation());
Object proxy = proxyFactory.getProxy();
@@ -284,8 +306,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
QueueChannel output = new QueueChannel();
messageBus.registerChannel("input", input);
messageBus.registerChannel("output", output);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
SplitterAnnotationTestEndpoint endpoint = new SplitterAnnotationTestEndpoint();
postProcessor.postProcessAfterInitialization(endpoint, "endpoint");
@@ -311,8 +332,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageBus messageBus = new MessageBus();
QueueChannel testChannel = new QueueChannel();
messageBus.registerChannel("testChannel", testChannel);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
AnnotatedEndpointWithNoHandlerMethod endpoint = new AnnotatedEndpointWithNoHandlerMethod();
postProcessor.postProcessAfterInitialization(endpoint, "endpoint");
@@ -323,12 +343,11 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageBus messageBus = new MessageBus();
QueueChannel testChannel = new QueueChannel();
messageBus.registerChannel("testChannel", testChannel);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation();
postProcessor.postProcessAfterInitialization(endpoint, "testBean");
HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean-endpoint");
HandlerEndpoint processedEndpoint = (HandlerEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint");
Schedule schedule = processedEndpoint.getSubscription().getSchedule();
assertEquals(PollingSchedule.class, schedule.getClass());
PollingSchedule pollingSchedule = (PollingSchedule) schedule;
@@ -338,31 +357,31 @@ public class MessageEndpointAnnotationPostProcessorTests {
assertEquals(TimeUnit.SECONDS, pollingSchedule.getTimeUnit());
}
@MessageEndpoint(output="testChannel")
private static class PolledAnnotationTestBean {
@Polled(period=100)
public String poller() {
return "test";
}
@Handler
public Message<?> handle(Message<?> message) {
return message;
}
@Test
public void testMessageSourceAnnotation() {
MessageBus messageBus = new MessageBus();
QueueChannel testChannel = new QueueChannel();
messageBus.registerChannel("testChannel", testChannel);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus);
postProcessor.afterPropertiesSet();
MessageSourceAnnotationTestBean testBean = new MessageSourceAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
messageBus.start();
Message<?> message = testChannel.receive(1000);
assertEquals("test", message.getPayload());
messageBus.stop();
}
@MessageEndpoint(input="testChannel")
private static class DefaultOutputAnnotationTestBean {
private static class TargetAnnotationTestBean {
private String messageText;
private CountDownLatch latch;
public DefaultOutputAnnotationTestBean(CountDownLatch latch) {
public TargetAnnotationTestBean(CountDownLatch latch) {
this.latch = latch;
}
@@ -370,12 +389,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
return this.messageText;
}
@Handler
public Message<?> handle(Message<?> message) {
return message;
}
@DefaultOutput
@MessageTarget
public void countdown(String input) {
this.messageText = input;
latch.countDown();
@@ -418,7 +432,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
}
@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=25)
@MessageEndpoint(input="inputChannel", output="outputChannel")
private static interface SimpleAnnotatedEndpointInterface {
String test(String input);
}
@@ -458,4 +472,43 @@ public class MessageEndpointAnnotationPostProcessorTests {
}
}
private static class HandlerAnnotatedBean {
@Handler
public String test(String s) {
return s + s;
}
}
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
@Handler
private static @interface CustomHandler {
}
private static class CustomHandlerAnnotatedBean {
@CustomHandler
public String test(String s) {
return s + s;
}
}
@MessageEndpoint(output="testChannel")
@Polled(period=100)
private static class MessageSourceAnnotationTestBean {
@MessageSource
public String test() {
return "test";
}
}
}

5
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/SimpleAnnotatedEndpoint.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java
View File

@@ -14,15 +14,16 @@
* limitations under the License.
*/
package org.springframework.integration.endpoint.annotation;
package org.springframework.integration.config.annotation;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.endpoint.annotation.ITestEndpoint;
/**
* @author Mark Fisher
*/
@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=10)
@MessageEndpoint(input="inputChannel", output="outputChannel")
public class SimpleAnnotatedEndpoint implements ITestEndpoint {
@Handler

6
org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategy.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.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,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.util.ArrayList;
import java.util.Collections;
@@ -60,7 +60,7 @@ public class TestAnnotatedEndpointWithCompletionStrategy {
public boolean completionChecker(List<Message<?>> messages) {
return true;
}
public ConcurrentMap<Object, Message<?>> getAggregatedMessages() {
return aggregatedMessages;
}

14
org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCompletionStrategyOnly.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.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,21 +14,13 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CompletionStrategy;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.router.MessageSequenceComparator;
import org.springframework.stereotype.Component;
/**
@@ -40,7 +32,7 @@ public class TestAnnotatedEndpointWithCompletionStrategyOnly {
@CompletionStrategy
public boolean checkCompleteness(List<Message<?>> messages) {
throw new UnsupportedOperationException("Not intended to being called");
throw new UnsupportedOperationException("Not intended to be called");
}
}

2
org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithCustomizedAggregator.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.util.ArrayList;
import java.util.Collections;

2
org.springframework.integration/src/test/java/org/springframework/integration/config/TestAnnotatedEndpointWithDefaultAggregator.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.config;
package org.springframework.integration.config.annotation;
import java.util.ArrayList;
import java.util.Collections;

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/TypeConvertingTestEndpoint.java → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.endpoint.annotation;
package org.springframework.integration.config.annotation;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
@@ -22,7 +22,7 @@ import org.springframework.integration.annotation.MessageEndpoint;
/**
* @author Mark Fisher
*/
@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=10)
@MessageEndpoint(input="inputChannel", output="outputChannel")
public class TypeConvertingTestEndpoint {
@Handler

23
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/handlerAnnotationPostProcessorTests.xml Normal file
View File

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd">
<message-bus/>
<channel id="inputChannel"/>
<channel id="outputChannel"/>
<handler-endpoint input-channel="inputChannel" handler="simpleHandler" output-channel="outputChannel"/>
<beans:bean class="org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor">
<beans:constructor-arg ref="internal.MessageBus"/>
</beans:bean>
<beans:bean id="simpleHandler" class="org.springframework.integration.handler.annotation.SimpleHandlerTestBean"/>
</beans:beans>

0
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/messageParameterAnnotatedEndpointTests.xml → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/messageParameterAnnotatedEndpointTests.xml
View File

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointTests.xml
View File

@@ -13,9 +13,9 @@
<integration:channel id="outputChannel"/>
<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.SimpleAnnotatedEndpoint"/>
<bean id="endpoint" class="org.springframework.integration.config.annotation.SimpleAnnotatedEndpoint"/>
<bean class="org.springframework.integration.config.MessageEndpointAnnotationPostProcessor">
<bean class="org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor">
<constructor-arg ref="bus"/>
</bean>

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml
View File

@@ -8,9 +8,9 @@
<property name="autoCreateChannels" value="true"/>
</bean>
<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.SimpleAnnotatedEndpoint"/>
<bean id="endpoint" class="org.springframework.integration.config.annotation.SimpleAnnotatedEndpoint"/>
<bean class="org.springframework.integration.config.MessageEndpointAnnotationPostProcessor">
<bean class="org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor">
<constructor-arg ref="bus"/>
</bean>

4
org.springframework.integration/src/test/java/org/springframework/integration/config/testAnnotatedAggregator.xml → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml
View File

@@ -21,9 +21,9 @@
<context:component-scan base-package="org.springframework.integration.config" use-default-filters="false">
<context:include-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpoint.*"/>
expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpoint.*"/>
<context:exclude-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
</context:component-scan>
</beans:beans>

2
org.springframework.integration/src/test/java/org/springframework/integration/config/testInvalidCompletionStrategyAnnotation.xml → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml
View File

@@ -21,7 +21,7 @@
<context:component-scan base-package="org.springframework.integration.config" use-default-filters="false">
<context:include-filter type="regex"
expression="org\.springframework\.integration\.config\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
expression="org\.springframework\.integration\.config\.annotation\.TestAnnotatedEndpointWithCompletionStrategyOnly"/>
</context:component-scan>
</beans:beans>

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/typeConvertingEndpointTests.xml → org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml
View File

@@ -13,9 +13,9 @@
<integration:channel id="outputChannel"/>
<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.TypeConvertingTestEndpoint"/>
<bean id="endpoint" class="org.springframework.integration.config.annotation.TypeConvertingTestEndpoint"/>
<bean class="org.springframework.integration.config.MessageEndpointAnnotationPostProcessor">
<bean class="org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor">
<constructor-arg ref="bus"/>
</bean>

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java
View File

@@ -19,14 +19,16 @@ package org.springframework.integration.endpoint.annotation;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Polled;
import org.springframework.integration.annotation.MessageSource;
/**
* @author Mark Fisher
*/
@MessageEndpoint(output="outputChannel")
@Polled(period=100)
public class InboundChannelAdapterTestBean {
@Polled(period=100)
@MessageSource
public String getName() {
return "world";
}

2
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java
View File

@@ -24,7 +24,7 @@ import org.springframework.integration.message.StringMessage;
/**
* @author Mark Fisher
*/
@MessageEndpoint(input="inputChannel", output="outputChannel", pollPeriod=10)
@MessageEndpoint(input="inputChannel", output="outputChannel")
public class MessageParameterAnnotatedEndpoint {
@Handler

18
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/OutboundChannelAdapterTestBean.java → org.springframework.integration/src/test/java/org/springframework/integration/handler/annotation/SimpleHandlerTestBean.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,26 +14,18 @@
* limitations under the License.
*/
package org.springframework.integration.endpoint.annotation;
package org.springframework.integration.handler.annotation;
import org.springframework.integration.annotation.DefaultOutput;
import org.springframework.integration.annotation.Handler;
import org.springframework.integration.annotation.MessageEndpoint;
/**
* @author Mark Fisher
*/
@MessageEndpoint(input="inputChannel")
public class OutboundChannelAdapterTestBean {
public class SimpleHandlerTestBean {
@Handler
public String sayHello(String name) {
return "hello " + name;
}
@DefaultOutput
public void sendGreeting(String greeting) {
System.out.println(greeting);
public String sayHello(String input) {
return "hello " + input;
}
}