From 41e6befa239c27fa9d5879502df05dc4c57f8d74 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 4 Mar 2008 17:25:58 +0000 Subject: [PATCH] MessageEndpointAnnotationPostProcessor now requires 'messageBus' as a constructor-arg (rather than setter). --- .../config/AnnotationDrivenParser.java | 2 +- ...essageEndpointAnnotationPostProcessor.java | 15 ++++++------- .../AggregatorMessageHandlerCreator.java | 20 ++++++++++------- ...eEndpointAnnotationPostProcessorTests.java | 22 ++++++++----------- .../simpleAnnotatedEndpointTests.xml | 2 +- 5 files changed, 30 insertions(+), 31 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java index b6fcc4770f..c8cffbe8ad 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AnnotationDrivenParser.java @@ -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); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 93983eaefc..fd2ddbead5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -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, MessageHandlerCreator> handlerCreators = + private final Map, MessageHandlerCreator> handlerCreators = new ConcurrentHashMap, 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, MessageHandlerCreator> customHandlerCreators) { for (Map.Entry, 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 adapter = new DefaultTargetAdapter(target); MessageHandler handler = endpoint.getHandler(); if (handler == null) { endpoint.setHandler(adapter); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java b/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java index 278e40cd08..0d831c389e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java @@ -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 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; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index 614f066373..64048d3072 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java @@ -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 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 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); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml index c0e883cf6a..2bb8a3c774 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointTests.xml @@ -16,7 +16,7 @@ - +