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

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.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-02 16:28:17 +00:00
parent e9050a8f82
commit 46b37df03d
3 changed files with 48 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

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

6
org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBusAwareBeanPostProcessor.java
View File

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

47
org.springframework.integration/src/main/java/org/springframework/integration/router/config/AggregatorMessageHandlerCreator.java
View File

@@ -64,26 +64,31 @@ public class AggregatorMessageHandlerCreator extends AbstractMessageHandlerCreat
public MessageHandler doCreateHandler(Object object, Method method, Map<String, ?> 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> T getAttribute(Map<String, ?> attributes, String name, Class<T> 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;
}
}