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

MessageEndpointAnnotationPostProcessor now requires 'messageBus' as a constructor-arg (rather than setter).

Browse Source
This commit is contained in:
Mark Fisher
2008-03-04 17:25:58 +00:00
parent f987d920aa
commit 41e6befa23
5 changed files with 30 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java
View File

@@ -70,7 +70,7 @@ public class AnnotationDrivenParser implements BeanDefinitionParser {
private void createMessageEndpointPostProcessor(ParserContext parserContext) {
BeanDefinition bd = new RootBeanDefinition(MessageEndpointAnnotationPostProcessor.class);
bd.getPropertyValues().addPropertyValue("messageBus",
bd.getConstructorArgumentValues().addGenericArgumentValue(
new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME));
BeanComponentDefinition bcd = new BeanComponentDefinition(
bd, MESSAGE_ENDPOINT_ANNOTATION_POST_PROCESSOR_BEAN_NAME);

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

@@ -42,7 +42,6 @@ import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.SimpleChannel;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.endpoint.DefaultMessageEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.MessageHandlerChain;
@@ -67,19 +66,20 @@ import org.springframework.util.StringUtils;
*/
public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor, InitializingBean {
private Log logger = LogFactory.getLog(this.getClass());
private final Log logger = LogFactory.getLog(this.getClass());
private Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators =
private final Map<Class<? extends Annotation>, MessageHandlerCreator> handlerCreators =
new ConcurrentHashMap<Class<? extends Annotation>, MessageHandlerCreator>();
private MessageBus messageBus;
private final MessageBus messageBus;
public void setMessageBus(MessageBus messageBus) {
Assert.notNull(messageBus, "messageBus must not be null");
public MessageEndpointAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "'messageBus' must not be null");
this.messageBus = messageBus;
}
public void setCustomHandlerCreators(
Map<Class<? extends Annotation>, MessageHandlerCreator> customHandlerCreators) {
for (Map.Entry<Class<? extends Annotation>, MessageHandlerCreator> entry : customHandlerCreators.entrySet()) {
@@ -88,7 +88,6 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
}
public void afterPropertiesSet() {
Assert.notNull(this.messageBus, "messageBus is required");
this.handlerCreators.put(Handler.class, new DefaultMessageHandlerCreator());
this.handlerCreators.put(Router.class, new RouterMessageHandlerCreator());
this.handlerCreators.put(Splitter.class, new SplitterMessageHandlerCreator());
@@ -180,7 +179,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
target.setObject(bean);
target.setMethod(method.getName());
target.afterPropertiesSet();
DefaultTargetAdapter adapter = new DefaultTargetAdapter(target);
DefaultTargetAdapter<Object> adapter = new DefaultTargetAdapter<Object>(target);
MessageHandler handler = endpoint.getHandler();
if (handler == null) {
endpoint.setHandler(adapter);

20
spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.router.config;
import java.lang.reflect.Method;
import java.util.Map;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.config.AbstractMessageHandlerCreator;
import org.springframework.integration.router.AggregatingMessageHandler;
@@ -47,27 +47,30 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat
private static final String TRACKED_CORRELATION_ID_CAPACITY = "trackedCorrelationIdCapacity";
private final MessageBus messageBus;
private final ChannelRegistry channelRegistry;
public AggregatorMessageHandlerCreator(MessageBus messageBus) {
this.messageBus = messageBus;
public AggregatorMessageHandlerCreator(ChannelRegistry channelRegistry) {
this.channelRegistry = channelRegistry;
}
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> attributes) {
AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(object, method));
if (attributes.containsKey(DEFAULT_REPLY_CHANNEL)) {
messageHandler.setDefaultReplyChannel(this.messageBus.lookupChannel((String) attributes.get(DEFAULT_REPLY_CHANNEL)));
messageHandler.setDefaultReplyChannel(this.channelRegistry.lookupChannel(
(String) attributes.get(DEFAULT_REPLY_CHANNEL)));
}
if (attributes.containsKey(DISCARD_CHANNEL)) {
messageHandler.setDiscardChannel(this.messageBus.lookupChannel((String) attributes.get(DISCARD_CHANNEL)));
messageHandler.setDiscardChannel(this.channelRegistry.lookupChannel(
(String) attributes.get(DISCARD_CHANNEL)));
}
if (attributes.containsKey(SEND_TIMEOUT)) {
messageHandler.setSendTimeout((Long) attributes.get(SEND_TIMEOUT));
}
if (attributes.containsKey(SEND_PARTIAL_RESULTS_ON_TIMEOUT)) {
messageHandler.setSendPartialResultOnTimeout((Boolean) attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT));
messageHandler.setSendPartialResultOnTimeout(
(Boolean) attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT));
}
if (attributes.containsKey(REAPER_INTERVAL)) {
messageHandler.setReaperInterval((Long) attributes.get(REAPER_INTERVAL));
@@ -76,7 +79,8 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat
messageHandler.setTimeout((Long) attributes.get(TIMEOUT));
}
if (attributes.containsKey(TRACKED_CORRELATION_ID_CAPACITY)) {
messageHandler.setTrackedCorrelationIdCapacity((Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY));
messageHandler.setTrackedCorrelationIdCapacity(
(Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY));
}
return messageHandler;
}

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

@@ -17,7 +17,6 @@
package org.springframework.integration.endpoint.annotation;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -48,7 +47,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
inputChannel.send(new StringMessage("world"));
Message<String> message = outputChannel.receive(1000);
Message<?> message = outputChannel.receive(1000);
assertEquals("hello world", message.getPayload());
context.stop();
}
@@ -60,7 +59,7 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel");
MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel");
inputChannel.send(new StringMessage("world"));
Message<String> message = outputChannel.receive(1000);
Message<?> message = outputChannel.receive(1000);
assertEquals("hello world", message.getPayload());
context.stop();
}
@@ -70,8 +69,8 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageBus messageBus = new MessageBus();
SimpleChannel testChannel = new SimpleChannel();
messageBus.registerChannel("testChannel", testChannel);
MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor();
postProcessor.setMessageBus(messageBus);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
PolledAnnotationTestBean testBean = new PolledAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
messageBus.start();
@@ -85,8 +84,8 @@ public class MessageEndpointAnnotationPostProcessorTests {
MessageBus messageBus = new MessageBus();
SimpleChannel testChannel = new SimpleChannel();
messageBus.registerChannel("testChannel", testChannel);
MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor();
postProcessor.setMessageBus(messageBus);
MessageEndpointAnnotationPostProcessor postProcessor =
new MessageEndpointAnnotationPostProcessor(messageBus);
CountDownLatch latch = new CountDownLatch(1);
DefaultOutputAnnotationTestBean testBean = new DefaultOutputAnnotationTestBean(latch);
postProcessor.postProcessAfterInitialization(testBean, "testBean");
@@ -98,12 +97,9 @@ public class MessageEndpointAnnotationPostProcessorTests {
messageBus.stop();
}
@Test
public void testPostProcessorWithNoMessageBus() {
MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor();
PolledAnnotationTestBean testBean = new PolledAnnotationTestBean();
Object result = postProcessor.postProcessAfterInitialization(testBean, "testBean");
assertSame(testBean, result);
@Test(expected=IllegalArgumentException.class)
public void testPostProcessorWithNullMessageBus() {
new MessageEndpointAnnotationPostProcessor(null);
}

2
spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml
View File

@@ -16,7 +16,7 @@
<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.SimpleAnnotatedEndpoint"/>
<bean class="org.springframework.integration.config.MessageEndpointAnnotationPostProcessor">
<property name="messageBus" ref="bus"/>
<constructor-arg ref="bus"/>
</bean>
</beans>