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..ecc4fa5760 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 @@ -22,11 +22,13 @@ import java.util.Random; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.annotation.Polled; /** * @author Mark Fisher */ @MessageEndpoint(input="tickers", output="quotes") +@Polled(period=300) public class QuoteService { @Handler 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 b6cfa54b49..756a84ab8f 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 @@ -18,18 +18,16 @@ package org.springframework.integration.samples.quote; import java.util.Random; -import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.MessageSource; -import org.springframework.integration.annotation.Polled; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.Pollable; /** * @author Mark Fisher */ -@MessageEndpoint(output="tickers") -@Polled(period=300) +@ChannelAdapter("tickers") public class TickerStream { - @MessageSource + @Pollable public String nextTicker() { char[] chars = new char[3]; for (int i = 0; i < 3; i++) { diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml index 1abd1ad39f..9e11e43b68 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/quoteDemo.xml @@ -11,8 +11,6 @@ - - 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 new file mode 100644 index 0000000000..5ec0a077bb --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java @@ -0,0 +1,42 @@ +/* + * 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 java.lang.annotation.Target; + +import org.springframework.stereotype.Component; + +/** + * Indicates that a class is capable of serving as a message channel. + * + * @author Mark Fisher + */ +@Target(ElementType.TYPE) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +@Component +public @interface ChannelAdapter { + + String value(); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java similarity index 93% rename from org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java rename to org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java index 0ed341fe4c..fb6daae146 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java @@ -28,7 +28,7 @@ import org.springframework.integration.message.Message; * Indicates that a method is capable of producing messages. The method must * accept no parameters and return either a {@link Message} or an Object to * be passed as the message payload. The enclosing class may also be annotated - * with {@link MessageEndpoint @MessageEndpoint}. + * with {@link ChannelAdapter @ChannelAdapter}. * * @author Mark Fisher */ @@ -36,6 +36,6 @@ import org.springframework.integration.message.Message; @Retention(RetentionPolicy.RUNTIME) @Inherited @Documented -public @interface MessageSource { +public @interface Pollable { } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java index 520c5af5f3..bf142c665d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java @@ -47,4 +47,6 @@ public @interface Polled { TimeUnit timeUnit() default TimeUnit.MILLISECONDS; + int maxMessagesPerPoll() default 1; + } 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 ffceb3d099..2be8d23d12 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 @@ -27,12 +27,17 @@ 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.Polled; 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.util.Assert; /** @@ -66,7 +71,7 @@ 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 SourceAnnotationPostProcessor(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)); } @@ -86,6 +91,29 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Init postProcessor.createEndpoint(bean, beanName, beanClass, endpointAnnotation); if (endpoint != null) { endpoint.setName(beanName + "." + entry.getKey().getSimpleName() + ".endpoint"); + Polled polledAnnotation = AnnotationUtils.findAnnotation(beanClass, Polled.class); + if (polledAnnotation != null) { + PollingSchedule schedule = new PollingSchedule(polledAnnotation.period()); + schedule.setInitialDelay(polledAnnotation.initialDelay()); + schedule.setFixedRate(polledAnnotation.fixedRate()); + schedule.setTimeUnit(polledAnnotation.timeUnit()); + String inputChannelName = endpointAnnotation.input(); + MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName); + if (inputChannel != null) { + if (inputChannel instanceof PollableChannel) { + PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule); + poller.setMaxMessagesPerPoll(polledAnnotation.maxMessagesPerPoll()); + endpoint.setSource(poller); + } + else { + endpoint.setSource(inputChannel); + } + } + else { + endpoint.setInputChannelName(inputChannelName); + } + endpoint.setSchedule(schedule); + } this.messageBus.registerEndpoint(endpoint); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java similarity index 51% rename from org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java rename to org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java index a9f5e695a9..460be58aca 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java @@ -22,26 +22,23 @@ import java.util.List; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.ConfigurationException; -import org.springframework.integration.annotation.Polled; +import org.springframework.integration.annotation.ChannelAdapter; +import org.springframework.integration.annotation.Pollable; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.dispatcher.DirectChannel; +import org.springframework.integration.channel.PollableChannelAdapter; import org.springframework.integration.endpoint.AbstractEndpoint; -import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.message.MethodInvokingSource; import org.springframework.integration.message.MessageSource; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.util.StringUtils; +import org.springframework.integration.message.MethodInvokingSource; /** - * Post-processor for classes annotated with {@link MessageSource @MessageSource}. + * Post-processor for methods annotated with {@link Pollable @Pollable}. * * @author Mark Fisher */ -public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor> { +public class PollableAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor> { - public SourceAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { - super(org.springframework.integration.annotation.MessageSource.class, messageBus, beanClassLoader); + public PollableAnnotationPostProcessor(MessageBus messageBus, ClassLoader beanClassLoader) { + super(Pollable.class, messageBus, beanClassLoader); } @@ -49,37 +46,25 @@ public class SourceAnnotationPostProcessor extends AbstractAnnotationMethodPostP MethodInvokingSource source = new MethodInvokingSource(); source.setObject(bean); source.setMethodName(method.getName()); + ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class); + if (channelAdapterAnnotation != null) { + String channelName = channelAdapterAnnotation.value(); + PollableChannelAdapter adapter = new PollableChannelAdapter(channelName, source, null); + this.getMessageBus().registerChannel(channelName, adapter); + } return source; } protected MessageSource processResults(List> results) { if (results.size() > 1) { - throw new ConfigurationException("At most one @MessageSource annotation is allowed per class."); + 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, String beanName, Class originalBeanClass, org.springframework.integration.annotation.MessageEndpoint endpointAnnotation) { - SourceEndpoint endpoint = new SourceEndpoint((MessageSource) bean); - Polled polledAnnotation = AnnotationUtils.findAnnotation(originalBeanClass, Polled.class); - int period = polledAnnotation.period(); - long initialDelay = polledAnnotation.initialDelay(); - boolean fixedRate = polledAnnotation.fixedRate(); - PollingSchedule schedule = new PollingSchedule(period); - schedule.setInitialDelay(initialDelay); - schedule.setFixedRate(fixedRate); - endpoint.setSchedule(schedule); - String outputChannelName = endpointAnnotation.output(); - if (!StringUtils.hasText(outputChannelName)) { - MessageChannel outputChannel = new DirectChannel(); - this.getMessageBus().registerChannel(beanName + ".output", outputChannel); - endpoint.setTarget(outputChannel); - } - else { - endpoint.setOutputChannelName(outputChannelName); - } - return endpoint; + return null; } } 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 84b6bc6efc..ceb7d5cace 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 @@ -39,11 +39,12 @@ 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.Concurrency; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.MessageSource; import org.springframework.integration.annotation.MessageTarget; +import org.springframework.integration.annotation.Pollable; import org.springframework.integration.annotation.Polled; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.annotation.Transformer; @@ -373,15 +374,14 @@ public class MessagingAnnotationPostProcessorTests { } @Test - public void testMessageSourceAnnotation() { + public void testChannelAdapterAnnotation() { MessageBus messageBus = new DefaultMessageBus(); - QueueChannel testChannel = new QueueChannel(); - messageBus.registerChannel("testChannel", testChannel); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); - MessageSourceAnnotationTestBean testBean = new MessageSourceAnnotationTestBean(); + ChannelAdapterAnnotationTestBean testBean = new ChannelAdapterAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); + PollableChannel testChannel = (PollableChannel) messageBus.lookupChannel("testChannel"); Message message = testChannel.receive(1000); assertEquals("test", message.getPayload()); messageBus.stop(); @@ -526,11 +526,10 @@ public class MessagingAnnotationPostProcessorTests { } - @MessageEndpoint(output="testChannel") - @Polled(period=100) - private static class MessageSourceAnnotationTestBean { + @ChannelAdapter("testChannel") + private static class ChannelAdapterAnnotationTestBean { - @MessageSource + @Pollable public String test() { return "test"; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java deleted file mode 100644 index 92999c499b..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/annotation/InboundChannelAdapterTestBean.java +++ /dev/null @@ -1,41 +0,0 @@ -/* - * Copyright 2002-2007 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.integration.endpoint.annotation; - -import org.springframework.integration.annotation.Handler; -import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.Polled; -import org.springframework.integration.annotation.MessageSource; - -/** - * @author Mark Fisher - */ -@MessageEndpoint(output="outputChannel") -@Polled(period=100) -public class InboundChannelAdapterTestBean { - - @MessageSource - public String getName() { - return "world"; - } - - @Handler - public String sayHello(String name) { - return "hello " + name; - } - -}