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

Removed the @MessageSource method-level annotation and added @Pollable as its replacement. Also added the @ChannelAdapter class-level annotation (refactoring to remove SourceEndpoint).

Browse Source
This commit is contained in:
Mark Fisher
2008-08-05 20:02:52 +00:00
parent bbd7a12a4e
commit 8274dfc428
10 changed files with 105 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
org.springframework.integration/src/main/java/org/springframework/integration/annotation/ChannelAdapter.java Normal file
View File

@@ -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();
}

4
org.springframework.integration/src/main/java/org/springframework/integration/annotation/MessageSource.java → org.springframework.integration/src/main/java/org/springframework/integration/annotation/Pollable.java
View File

@@ -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 {
}

2
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Polled.java
View File

@@ -47,4 +47,6 @@ public @interface Polled {
TimeUnit timeUnit() default TimeUnit.MILLISECONDS;
int maxMessagesPerPoll() default 1;
}

30
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -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);
}
}

47
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SourceAnnotationPostProcessor.java → org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java
View File

@@ -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<MessageSource<?>> {
public class PollableAnnotationPostProcessor extends AbstractAnnotationMethodPostProcessor<MessageSource<?>> {
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<MessageSource<?>> 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;
}
}

17
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -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";
}

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

@@ -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;
}
}