From 46b37df03dbb2db2a4b8f6e67b951bcff67c1dae Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sat, 2 Aug 2008 16:28:17 +0000 Subject: [PATCH] The MessageBus no longer calls registerChannels() from setApplicationContext(). Instead it maintains a reference to the applicationContext so that it can lookup channels on demand during startup. Then, the registerChannels() method is invoked after the ApplicationContext fires the ContextRefreshedEvent to register all channels that have not yet been registered. The AggregatorMessageHandlerCreator now performs type-safe retrieval from its attribute map. --- .../integration/bus/DefaultMessageBus.java | 14 +++++- .../bus/MessageBusAwareBeanPostProcessor.java | 6 ++- .../AggregatorMessageHandlerCreator.java | 47 +++++++++++++------ 3 files changed, 48 insertions(+), 19 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index ec9d4321e4..bbe5985ae4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java @@ -94,6 +94,8 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private volatile TaskScheduler taskScheduler; + private volatile ApplicationContext applicationContext; + private volatile boolean configureAsyncEventMulticaster = false; private volatile boolean autoCreateChannels = false; @@ -127,7 +129,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A throw new ConfigurationException("Only one instance of '" + this.getClass().getSimpleName() + "' is allowed per ApplicationContext."); } - this.registerChannels(applicationContext); + this.applicationContext = applicationContext; } /** @@ -229,7 +231,15 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A } public MessageChannel lookupChannel(String channelName) { - return this.channelRegistry.lookupChannel(channelName); + MessageChannel channel = this.channelRegistry.lookupChannel(channelName); + if (channel == null && this.applicationContext != null && this.applicationContext.containsBean(channelName)) { + Object bean = this.applicationContext.getBean(channelName); + if (bean instanceof MessageChannel) { + channel = (MessageChannel) bean; + this.registerChannel(channelName, channel); + } + } + return channel; } public void registerChannel(String name, MessageChannel channel) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java index 4f47b5f8fc..6ab86f925d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java @@ -31,12 +31,14 @@ import org.springframework.util.Assert; public class MessageBusAwareBeanPostProcessor implements BeanPostProcessor { private final MessageBus messageBus; - + + public MessageBusAwareBeanPostProcessor(MessageBus messageBus) { Assert.notNull(messageBus, "'messageBus' must not be null"); this.messageBus = messageBus; } - + + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { return postProcessIfNecessary(bean); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java index e3a8215004..8556448b60 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java @@ -64,26 +64,31 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat public MessageHandler doCreateHandler(Object object, Method method, Map attributes) { AggregatingMessageHandler messageHandler = new AggregatingMessageHandler(new AggregatorAdapter(object, method)); this.configureDefaultReplyChannel(messageHandler, object); - if (attributes.containsKey(DISCARD_CHANNEL)) { - messageHandler.setDiscardChannel(this.channelRegistry.lookupChannel( - (String) attributes.get(DISCARD_CHANNEL))); + String discardChannelName = this.getAttribute(attributes, DISCARD_CHANNEL, String.class); + if (discardChannelName != null) { + messageHandler.setDiscardChannel(this.channelRegistry.lookupChannel(discardChannelName)); } - if (attributes.containsKey(SEND_TIMEOUT)) { - messageHandler.setSendTimeout((Long) attributes.get(SEND_TIMEOUT)); + Long sendTimeout = this.getAttribute(attributes, SEND_TIMEOUT, Long.class); + if (sendTimeout != null) { + messageHandler.setSendTimeout(sendTimeout); } - if (attributes.containsKey(SEND_PARTIAL_RESULTS_ON_TIMEOUT)) { - messageHandler.setSendPartialResultOnTimeout( - (Boolean) attributes.get(SEND_PARTIAL_RESULTS_ON_TIMEOUT)); + Boolean sendPartialResultOnTimeout = this.getAttribute( + attributes, SEND_PARTIAL_RESULTS_ON_TIMEOUT, Boolean.class); + if (sendPartialResultOnTimeout != null) { + messageHandler.setSendPartialResultOnTimeout(sendPartialResultOnTimeout); } - if (attributes.containsKey(REAPER_INTERVAL)) { - messageHandler.setReaperInterval((Long) attributes.get(REAPER_INTERVAL)); + Long reaperInterval = this.getAttribute(attributes, REAPER_INTERVAL, Long.class); + if (reaperInterval != null) { + messageHandler.setReaperInterval(reaperInterval); } - if (attributes.containsKey(TIMEOUT)) { - messageHandler.setTimeout((Long) attributes.get(TIMEOUT)); + Long timeout = this.getAttribute(attributes, TIMEOUT, Long.class); + if (timeout != null) { + messageHandler.setTimeout(timeout); } - if (attributes.containsKey(TRACKED_CORRELATION_ID_CAPACITY)) { - messageHandler.setTrackedCorrelationIdCapacity( - (Integer) attributes.get(TRACKED_CORRELATION_ID_CAPACITY)); + Integer trackedCorrelationIdCapacity = this.getAttribute( + attributes, TRACKED_CORRELATION_ID_CAPACITY, Integer.class); + if (trackedCorrelationIdCapacity != null) { + messageHandler.setTrackedCorrelationIdCapacity(trackedCorrelationIdCapacity); } this.configureCompletionStrategy(object, messageHandler); return messageHandler; @@ -111,4 +116,16 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat }); } + @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; + } + }