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

AbstractEndpointParser creates a DirectChannel for an 'input-channel' that is not explicitly configured within the context.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-06 20:11:49 +00:00
parent 97306c4b00
commit 709f6d72df
3 changed files with 31 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java
View File

@@ -18,10 +18,12 @@ package org.springframework.integration.config;
import org.w3c.dom.Element; import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext; import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.config.ConsumerEndpointFactoryBean; import org.springframework.integration.endpoint.config.ConsumerEndpointFactoryBean;
import org.springframework.util.Assert; import org.springframework.util.Assert;
import org.springframework.util.StringUtils; import org.springframework.util.StringUtils;
@@ -97,6 +99,11 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
String inputChannelAttributeName = this.getInputChannelAttributeName(); String inputChannelAttributeName = this.getInputChannelAttributeName();
String inputChannelName = element.getAttribute(inputChannelAttributeName); String inputChannelName = element.getAttribute(inputChannelAttributeName);
Assert.hasText(inputChannelName, "the '" + inputChannelAttributeName + "' attribute is required"); Assert.hasText(inputChannelName, "the '" + inputChannelAttributeName + "' attribute is required");
if (!parserContext.getRegistry().containsBeanDefinition(inputChannelName)) {
BeanDefinitionBuilder channelDef = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class);
BeanDefinitionHolder holder = new BeanDefinitionHolder(channelDef.getBeanDefinition(), inputChannelName);
BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry());
}
builder.addPropertyValue("inputChannelName", inputChannelName); builder.addPropertyValue("inputChannelName", inputChannelName);
Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT); Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
if (pollerElement != null) { if (pollerElement != null) {

2
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageHandlingEndpoint.java
View File

@@ -53,7 +53,7 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon
this.outputChannel = outputChannel; this.outputChannel = outputChannel;
} }
public MessageChannel getOutputChannel() { protected MessageChannel getOutputChannel() {
return this.outputChannel; return this.outputChannel;
} }

53
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java
View File

@@ -24,7 +24,6 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.core.task.TaskExecutor; import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel; import org.springframework.integration.channel.SubscribableChannel;
@@ -141,37 +140,31 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, ChannelRegistry
if (this.initialized) { if (this.initialized) {
return; return;
} }
if (!this.beanFactory.containsBean(this.inputChannelName)) { Assert.isTrue(this.beanFactory.containsBean(this.inputChannelName),
DirectChannel channel = new DirectChannel(); "no such input channel '" + this.inputChannelName + "'");
channel.setBeanName(this.inputChannelName); MessageChannel channel = (MessageChannel)
this.beanFactory.registerSingleton(this.inputChannelName, channel); this.beanFactory.getBean(this.inputChannelName, MessageChannel.class);
this.endpoint = new SubscribingConsumerEndpoint(this.consumer, channel); if (channel instanceof SubscribableChannel) {
this.endpoint = new SubscribingConsumerEndpoint(
this.consumer, (SubscribableChannel) channel);
}
else if (channel instanceof PollableChannel) {
if (this.trigger == null) {
this.trigger = new IntervalTrigger(0);
}
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
this.consumer, (PollableChannel) channel);
pollingEndpoint.setTrigger(this.trigger);
pollingEndpoint.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
pollingEndpoint.setReceiveTimeout(this.receiveTimeout);
pollingEndpoint.setTaskExecutor(this.taskExecutor);
pollingEndpoint.setTransactionManager(this.transactionManager);
pollingEndpoint.setTransactionDefinition(this.transactionDefinition);
this.endpoint = pollingEndpoint;
} }
else { else {
MessageChannel channel = (MessageChannel) throw new IllegalArgumentException(
this.beanFactory.getBean(this.inputChannelName, MessageChannel.class); "unsupported channel type: [" + channel.getClass() + "]");
if (channel instanceof SubscribableChannel) {
this.endpoint = new SubscribingConsumerEndpoint(
this.consumer, (SubscribableChannel) channel);
}
else if (channel instanceof PollableChannel) {
if (this.trigger == null) {
this.trigger = new IntervalTrigger(0);
}
PollingConsumerEndpoint pollingEndpoint = new PollingConsumerEndpoint(
this.consumer, (PollableChannel) channel);
pollingEndpoint.setTrigger(this.trigger);
pollingEndpoint.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
pollingEndpoint.setReceiveTimeout(this.receiveTimeout);
pollingEndpoint.setTaskExecutor(this.taskExecutor);
pollingEndpoint.setTransactionManager(this.transactionManager);
pollingEndpoint.setTransactionDefinition(this.transactionDefinition);
this.endpoint = pollingEndpoint;
}
else {
throw new IllegalArgumentException(
"unsupported channel type: [" + channel.getClass() + "]");
}
} }
this.initialized = true; this.initialized = true;
} }