diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DeliveryAssembler.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DeliveryAssembler.java index 3e83668188..7c7fab9397 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DeliveryAssembler.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DeliveryAssembler.java @@ -24,10 +24,10 @@ import org.springframework.integration.annotation.MessageEndpoint; /** * @author Marius Bogoevici */ -@MessageEndpoint(input = "preparedDrinks",output = "deliveries") +@MessageEndpoint public class DeliveryAssembler { - @Aggregator + @Aggregator(inputChannel = "preparedDrinks", outputChannel = "deliveries") public Delivery prepareDelivery(List drinks) { return new Delivery(drinks); } diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DrinkRouter.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DrinkRouter.java index 2a04b1bec5..2546a9d6cd 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DrinkRouter.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/DrinkRouter.java @@ -22,10 +22,10 @@ import org.springframework.integration.annotation.Router; /** * @author Mark Fisher */ -@MessageEndpoint(input="drinks") +@MessageEndpoint public class DrinkRouter { - @Router + @Router(inputChannel="drinks") public String resolveOrderItemChannel(OrderItem orderItem) { return (orderItem.isIced()) ? "coldDrinks" : "hotDrinks"; } diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java index 6e3ed9f4bd..088da09461 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/cafe/OrderSplitter.java @@ -24,10 +24,10 @@ import org.springframework.integration.annotation.Splitter; /** * @author Mark Fisher */ -@MessageEndpoint(input="orders", output="drinks") +@MessageEndpoint public class OrderSplitter { - @Splitter + @Splitter(inputChannel="orders", outputChannel="drinks") public List split(Order order) { return order.getDrinks(); } diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java index 840eba411a..185c254cf7 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java @@ -26,10 +26,10 @@ import org.springframework.integration.annotation.MessageEndpoint; /** * @author Mark Fisher */ -@MessageEndpoint(input="tickers", output="quotes") +@MessageEndpoint public class QuoteService { - @Handler + @Handler(inputChannel="tickers", outputChannel="quotes") public Quote lookupQuote(String ticker) { BigDecimal price = new BigDecimal(new Random().nextDouble() * 100); return new Quote(ticker, price.setScale(2, RoundingMode.HALF_EVEN)); diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java index 825a530fe0..84cb1850c2 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java @@ -20,18 +20,16 @@ import java.util.Random; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.Pollable; import org.springframework.integration.annotation.Poller; /** * @author Mark Fisher */ @MessageEndpoint -@ChannelAdapter("tickers") -@Poller(period = 300) public class TickerStream { - @Pollable + @ChannelAdapter("tickers") + @Poller(period = 300) public String nextTicker() { char[] chars = new char[3]; for (int i = 0; i < 3; i++) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorMessageHandlerCreator.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorMessageHandlerCreator.java deleted file mode 100644 index 0b6e8b23f4..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregatorMessageHandlerCreator.java +++ /dev/null @@ -1,128 +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.aggregator; - -import java.lang.annotation.Annotation; -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.CompletionStrategy; -import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.channel.ChannelRegistry; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.handler.config.AbstractMessageHandlerCreator; -import org.springframework.util.ReflectionUtils; -import org.springframework.util.StringUtils; - -/** - * Creates an {@link AggregatorAdapter AggregatorAdapter} for methods that aggregate messages. - * - * @author Marius Bogoevici - */ -public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreator { - - private static final String DISCARD_CHANNEL = "discardChannel"; - - private static final String SEND_TIMEOUT = "sendTimeout"; - - private static final String SEND_PARTIAL_RESULTS_ON_TIMEOUT = "sendPartialResultsOnTimeout"; - - private static final String REAPER_INTERVAL = "reaperInterval"; - - private static final String TIMEOUT = "timeout"; - - private static final String TRACKED_CORRELATION_ID_CAPACITY = "trackedCorrelationIdCapacity"; - - - private final ChannelRegistry channelRegistry; - - - public AggregatorMessageHandlerCreator(ChannelRegistry channelRegistry) { - this.channelRegistry = channelRegistry; - } - - - public MessageHandler doCreateHandler(Object object, Method method, Map attributes) { - AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(object, method)); - this.configureDefaultReplyChannel(messageHandler, object); - String discardChannelName = this.getAttribute(attributes, DISCARD_CHANNEL, String.class); - if (discardChannelName != null) { - messageHandler.setDiscardChannel(this.channelRegistry.lookupChannel(discardChannelName)); - } - Long sendTimeout = this.getAttribute(attributes, SEND_TIMEOUT, Long.class); - if (sendTimeout != null) { - messageHandler.setSendTimeout(sendTimeout); - } - Boolean sendPartialResultOnTimeout = this.getAttribute( - attributes, SEND_PARTIAL_RESULTS_ON_TIMEOUT, Boolean.class); - if (sendPartialResultOnTimeout != null) { - messageHandler.setSendPartialResultOnTimeout(sendPartialResultOnTimeout); - } - Long reaperInterval = this.getAttribute(attributes, REAPER_INTERVAL, Long.class); - if (reaperInterval != null) { - messageHandler.setReaperInterval(reaperInterval); - } - Long timeout = this.getAttribute(attributes, TIMEOUT, Long.class); - if (timeout != null) { - messageHandler.setTimeout(timeout); - } - Integer trackedCorrelationIdCapacity = this.getAttribute( - attributes, TRACKED_CORRELATION_ID_CAPACITY, Integer.class); - if (trackedCorrelationIdCapacity != null) { - messageHandler.setTrackedCorrelationIdCapacity(trackedCorrelationIdCapacity); - } - this.configureCompletionStrategy(object, messageHandler); - return messageHandler; - } - - private void configureDefaultReplyChannel(AggregatingMessageHandler handler, Object originalObject) { - MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation( - AopUtils.getTargetClass(originalObject), MessageEndpoint.class); - if (endpointAnnotation != null) { - String outputChannelName = endpointAnnotation.output(); - if (StringUtils.hasText(outputChannelName)) { - handler.setOutputChannel(this.channelRegistry.lookupChannel(outputChannelName)); - } - } - } - - 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)); - } - } - }); - } - - @SuppressWarnings("unchecked") - private T getAttribute(Map attributes, String name, Class expectedType) { - Object value = attributes.get(name); - if (value == null || !expectedType.isAssignableFrom(value.getClass())) { - return null; - } - if (value instanceof String && !StringUtils.hasText((String) value)) { - return null; - } - return (T) value; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java index 732bb1ff95..55a7230d39 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Aggregator.java @@ -39,6 +39,16 @@ import org.springframework.integration.aggregator.AggregatingMessageHandler; @Handler public @interface Aggregator { + /** + * channel name for receiving messages to be aggregated + */ + String inputChannel() default ""; + + /** + * channel name for sending aggregated result messages + */ + String outputChannel() default ""; + /** * channel name for sending discarded messages (due to a timeout) */ diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java index 5ec0a077bb..0c490b16b2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java @@ -26,17 +26,26 @@ import java.lang.annotation.Target; import org.springframework.stereotype.Component; /** - * Indicates that a class is capable of serving as a message channel. + * Indicates that a method is capable of serving as a message channel. + * If the method accepts no arguments but does define a non-void return + * type, an inbound Channel Adapter will be created. If the method does + * accept an argument and has a void return, an outbound Channel Adapter + * will be created. If the method does not conform to either contract, + * an Exception will be thrown. * * @author Mark Fisher */ -@Target(ElementType.TYPE) +@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented @Component public @interface ChannelAdapter { + /** + * The name of the channel being adapted. If the channel + * name is not resolvable, a new channel will be created. + */ String value(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Handler.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Handler.java index 74e5c83ba4..075e80b4df 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Handler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Handler.java @@ -45,4 +45,8 @@ import java.lang.annotation.Target; @Documented public @interface Handler { + String inputChannel() default ""; + + String outputChannel() default ""; + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java index 49c0652006..2042473759 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageEndpoint.java @@ -26,7 +26,8 @@ import java.lang.annotation.Target; import org.springframework.stereotype.Component; /** - * Indicates that a class is capable of serving as a message endpoint. + * Stereotype annotation indicating that a class is capable of serving as a + * Message Endpoint. * * @author Mark Fisher */ @@ -37,8 +38,12 @@ import org.springframework.stereotype.Component; @Component public @interface MessageEndpoint { - String input() default ""; - - String output() default ""; + /** + * The value may indicate a suggestion for a logical component name, + * to be turned into a Spring bean in case of an autodetected component. + * + * @return the suggested component name, if any + */ + String value() default ""; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java deleted file mode 100644 index ad52b342f1..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageTarget.java +++ /dev/null @@ -1,42 +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.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 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 - */ -@java.lang.annotation.Target(ElementType.METHOD) -@Retention(RetentionPolicy.RUNTIME) -@Inherited -@Documented -public @interface MessageTarget { - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java index a68535a727..df59a8e6be 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java @@ -27,13 +27,13 @@ import java.util.concurrent.TimeUnit; 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 + * Annotation that can be specified at method-level alongside a Message Endpoint + * annotation (e.g. @Splitter, @ChannelAdapter, etc.) in order to provide the * polling metadata and scheduling information for that endpoint. * * @author Mark Fisher */ -@Target(ElementType.TYPE) +@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Transformer.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Transformer.java index 9adfbfe0cb..7315af0255 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Transformer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Transformer.java @@ -36,4 +36,8 @@ import java.lang.annotation.Target; @Handler public @interface Transformer { + String inputChannel() default ""; + + String outputChannel() default ""; + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java index 9bd8add796..9d66415b13 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ServiceActivatorParser.java @@ -16,9 +16,9 @@ package org.springframework.integration.config; +import org.springframework.integration.endpoint.DefaultServiceInvoker; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.ServiceActivatorEndpoint; -import org.springframework.integration.message.MessageMappingMethodInvoker; /** * Parser for the <service-activator> element. @@ -34,7 +34,7 @@ public class ServiceActivatorParser extends AbstractEndpointParser { @Override protected Class getMethodInvokingAdapterClass() { - return MessageMappingMethodInvoker.class; + return DefaultServiceInvoker.class; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java deleted file mode 100644 index c02317c469..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractAnnotationMethodPostProcessor.java +++ /dev/null @@ -1,102 +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.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.bus.MessageBus; -import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; -import org.springframework.util.ReflectionUtils; - -/** - * Base class for post-processing annotated methods. - * - * @author Mark Fisher - */ -public abstract class AbstractAnnotationMethodPostProcessor implements AnnotationMethodPostProcessor { - - protected final Log logger = LogFactory.getLog(this.getClass()); - - private final Class annotationType; - - private final MessageBus messageBus; - - private final ClassLoader beanClassLoader; - - - public AbstractAnnotationMethodPostProcessor(Class 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 results = new ArrayList(); - 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 abstract T processMethod(Object bean, Method method, Annotation annotation); - - protected abstract T processResults(List results); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java new file mode 100644 index 0000000000..d0a7648298 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -0,0 +1,131 @@ +/* + * 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 org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.ChannelRegistry; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.StringUtils; + +/** + * Base class for Method-level annotation post-processors. + * + * @author Mark Fisher + */ +public abstract class AbstractMethodAnnotationPostProcessor implements MethodAnnotationPostProcessor { + + private static final String INPUT_CHANNEL_ATTRIBUTE = "inputChannel"; + + private static final String OUTPUT_CHANNEL_ATTRIBUTE = "outputChannel"; + + + private final MessageBus messageBus; + + + public AbstractMethodAnnotationPostProcessor(MessageBus messageBus) { + Assert.notNull(messageBus, "MessageBus must not be null"); + this.messageBus = messageBus; + } + + + protected ChannelRegistry getChannelRegistry() { + return this.messageBus; + } + + public Object postProcess(Object bean, String beanName, Method method, T annotation) { + Object adapter = this.createMethodInvokingAdapter(bean, method, annotation); + if (adapter != null && this.shouldCreateEndpoint(annotation)) { + AbstractEndpoint endpoint = this.createEndpoint(adapter); + if (endpoint != null) { + Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class); + this.configureEndpoint(endpoint, annotation, pollerAnnotation); + if (endpoint.getName() == null) { + endpoint.setBeanName(this.generateEndpointName(beanName, annotation)); + } + return endpoint; + } + } + return adapter; + } + + protected boolean shouldCreateEndpoint(T annotation) { + return (StringUtils.hasText((String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE))); + } + + protected void configureEndpoint(AbstractEndpoint endpoint, T annotation, Poller pollerAnnotation) { + String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE); + if (StringUtils.hasText(inputChannelName)) { + MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName); + if (inputChannel == null) { + throw new ConfigurationException("unable to resolve inputChannel '" + inputChannelName + "'"); + } + if (pollerAnnotation != null) { + if (inputChannel instanceof PollableChannel) { + PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period()); + schedule.setInitialDelay(pollerAnnotation.initialDelay()); + schedule.setFixedRate(pollerAnnotation.fixedRate()); + schedule.setTimeUnit(pollerAnnotation.timeUnit()); + PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule); + poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll()); + endpoint.setSource(poller); + } + else { + throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource"); + } + } + else { + endpoint.setSource(inputChannel); + } + String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE); + if (StringUtils.hasText(outputChannelName)) { + MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName); + if (outputChannel == null) { + throw new ConfigurationException("unable to resolve outputChannel '" + outputChannelName + "'"); + } + endpoint.setTarget(outputChannel); + } + } + } + + private String generateEndpointName(String beanName, T annotation) { + String endpointName = beanName + "." + ClassUtils.getShortNameAsProperty(annotation.annotationType()); + String id = endpointName; + int counter = 0; + while (this.messageBus.lookupEndpoint(id) != null) { + id = endpointName + "#" + counter; + counter++; + } + return id; + } + + protected abstract Object createMethodInvokingAdapter(Object bean, Method method, T annotation); + + protected abstract AbstractEndpoint createEndpoint(Object adapter); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java new file mode 100644 index 0000000000..014a4fa4a6 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -0,0 +1,88 @@ +/* + * 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 org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.aggregator.AggregatorAdapter; +import org.springframework.integration.aggregator.CompletionStrategyAdapter; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.CompletionStrategy; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.DefaultEndpoint; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.util.ReflectionUtils; +import org.springframework.util.StringUtils; + +/** + * Post-processor for the {@link Aggregator @Aggregator} annotation. + * + * @author Mark Fisher + */ +public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public AggregatorAnnotationPostProcessor(MessageBus messageBus) { + super(messageBus); + } + + + @Override + protected Object createMethodInvokingAdapter(Object bean, Method method, Aggregator annotation) { + Aggregator aggregatorAnnotation = (Aggregator) annotation; + AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(bean, method)); + String outputChannelName = aggregatorAnnotation.outputChannel(); + if (StringUtils.hasText(outputChannelName)) { + messageHandler.setOutputChannel(this.getChannelRegistry().lookupChannel(outputChannelName)); + } + String discardChannelName = aggregatorAnnotation.discardChannel(); + if (StringUtils.hasText(discardChannelName)) { + messageHandler.setDiscardChannel(this.getChannelRegistry().lookupChannel(discardChannelName)); + } + messageHandler.setSendTimeout(aggregatorAnnotation.sendTimeout()); + messageHandler.setSendPartialResultOnTimeout(aggregatorAnnotation.sendPartialResultsOnTimeout()); + messageHandler.setReaperInterval(aggregatorAnnotation.reaperInterval()); + messageHandler.setTimeout(aggregatorAnnotation.timeout()); + messageHandler.setTrackedCorrelationIdCapacity(aggregatorAnnotation.trackedCorrelationIdCapacity()); + this.configureCompletionStrategy(bean, messageHandler); + messageHandler.afterPropertiesSet(); + return messageHandler; + } + + @Override + protected AbstractEndpoint createEndpoint(Object adapter) { + if (adapter instanceof MessageHandler) { + return new DefaultEndpoint((MessageHandler) adapter); + } + return null; + } + + 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)); + } + } + }); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java new file mode 100644 index 0000000000..a25905c125 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java @@ -0,0 +1,145 @@ +/* + * 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.reflect.Method; + +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.endpoint.InboundChannelAdapter; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.OutboundChannelAdapter; +import org.springframework.integration.handler.MethodInvokingTarget; +import org.springframework.integration.message.MethodInvokingSource; +import org.springframework.integration.message.PollableSource; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.util.Assert; + +/** + * Post-processor for methods annotated with {@link ChannelAdapter @ChannelAdapter}. + * + * @author Mark Fisher + */ +public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPostProcessor { + + private final MessageBus messageBus; + + + public ChannelAdapterAnnotationPostProcessor(MessageBus messageBus) { + Assert.notNull(messageBus, "MessageBus must not be null"); + this.messageBus = messageBus; + } + + + public Object postProcess(Object bean, String beanName, Method method, ChannelAdapter annotation) { + MessageEndpoint endpoint = null; + String channelName = annotation.value(); + MessageChannel channel = this.messageBus.lookupChannel(channelName); + if (channel == null) { + DirectChannel directChannel = new DirectChannel(); + directChannel.setBeanName(channelName); + this.messageBus.registerChannel(directChannel); + channel = directChannel; + } + Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class); + if (method.getParameterTypes().length == 0 && hasReturnValue(method)) { + MethodInvokingSource source = new MethodInvokingSource(); + source.setObject(bean); + source.setMethod(method); + endpoint = this.createInboundChannelAdapter(source, channel, pollerAnnotation); + } + else if (method.getParameterTypes().length > 0 && !hasReturnValue(method)) { + MethodInvokingTarget target = new MethodInvokingTarget(bean, method); + endpoint = this.createOutboundChannelAdapter(target, channel, pollerAnnotation); + } + else { + throw new ConfigurationException("The @ChannelAdapter can only be applied to methods that accept no arguments but have" + + " a return value (inbound) or methods that have no return value but do accept arguments (outbound)"); + } + if (endpoint != null) { + this.messageBus.registerEndpoint(endpoint); + } + return bean; + } + + private InboundChannelAdapter createInboundChannelAdapter(MethodInvokingSource source, MessageChannel channel, Poller pollerAnnotation) { + if (pollerAnnotation == null) { + throw new ConfigurationException("The @Poller annotation is required (at method-level) " + + "when using the @ChannelAdapter annotation with a no-arg method."); + } + PollingDispatcher poller = this.createPoller(source, pollerAnnotation); + InboundChannelAdapter adapter = new InboundChannelAdapter(); + adapter.setSource(poller); + adapter.setTarget(channel); + adapter.setBeanName(this.generateUniqueName(channel.getName() + ".inboundAdapter")); + return adapter; + } + + private OutboundChannelAdapter createOutboundChannelAdapter(MethodInvokingTarget target, MessageChannel channel, Poller pollerAnnotation) { + OutboundChannelAdapter adapter = new OutboundChannelAdapter(); + adapter.setTarget(target); + if (channel instanceof PollableChannel) { + PollingDispatcher poller = (pollerAnnotation != null) + ? this.createPoller((PollableChannel) channel, pollerAnnotation) + : new PollingDispatcher((PollableSource) channel, new PollingSchedule(0)); + adapter.setSource(poller); + } + else { + adapter.setSource(channel); + } + adapter.setBeanName(this.generateUniqueName(channel.getName() + ".outboundAdapter")); + return adapter; + } + + private PollingDispatcher createPoller(PollableSource source, Poller pollerAnnotation) { + PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period()); + schedule.setInitialDelay(pollerAnnotation.initialDelay()); + schedule.setFixedRate(pollerAnnotation.fixedRate()); + schedule.setTimeUnit(pollerAnnotation.timeUnit()); + PollingDispatcher poller = new PollingDispatcher((PollableSource) source, schedule); + int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll(); + if (maxMessagesPerPoll == -1) { + // the default is 1 since a MethodInvokingSource might return a non-null value + // every time it is invoked, thus producing an infinite number of messages per poll + maxMessagesPerPoll = 1; + } + poller.setMaxMessagesPerPoll(maxMessagesPerPoll); + return poller; + } + + private boolean hasReturnValue(Method method) { + return !method.getReturnType().equals(void.class); + } + + private String generateUniqueName(String name) { + int counter = 0; + String id = name; + while (this.messageBus.lookupEndpoint(id) != null) { + id = name + "#" + counter; + counter++; + } + return id; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java deleted file mode 100644 index 84c7aa7179..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/HandlerAnnotationPostProcessor.java +++ /dev/null @@ -1,134 +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.annotation; - -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.springframework.beans.factory.InitializingBean; -import org.springframework.core.OrderComparator; -import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.core.annotation.Order; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.aggregator.AggregatorMessageHandlerCreator; -import org.springframework.integration.annotation.Aggregator; -import org.springframework.integration.annotation.Handler; -import org.springframework.integration.annotation.Router; -import org.springframework.integration.annotation.Splitter; -import org.springframework.integration.annotation.Transformer; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.endpoint.DefaultEndpoint; -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.RouterMessageHandlerCreator; -import org.springframework.integration.splitter.SplitterMessageHandlerCreator; -import org.springframework.integration.transformer.config.TransformerMessageHandlerCreator; - -/** - * Post-processor for the {@link Handler @Handler} annotation. - * - * @author Mark Fisher - */ -public class HandlerAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor { - - private final Map, MessageHandlerCreator> handlerCreators = - new ConcurrentHashMap, 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)); - this.handlerCreators.put(Transformer.class, new TransformerMessageHandlerCreator()); - } - - - public void setCustomHandlerCreators(Map, MessageHandlerCreator> customHandlerCreators) { - for (Map.Entry, 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."); - } - } - Map attributes = AnnotationUtils.getAnnotationAttributes(annotation); - Order order = AnnotationUtils.findAnnotation(method, Order.class); - if (order != null) { - attributes.put("order", order.value()); - } - MessageHandler handler = handlerCreator.createHandler(bean, method, attributes); - 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; - } - - @SuppressWarnings("unchecked") - protected MessageHandler processResults(List 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); - } - List handlers = new ArrayList(handlerChain.getHandlers()); - Collections.sort(handlers, new OrderComparator()); - handlerChain.setHandlers(handlers); - return handlerChain; - } - - public AbstractEndpoint createEndpoint(Object bean) { - if (bean instanceof MessageHandler) { - return new DefaultEndpoint((MessageHandler) bean); - } - return null; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index 49abc74f93..76f8a9e5fc 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -17,40 +17,40 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; +import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.AopUtils; +import org.springframework.aop.support.DelegatingIntroductionInterceptor; 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.annotation.Poller; +import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.Handler; +import org.springframework.integration.annotation.Router; +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.annotation.Transformer; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; -import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.stereotype.Component; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; +import org.springframework.util.ClassUtils; +import org.springframework.util.ReflectionUtils; /** * 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. + * messaging annotations such as @Transformer, @Splitter, and @Router. * * @author Mark Fisher * @author Marius Bogoevici @@ -59,10 +59,10 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init private final MessageBus messageBus; - private volatile ClassLoader beanClassLoader; + private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader(); - private final Map, AnnotationMethodPostProcessor> postProcessors = - new HashMap, AnnotationMethodPostProcessor>(); + private final Map, MethodAnnotationPostProcessor> postProcessors = + new HashMap, MethodAnnotationPostProcessor>(); public MessagingAnnotationPostProcessor(MessageBus messageBus) { @@ -76,77 +76,69 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init } public void afterPropertiesSet() { - this.postProcessors.put(MessageHandler.class, new HandlerAnnotationPostProcessor(this.messageBus, this.beanClassLoader)); - this.postProcessors.put(MessageSource.class, new PollableAnnotationPostProcessor(this.messageBus, this.beanClassLoader)); - this.postProcessors.put(MessageTarget.class, new TargetAnnotationPostProcessor(this.messageBus, this.beanClassLoader)); + postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.messageBus)); + postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.messageBus)); + postProcessors.put(Handler.class, new ServiceActivatorAnnotationPostProcessor(this.messageBus)); + postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.messageBus)); + postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.messageBus)); + postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.messageBus)); } 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); + public Object postProcessAfterInitialization(Object bean, final String beanName) throws BeansException { + final Object originalBean = bean; + final Class beanClass = this.getBeanClass(bean); if (!this.isStereotype(beanClass)) { // we only post-process stereotype components return bean; } - MessageEndpoint endpointAnnotation = AnnotationUtils.findAnnotation(beanClass, MessageEndpoint.class); - for (Map.Entry, AnnotationMethodPostProcessor> entry : this.postProcessors.entrySet()) { - AnnotationMethodPostProcessor postProcessor = entry.getValue(); - bean = postProcessor.postProcess(bean, beanName, beanClass); - if (endpointAnnotation != null && entry.getKey().isAssignableFrom(bean.getClass())) { - AbstractEndpoint endpoint = postProcessor.createEndpoint(bean); - if (endpoint != null) { - endpoint.setBeanName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint"); - String inputChannelName = endpointAnnotation.input(); - if (!StringUtils.hasText(inputChannelName)) { - continue; - } - MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName); - if (inputChannel == null) { - throw new ConfigurationException("unable to resolve input channel '" + inputChannelName + "'"); - } - Poller pollerAnnotation = AnnotationUtils.findAnnotation(beanClass, Poller.class); - if (pollerAnnotation != null) { - if (inputChannel instanceof PollableChannel) { - PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period()); - schedule.setInitialDelay(pollerAnnotation.initialDelay()); - schedule.setFixedRate(pollerAnnotation.fixedRate()); - schedule.setTimeUnit(pollerAnnotation.timeUnit()); - PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule); - poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll()); - endpoint.setSource(poller); - } - else { - throw new ConfigurationException("The @Poller annotation should only be provided for a PollableSource"); + final ProxyFactory proxyFactory = new ProxyFactory(bean); + final AtomicBoolean isProxy = new AtomicBoolean(false); + ReflectionUtils.doWithMethods(beanClass, new ReflectionUtils.MethodCallback() { + @SuppressWarnings("unchecked") + public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + Annotation[] annotations = AnnotationUtils.getAnnotations(method); + for (Annotation annotation : annotations) { + MethodAnnotationPostProcessor postProcessor = postProcessors.get(annotation.annotationType()); + if (postProcessor != null) { + Object result = postProcessor.postProcess(originalBean, beanName, method, annotation); + if (result != null) { + if (result instanceof MessageEndpoint) { + messageBus.registerEndpoint((MessageEndpoint) result); + } + else { + boolean shouldProxy = false; + Class[] interfaces = ClassUtils.getAllInterfaces(result); + for (Class iface : interfaces) { + if (!iface.getPackage().getName().startsWith("org.springframework.integration")) { + continue; + } + if (proxyFactory.isInterfaceProxied(iface)) { + throw new ConfigurationException("interface [" + iface + "] is already proxied"); + } + shouldProxy = true; + } + if (result instanceof ChannelRegistryAware) { + ((ChannelRegistryAware) result).setChannelRegistry(messageBus); + } + if (shouldProxy) { + proxyFactory.addAdvice(new DelegatingIntroductionInterceptor(result)); + isProxy.set(true); + } + } } } - else { - endpoint.setSource(inputChannel); - } - String outputChannelName = endpointAnnotation.output(); - if (StringUtils.hasText(outputChannelName)) { - MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName); - if (outputChannel == null) { - throw new ConfigurationException("unable to resolve output channel '" + outputChannelName + "'"); - } - endpoint.setTarget(outputChannel); - } - this.messageBus.registerEndpoint(endpoint); } } - } + }); if (bean instanceof ChannelRegistryAware) { - ((ChannelRegistryAware) bean).setChannelRegistry(this.messageBus); + ((ChannelRegistryAware) bean).setChannelRegistry(messageBus); } - if (!bean.equals(originalBean) && originalBean instanceof ChannelRegistryAware) { - ((ChannelRegistryAware) originalBean).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."); + if (isProxy.get()) { + return proxyFactory.getProxy(this.beanClassLoader); } return bean; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java similarity index 76% rename from org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java index 9fd29dac19..932b0509af 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationMethodPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java @@ -16,17 +16,16 @@ package org.springframework.integration.config.annotation; -import org.springframework.integration.endpoint.AbstractEndpoint; +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; /** * Strategy interface for post-processing annotated methods. * * @author Mark Fisher */ -public interface AnnotationMethodPostProcessor { +public interface MethodAnnotationPostProcessor { - Object postProcess(Object bean, String beanName, Class originalBeanClass); - - AbstractEndpoint createEndpoint(Object bean); + Object postProcess(Object bean, String beanName, Method method, T annotation); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java deleted file mode 100644 index 238ce4cfe3..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java +++ /dev/null @@ -1,105 +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.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.ChannelAdapter; -import org.springframework.integration.annotation.Pollable; -import org.springframework.integration.annotation.Poller; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; -import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.endpoint.InboundChannelAdapter; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.MethodInvokingSource; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.scheduling.PollingSchedule; - -/** - * Post-processor for methods annotated with {@link Pollable @Pollable}. - * - * @author Mark Fisher - */ -public class PollableAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor> { - - public PollableAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { - super(Pollable.class, messageBus, beanClassLoader); - } - - - protected MessageSource processMethod(Object bean, Method method, Annotation annotation) { - MethodInvokingSource source = new MethodInvokingSource(); - source.setObject(bean); - source.setMethod(method); - ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class); - if (channelAdapterAnnotation != null) { - Poller pollerAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), Poller.class); - if (pollerAnnotation == null) { - throw new ConfigurationException("The @Poller annotation is required (at class-level) " - + "when using the @ChannelAdapter annotation with a @Pollable method annotation."); - } - PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period()); - schedule.setInitialDelay(pollerAnnotation.initialDelay()); - schedule.setFixedRate(pollerAnnotation.fixedRate()); - schedule.setTimeUnit(pollerAnnotation.timeUnit()); - PollingDispatcher poller = new PollingDispatcher((PollableSource) source, schedule); - int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll(); - if (maxMessagesPerPoll == -1) { - // the default is 1 since a MethodInvokingSource might return a non-null value - // every time it is invoked, thus producing an infinite number of messages per poll - maxMessagesPerPoll = 1; - } - poller.setMaxMessagesPerPoll(maxMessagesPerPoll); - InboundChannelAdapter adapter = new InboundChannelAdapter(); - adapter.setSource(poller); - String channelName = channelAdapterAnnotation.value(); - MessageChannel channel = this.getMessageBus().lookupChannel(channelName); - if (channel == null) { - adapter.setBeanName(channelName + ".adapter"); - DirectChannel directChannel = new DirectChannel(); - directChannel.setBeanName(channelName); - this.getMessageBus().registerChannel(directChannel); - channel = directChannel; - } - else { - adapter.setBeanName(channelName); - } - adapter.setTarget(channel); - this.getMessageBus().registerEndpoint(adapter); - } - return source; - } - - protected MessageSource processResults(List> results) { - if (results.size() > 1) { - throw new ConfigurationException("At most one @Pollable annotation is allowed per class."); - } - return (results.size() == 1) ? results.get(0) : null; - } - - public AbstractEndpoint createEndpoint(Object bean) { - return null; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java new file mode 100644 index 0000000000..d12ebce593 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java @@ -0,0 +1,69 @@ +/* + * 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.reflect.Method; + +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.annotation.Poller; +import org.springframework.integration.annotation.Router; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.router.MethodInvokingRouter; +import org.springframework.integration.router.RouterEndpoint; +import org.springframework.util.StringUtils; + +/** + * Post-processor for Methods annotated with {@link Router @Router}. + * + * @author Mark Fisher + */ +public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public RouterAnnotationPostProcessor(MessageBus messageBus) { + super(messageBus); + } + + + @Override + protected Object createMethodInvokingAdapter(Object bean, Method method, Router annotation) { + return new MethodInvokingRouter(bean, method); + } + + @Override + protected AbstractEndpoint createEndpoint(Object adapter) { + if (adapter instanceof MethodInvokingRouter) { + return new RouterEndpoint((MethodInvokingRouter) adapter); + } + return null; + } + + @Override + protected void configureEndpoint(AbstractEndpoint endpoint, Router annotation, Poller pollerAnnotation) { + super.configureEndpoint(endpoint, annotation, pollerAnnotation); + String defaultOutputChannelName = annotation.defaultOutputChannel(); + if (StringUtils.hasText(defaultOutputChannelName)) { + MessageChannel defaultOutputChannel = this.getChannelRegistry().lookupChannel(defaultOutputChannelName); + if (defaultOutputChannel == null) { + throw new ConfigurationException("unable to resolve defaultOutputChannel '" + defaultOutputChannelName + "'"); + } + ((RouterEndpoint) endpoint).setDefaultOutputChannel(defaultOutputChannel); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java new file mode 100644 index 0000000000..499fc5e4c8 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java @@ -0,0 +1,51 @@ +/* + * 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.reflect.Method; + +import org.springframework.integration.annotation.Handler; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.endpoint.DefaultServiceInvoker; +import org.springframework.integration.endpoint.ServiceActivatorEndpoint; +import org.springframework.integration.endpoint.ServiceInvoker; + +/** + * @author Mark Fisher + */ +public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public ServiceActivatorAnnotationPostProcessor(MessageBus messageBus) { + super(messageBus); + } + + + @Override + protected Object createMethodInvokingAdapter(Object bean, Method method, Handler annotation) { + return new DefaultServiceInvoker(bean, method); + } + + @Override + protected AbstractEndpoint createEndpoint(Object adapter) { + if (adapter instanceof ServiceInvoker) { + return new ServiceActivatorEndpoint((ServiceInvoker) adapter); + } + return null; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java new file mode 100644 index 0000000000..3d34e453d2 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java @@ -0,0 +1,52 @@ +/* + * 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.reflect.Method; + +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.splitter.MethodInvokingSplitter; +import org.springframework.integration.splitter.SplitterEndpoint; + +/** + * Post-processor for Methods annotated with {@link Splitter @Splitter}. + * + * @author Mark Fisher + */ +public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public SplitterAnnotationPostProcessor(MessageBus messageBus) { + super(messageBus); + } + + + @Override + protected Object createMethodInvokingAdapter(Object bean, Method method, Splitter annotation) { + return new MethodInvokingSplitter(bean, method); + } + + @Override + protected AbstractEndpoint createEndpoint(Object adapter) { + if (adapter instanceof MethodInvokingSplitter) { + return new SplitterEndpoint((MethodInvokingSplitter) adapter); + } + return null; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java index 6e9d0e8da7..5c0f93da04 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SubscriberAnnotationPostProcessor.java @@ -30,8 +30,8 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.Subscriber; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.endpoint.DefaultServiceInvoker; import org.springframework.integration.endpoint.ServiceActivatorEndpoint; -import org.springframework.integration.message.MessageMappingMethodInvoker; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.ReflectionUtils; @@ -92,7 +92,7 @@ public class SubscriberAnnotationPostProcessor implements BeanPostProcessor { if (!StringUtils.hasText(channelName)) { throw new ConfigurationException("no channel name provided for subscriber"); } - MessageMappingMethodInvoker invoker = new MessageMappingMethodInvoker(bean, method); + DefaultServiceInvoker invoker = new DefaultServiceInvoker(bean, method); invoker.afterPropertiesSet(); String endpointName = ClassUtils.getShortNameAsProperty(targetClass) + "." + method.getName() + ".endpoint"; diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java deleted file mode 100644 index 2b8cafe523..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java +++ /dev/null @@ -1,91 +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.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.ChannelAdapter; -import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.PollingDispatcher; -import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.endpoint.OutboundChannelAdapter; -import org.springframework.integration.handler.MethodInvokingTarget; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.scheduling.PollingSchedule; - -/** - * Post-processor for classes annotated with {@link MessageTarget @MessageTarget}. - * - * @author Mark Fisher - */ -public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor { - - public TargetAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { - super(org.springframework.integration.annotation.MessageTarget.class, messageBus, beanClassLoader); - } - - - protected MessageTarget processMethod(Object bean, Method method, Annotation annotation) { - MethodInvokingTarget target = new MethodInvokingTarget(bean, method); - ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class); - if (channelAdapterAnnotation != null) { - OutboundChannelAdapter adapter = new OutboundChannelAdapter(); - String channelName = channelAdapterAnnotation.value(); - MessageChannel channel = this.getMessageBus().lookupChannel(channelName); - if (channel == null) { - adapter.setBeanName(channelName + ".adapter"); - DirectChannel directChannel = new DirectChannel(); - directChannel.setBeanName(channelName); - this.getMessageBus().registerChannel(directChannel); - channel = directChannel; - } - else { - adapter.setBeanName(channelName); - } - if (channel instanceof PollableSource) { - // TODO: add poller config if period, etc is provided (add to @Pollable) - PollingDispatcher poller = new PollingDispatcher((PollableSource) channel, new PollingSchedule(0)); - adapter.setSource(poller); - } - else { - adapter.setSource(channel); - } - adapter.setTarget(target); - this.getMessageBus().registerEndpoint(adapter); - } - return target; - } - - protected MessageTarget processResults(List 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 AbstractEndpoint createEndpoint(Object bean) { - return null; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java new file mode 100644 index 0000000000..1e8a92b63c --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java @@ -0,0 +1,52 @@ +/* + * 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.reflect.Method; + +import org.springframework.integration.annotation.Transformer; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.transformer.MethodInvokingTransformer; +import org.springframework.integration.transformer.TransformerEndpoint; + +/** + * Post-processor for Methods annotated with {@link Transformer @Transformer}. + * + * @author Mark Fisher + */ +public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public TransformerAnnotationPostProcessor(MessageBus messageBus) { + super(messageBus); + } + + + @Override + protected Object createMethodInvokingAdapter(Object bean, Method method, Transformer annotation) { + return new MethodInvokingTransformer(bean, method); + } + + @Override + protected AbstractEndpoint createEndpoint(Object adapter) { + if (adapter instanceof MethodInvokingTransformer) { + return new TransformerEndpoint((MethodInvokingTransformer) adapter); + } + return null; + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultServiceInvoker.java similarity index 53% rename from org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.java rename to org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultServiceInvoker.java index e0d005c683..3b5da4eaa1 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategyOnly.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultServiceInvoker.java @@ -14,25 +14,29 @@ * limitations under the License. */ -package org.springframework.integration.config.annotation; +package org.springframework.integration.endpoint; -import java.util.List; +import java.lang.reflect.Method; -import org.springframework.integration.annotation.CompletionStrategy; -import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.message.Message; -import org.springframework.stereotype.Component; +import org.springframework.integration.message.MessageMappingMethodInvoker; /** - * @author Marius Bogoevici + * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel") -@Component("endpointWithoutAggregatorAndWithCompletionStrategy") -public class TestAnnotatedEndpointWithCompletionStrategyOnly { +public class DefaultServiceInvoker extends MessageMappingMethodInvoker implements ServiceInvoker { - @CompletionStrategy - public boolean checkCompleteness(List> messages) { - throw new UnsupportedOperationException("Not intended to be called"); + public DefaultServiceInvoker(Object object, Method method) { + super(object, method); + } + + public DefaultServiceInvoker(Object object, String methodName) { + super(object, methodName); + } + + + public Object invoke(Message message) { + return this.invokeMethod(message); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java index defb04d354..603793631c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceActivatorEndpoint.java @@ -19,7 +19,6 @@ package org.springframework.integration.endpoint; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageMappingMethodInvoker; import org.springframework.util.Assert; /** @@ -27,12 +26,12 @@ import org.springframework.util.Assert; */ public class ServiceActivatorEndpoint extends AbstractInOutEndpoint implements InitializingBean { - private final MessageMappingMethodInvoker invoker; + private final ServiceInvoker invoker; private final MessageHandler handler; - public ServiceActivatorEndpoint(MessageMappingMethodInvoker invoker) { + public ServiceActivatorEndpoint(ServiceInvoker invoker) { Assert.notNull(invoker, "invoker must not be null"); this.invoker = invoker; this.handler = null; @@ -46,15 +45,15 @@ public class ServiceActivatorEndpoint extends AbstractInOutEndpoint implements I public void afterPropertiesSet() throws Exception { - if (this.invoker != null) { - this.invoker.afterPropertiesSet(); + if (this.invoker != null && (this.invoker instanceof InitializingBean)) { + ((InitializingBean) this.invoker).afterPropertiesSet(); } } @Override protected Object handle(Message message) { if (this.invoker != null) { - return this.invoker.invokeMethod(message); + return this.invoker.invoke(message); } return this.handler.handle(message); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceInvoker.java similarity index 51% rename from org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java rename to org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceInvoker.java index fb6daae146..c2b5bcec67 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ServiceInvoker.java @@ -14,28 +14,15 @@ * 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; +package org.springframework.integration.endpoint; 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 ChannelAdapter @ChannelAdapter}. - * * @author Mark Fisher */ -@java.lang.annotation.Target(ElementType.METHOD) -@Retention(RetentionPolicy.RUNTIME) -@Inherited -@Documented -public @interface Pollable { +public interface ServiceInvoker { + + Object invoke(Message message); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingTarget.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingTarget.java index 1a147ba221..6c6121f62a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingTarget.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MethodInvokingTarget.java @@ -18,8 +18,8 @@ package org.springframework.integration.handler; import java.lang.reflect.Method; +import org.springframework.integration.endpoint.DefaultServiceInvoker; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageMappingMethodInvoker; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.MessagingException; @@ -28,7 +28,7 @@ import org.springframework.integration.message.MessagingException; * * @author Mark Fisher */ -public class MethodInvokingTarget extends MessageMappingMethodInvoker implements MessageTarget { +public class MethodInvokingTarget extends DefaultServiceInvoker implements MessageTarget { public MethodInvokingTarget(Object object, Method method) { super(object, method); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index 47248bef74..a0551d02fa 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -122,20 +122,20 @@ public class DirectChannelSubscriptionTests { } - @MessageEndpoint(input="sourceChannel", output="targetChannel") + @MessageEndpoint public static class TestEndpoint { - @Handler + @Handler(inputChannel="sourceChannel", outputChannel="targetChannel") public Message handle(Message message) { return new StringMessage(message.getPayload() + "-from-annotated-endpoint"); } } - @MessageEndpoint(input="sourceChannel", output="targetChannel") + @MessageEndpoint public static class FailingTestEndpoint { - @Handler + @Handler(inputChannel="sourceChannel", outputChannel="targetChannel") public Message handle(Message message) { throw new RuntimeException("intentional test failure"); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 0e7235a92d..ee5c3a691e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -25,7 +25,6 @@ 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.aggregator.AggregatingMessageHandler; @@ -91,27 +90,23 @@ public class AggregatorAnnotationTests { 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")); + DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); + Object completionStrategy = aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy"); + Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter); + CompletionStrategyAdapter completionStrategyAdapter = (CompletionStrategyAdapter) completionStrategy; + DirectFieldAccessor invokerAccessor = new DirectFieldAccessor( + new DirectFieldAccessor(completionStrategyAdapter).getPropertyValue("invoker")); + Object targetObject = invokerAccessor.getPropertyValue("object"); + Assert.assertSame(context.getBean(endpointName), targetObject); 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); - DefaultEndpoint endpoint = (DefaultEndpoint) messageBus.lookupEndpoint(endpointName + ".MessageHandler.endpoint"); + DefaultEndpoint endpoint = (DefaultEndpoint) messageBus.lookupEndpoint(endpointName + ".aggregator"); MessageHandler handler = (MessageHandler) new DirectFieldAccessor(endpoint).getPropertyValue("handler"); try { if (AopUtils.isAopProxy(handler)) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index d581b8b359..36cf03ccf2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -21,10 +21,6 @@ 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 java.util.concurrent.atomic.AtomicReference; @@ -35,15 +31,10 @@ import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.core.annotation.Order; -import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.MessageTarget; -import org.springframework.integration.annotation.Pollable; import org.springframework.integration.annotation.Poller; -import org.springframework.integration.annotation.Splitter; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.bus.MessageBus; @@ -53,8 +44,8 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.DefaultEndpoint; -import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.endpoint.ServiceActivatorEndpoint; +import org.springframework.integration.endpoint.ServiceInvoker; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.StringMessage; @@ -74,26 +65,16 @@ public class MessagingAnnotationPostProcessorTests { postProcessor.afterPropertiesSet(); HandlerAnnotatedBean bean = new HandlerAnnotatedBean(); Object result = postProcessor.postProcessAfterInitialization(bean, "testBean"); - assertTrue(result instanceof MessageHandler); - } - - @Test - public void testCustomHandlerAnnotation() { - MessageBus messageBus = new DefaultMessageBus(); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); - postProcessor.afterPropertiesSet(); - CustomHandlerAnnotatedBean bean = new CustomHandlerAnnotatedBean(); - Object result = postProcessor.postProcessAfterInitialization(bean, "testBean"); - assertTrue(result instanceof MessageHandler); + assertTrue(result instanceof ServiceInvoker); } @Test public void testSimpleHandlerWithContext() { AbstractApplicationContext 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()); + ServiceInvoker invoker = (ServiceInvoker) context.getBean("simpleHandler"); + String reply = (String) invoker.invoke(new StringMessage("world")); + assertEquals("hello world", reply); context.stop(); } @@ -303,49 +284,6 @@ public class MessagingAnnotationPostProcessorTests { messageBus.stop(); } - @Test - public void testSplitterAnnotation() throws InterruptedException { - MessageBus messageBus = new DefaultMessageBus(); - QueueChannel input = new QueueChannel(); - QueueChannel output = new QueueChannel(); - input.setBeanName("input"); - output.setBeanName("output"); - messageBus.registerChannel(input); - messageBus.registerChannel(output); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); - postProcessor.afterPropertiesSet(); - SplitterAnnotationTestEndpoint endpoint = new SplitterAnnotationTestEndpoint(); - postProcessor.postProcessAfterInitialization(endpoint, "endpoint"); - messageBus.start(); - input.send(new StringMessage("this.is.a.test")); - Message message1 = output.receive(500); - assertNotNull(message1); - assertEquals("this", message1.getPayload()); - Message message2 = output.receive(500); - assertNotNull(message2); - assertEquals("is", message2.getPayload()); - Message message3 = output.receive(500); - assertNotNull(message3); - assertEquals("a", message3.getPayload()); - Message message4 = output.receive(500); - assertNotNull(message4); - assertEquals("test", message4.getPayload()); - assertNull(output.receive(500)); - messageBus.stop(); - } - - @Test(expected=ConfigurationException.class) - public void testEndpointWithNoHandlerMethod() { - MessageBus messageBus = new DefaultMessageBus(); - QueueChannel testChannel = new QueueChannel(); - testChannel.setBeanName("testChannel"); - messageBus.registerChannel(testChannel); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); - postProcessor.afterPropertiesSet(); - AnnotatedEndpointWithNoHandlerMethod endpoint = new AnnotatedEndpointWithNoHandlerMethod(); - postProcessor.postProcessAfterInitialization(endpoint, "endpoint"); - } - @Test public void testEndpointWithPollerAnnotation() { MessageBus messageBus = new DefaultMessageBus(); @@ -356,7 +294,7 @@ public class MessagingAnnotationPostProcessorTests { postProcessor.afterPropertiesSet(); AnnotatedEndpointWithPolledAnnotation endpoint = new AnnotatedEndpointWithPolledAnnotation(); postProcessor.postProcessAfterInitialization(endpoint, "testBean"); - DefaultEndpoint processedEndpoint = (DefaultEndpoint) messageBus.lookupEndpoint("testBean.MessageHandler.endpoint"); + ServiceActivatorEndpoint processedEndpoint = (ServiceActivatorEndpoint) messageBus.lookupEndpoint("testBean.handler"); DirectFieldAccessor accessor = new DirectFieldAccessor(processedEndpoint); MessageSource source = (MessageSource) accessor.getPropertyValue("source"); assertTrue(source instanceof SubscribableSource); @@ -395,19 +333,19 @@ public class MessagingAnnotationPostProcessorTests { } @Test - public void testHandlerWithTransformers() { + public void testTransformer() { MessageBus messageBus = new DefaultMessageBus(); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); - HandlerWithTransformers testBean = new HandlerWithTransformers(); - MessageHandler handler = (MessageHandler) postProcessor.postProcessAfterInitialization(testBean, "testBean"); - Message reply = handler.handle(new StringMessage("foo")); - assertEquals("PRE.FOO.post", reply.getPayload()); + TransformerAnnotationTestBean testBean = new TransformerAnnotationTestBean(); + org.springframework.integration.transformer.Transformer transformer = + (org.springframework.integration.transformer.Transformer) postProcessor.postProcessAfterInitialization(testBean, "testBean"); + Message reply = transformer.transform(new StringMessage("foo")); + assertEquals("FOO", reply.getPayload()); } @MessageEndpoint - @ChannelAdapter("testChannel") private static class TargetAnnotationTestBean { private String messageText; @@ -423,7 +361,7 @@ public class MessagingAnnotationPostProcessorTests { return this.messageText; } - @MessageTarget + @ChannelAdapter("testChannel") public void countdown(String input) { this.messageText = input; latch.countDown(); @@ -431,7 +369,7 @@ public class MessagingAnnotationPostProcessorTests { } - @MessageEndpoint(input="inputChannel") + @MessageEndpoint private static class ChannelRegistryAwareTestBean implements ChannelRegistryAware { private ChannelRegistry channelRegistry; @@ -444,7 +382,7 @@ public class MessagingAnnotationPostProcessorTests { return this.channelRegistry; } - @Handler + @Handler(inputChannel="inputChannel") public Message handle(Message message) { return null; } @@ -455,7 +393,7 @@ public class MessagingAnnotationPostProcessorTests { } - @MessageEndpoint(input="inputChannel", output="outputChannel") + @MessageEndpoint private static interface SimpleAnnotatedEndpointInterface { String test(String input); } @@ -463,33 +401,18 @@ public class MessagingAnnotationPostProcessorTests { private static class SimpleAnnotatedEndpointImplementation implements SimpleAnnotatedEndpointInterface { - @Handler + @Handler(inputChannel="inputChannel", outputChannel="outputChannel") public String test(String input) { return "test-" + input; } } - @MessageEndpoint(input="input", output="output") - private static class SplitterAnnotationTestEndpoint { - - @Splitter - public String[] split(String input) { - return input.split("\\."); - } - } - - - @MessageEndpoint(input="testChannel") - private static class AnnotatedEndpointWithNoHandlerMethod { - } - - - @MessageEndpoint(input="testChannel") - @Poller(period=1234, initialDelay=5678, fixedRate=true, timeUnit=TimeUnit.SECONDS) + @MessageEndpoint private static class AnnotatedEndpointWithPolledAnnotation { - @Handler + @Handler(inputChannel="testChannel") + @Poller(period=1234, initialDelay=5678, fixedRate=true, timeUnit=TimeUnit.SECONDS) public String prependFoo(String s) { return "foo" + s; } @@ -507,30 +430,11 @@ public class MessagingAnnotationPostProcessorTests { } - @Target(ElementType.METHOD) - @Retention(RetentionPolicy.RUNTIME) - @Handler - private static @interface CustomHandler { - } - - @MessageEndpoint - private static class CustomHandlerAnnotatedBean { - - @CustomHandler - public String test(String s) { - return s + s; - } - - } - - - @MessageEndpoint - @ChannelAdapter("testChannel") - @Poller(period = 1000, initialDelay = 0, maxMessagesPerPoll = 1) private static class ChannelAdapterAnnotationTestBean { - @Pollable + @ChannelAdapter("testChannel") + @Poller(period = 1000, initialDelay = 0, maxMessagesPerPoll = 1) public String test() { return "test"; } @@ -538,25 +442,12 @@ public class MessagingAnnotationPostProcessorTests { @MessageEndpoint - private static class HandlerWithTransformers { + private static class TransformerAnnotationTestBean { @Transformer - @Order(-1) public String transformBefore(String input) { - return "pre." + input; - } - - @Handler - @Order(0) - public String handle(String input) { return input.toUpperCase(); } - - @Transformer - @Order(1) - public String transformAfter(String input) { - return input + ".post"; - } } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java new file mode 100644 index 0000000000..71c137fa2f --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java @@ -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 static org.junit.Assert.assertEquals; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Router; +import org.springframework.integration.bus.DefaultMessageBus; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class RouterAnnotationPostProcessorTests { + + private MessageBus messageBus; + + private DirectChannel inputChannel; + + private QueueChannel outputChannel; + + + @Before + public void init() { + inputChannel = new DirectChannel(); + outputChannel = new QueueChannel(); + inputChannel.setBeanName("input"); + outputChannel.setBeanName("output"); + messageBus = new DefaultMessageBus(); + messageBus.registerChannel(inputChannel); + messageBus.registerChannel(outputChannel); + } + + + @Test + public void testRouter() { + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + messageBus.start(); + TestRouter testRouter = new TestRouter(); + postProcessor.postProcessAfterInitialization(testRouter, "test"); + inputChannel.send(new StringMessage("foo")); + Message replyMessage = outputChannel.receive(0); + assertEquals("foo", replyMessage.getPayload()); + } + + + @MessageEndpoint + public static class TestRouter { + + @Router(inputChannel="input", defaultOutputChannel="output") + public String test(String s) { + return null; + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java index 8a74a4f78b..d73f97996f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SimpleAnnotatedEndpoint.java @@ -23,10 +23,10 @@ import org.springframework.integration.endpoint.annotation.ITestEndpoint; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel", output="outputChannel") +@MessageEndpoint public class SimpleAnnotatedEndpoint implements ITestEndpoint { - @Handler + @Handler(inputChannel="inputChannel", outputChannel="outputChannel") public String sayHello(String name) { return "hello " + name; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java new file mode 100644 index 0000000000..71fb21c248 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java @@ -0,0 +1,93 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.junit.Before; +import org.junit.Test; + +import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Splitter; +import org.springframework.integration.bus.DefaultMessageBus; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class SplitterAnnotationPostProcessorTests { + + private MessageBus messageBus; + + private DirectChannel inputChannel; + + private QueueChannel outputChannel; + + + @Before + public void init() { + inputChannel = new DirectChannel(); + outputChannel = new QueueChannel(); + inputChannel.setBeanName("input"); + outputChannel.setBeanName("output"); + messageBus = new DefaultMessageBus(); + messageBus.registerChannel(inputChannel); + messageBus.registerChannel(outputChannel); + } + + + @Test + public void testSplitterAnnotation() throws InterruptedException { + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + postProcessor.afterPropertiesSet(); + TestSplitter splitter = new TestSplitter(); + postProcessor.postProcessAfterInitialization(splitter, "testSplitter"); + messageBus.start(); + inputChannel.send(new StringMessage("this.is.a.test")); + Message message1 = outputChannel.receive(500); + assertNotNull(message1); + assertEquals("this", message1.getPayload()); + Message message2 = outputChannel.receive(500); + assertNotNull(message2); + assertEquals("is", message2.getPayload()); + Message message3 = outputChannel.receive(500); + assertNotNull(message3); + assertEquals("a", message3.getPayload()); + Message message4 = outputChannel.receive(500); + assertNotNull(message4); + assertEquals("test", message4.getPayload()); + assertNull(outputChannel.receive(0)); + messageBus.stop(); + } + + + @MessageEndpoint + public static class TestSplitter { + + @Splitter(inputChannel="input", outputChannel="output") + public String[] split(String s) { + return s.split("\\."); + } + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.java index bdd20109c7..4a48aa999e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCompletionStrategy.java @@ -28,18 +28,16 @@ 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.stereotype.Component; /** * @author Marius Bogoevici */ -@MessageEndpoint(input="inputChannel") -@Component("endpointWithDefaultAnnotationAndCustomCompletionStrategy") +@MessageEndpoint("endpointWithDefaultAnnotationAndCustomCompletionStrategy") public class TestAnnotatedEndpointWithCompletionStrategy { private final ConcurrentMap> aggregatedMessages = new ConcurrentHashMap>(); - @Aggregator + @Aggregator(inputChannel = "inputChannel") public Message aggregatingMethod(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java index ba564bd822..e7ffb3e936 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithCustomizedAggregator.java @@ -24,7 +24,6 @@ import java.util.concurrent.ConcurrentMap; import org.springframework.integration.aggregator.MessageSequenceComparator; import org.springframework.integration.annotation.Aggregator; -import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.stereotype.Component; @@ -32,13 +31,15 @@ import org.springframework.stereotype.Component; /** * @author Marius Bogoevici */ -@MessageEndpoint(input = "inputChannel", output = "outputChannel") @Component("endpointWithCustomizedAnnotation") public class TestAnnotatedEndpointWithCustomizedAggregator { private final ConcurrentMap> aggregatedMessages = new ConcurrentHashMap>(); - @Aggregator(discardChannel = "discardChannel", + @Aggregator( + inputChannel = "inputChannel", + outputChannel = "outputChannel", + discardChannel = "discardChannel", reaperInterval = 1234, sendPartialResultsOnTimeout = true, sendTimeout = 98765432, timeout = 4567890, trackedCorrelationIdCapacity = 42) public Message aggregatingMethod(List> messages) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java index ce0c184591..2e329019cd 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TestAnnotatedEndpointWithDefaultAggregator.java @@ -27,18 +27,16 @@ import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; -import org.springframework.stereotype.Component; /** * @author Marius Bogoevici */ -@MessageEndpoint(input="inputChannel") -@Component("endpointWithDefaultAnnotation") +@MessageEndpoint("endpointWithDefaultAnnotation") public class TestAnnotatedEndpointWithDefaultAggregator { private final ConcurrentMap> aggregatedMessages = new ConcurrentHashMap>(); - @Aggregator + @Aggregator(inputChannel="inputChannel") public Message aggregatingMethod(List> messages) { List> sortableList = new ArrayList>(messages); Collections.sort(sortableList, new MessageSequenceComparator()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java index c07f722bd2..5ed0915b07 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/TypeConvertingTestEndpoint.java @@ -22,10 +22,10 @@ import org.springframework.integration.annotation.MessageEndpoint; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel", output="outputChannel") +@MessageEndpoint public class TypeConvertingTestEndpoint { - @Handler + @Handler(inputChannel="inputChannel", outputChannel="outputChannel") public int multiplyByTwo(int number) { return number * 2; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml deleted file mode 100644 index 988f7595b1..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/testInvalidCompletionStrategyAnnotation.xml +++ /dev/null @@ -1,27 +0,0 @@ - - - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java index 56173982d9..ee09fc6356 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/MessageParameterAnnotatedEndpoint.java @@ -24,10 +24,10 @@ import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -@MessageEndpoint(input="inputChannel", output="outputChannel") +@MessageEndpoint public class MessageParameterAnnotatedEndpoint { - @Handler + @Handler(inputChannel="inputChannel", outputChannel="outputChannel") public StringMessage sayHello(Message message) { return new StringMessage("hello " + message.getPayload()); }