From 709f6d72df1ff688ccecd12db2b9db10fdcca949 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 6 Oct 2008 20:11:49 +0000 Subject: [PATCH] AbstractEndpointParser creates a DirectChannel for an 'input-channel' that is not explicitly configured within the context. --- .../config/AbstractEndpointParser.java | 7 +++ .../AbstractMessageHandlingEndpoint.java | 2 +- .../config/ConsumerEndpointFactoryBean.java | 53 ++++++++----------- 3 files changed, 31 insertions(+), 31 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java index 05acc20710..a7200f2668 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java @@ -18,10 +18,12 @@ package org.springframework.integration.config; 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.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.endpoint.config.ConsumerEndpointFactoryBean; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -97,6 +99,11 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio String inputChannelAttributeName = this.getInputChannelAttributeName(); String inputChannelName = element.getAttribute(inputChannelAttributeName); 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); Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT); if (pollerElement != null) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageHandlingEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageHandlingEndpoint.java index 654c8f9071..eee2a2247b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageHandlingEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageHandlingEndpoint.java @@ -53,7 +53,7 @@ public abstract class AbstractMessageHandlingEndpoint extends AbstractMessageCon this.outputChannel = outputChannel; } - public MessageChannel getOutputChannel() { + protected MessageChannel getOutputChannel() { return this.outputChannel; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java index 52f0ece4a1..c9fe157da3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java @@ -24,7 +24,6 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; @@ -141,37 +140,31 @@ public class ConsumerEndpointFactoryBean implements FactoryBean, ChannelRegistry if (this.initialized) { return; } - if (!this.beanFactory.containsBean(this.inputChannelName)) { - DirectChannel channel = new DirectChannel(); - channel.setBeanName(this.inputChannelName); - this.beanFactory.registerSingleton(this.inputChannelName, channel); - this.endpoint = new SubscribingConsumerEndpoint(this.consumer, channel); + Assert.isTrue(this.beanFactory.containsBean(this.inputChannelName), + "no such input channel '" + this.inputChannelName + "'"); + MessageChannel channel = (MessageChannel) + this.beanFactory.getBean(this.inputChannelName, MessageChannel.class); + 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 { - MessageChannel channel = (MessageChannel) - this.beanFactory.getBean(this.inputChannelName, MessageChannel.class); - 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() + "]"); - } + throw new IllegalArgumentException( + "unsupported channel type: [" + channel.getClass() + "]"); } this.initialized = true; }