From 08e6db266b9a23fb0ab6f05d965fc8d05fa4f27c Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sun, 5 Oct 2008 22:18:54 +0000 Subject: [PATCH] Added ConsumerEndpointFactoryBean. --- .../config/ConsumerEndpointFactoryBean.java | 168 ++++++++++++++++++ 1 file changed, 168 insertions(+) create mode 100644 org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java 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 new file mode 100644 index 0000000000..6c25d1b9a6 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/config/ConsumerEndpointFactoryBean.java @@ -0,0 +1,168 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.endpoint.config; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.channel.SubscribableChannel; +import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.PollingConsumerEndpoint; +import org.springframework.integration.endpoint.SubscribingConsumerEndpoint; +import org.springframework.integration.message.MessageConsumer; +import org.springframework.integration.scheduling.IntervalTrigger; +import org.springframework.integration.scheduling.Trigger; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + */ +public class ConsumerEndpointFactoryBean implements FactoryBean, BeanFactoryAware, InitializingBean { + + private final MessageConsumer consumer; + + private volatile String inputChannelName; + + private volatile Trigger trigger; + + private volatile int maxMessagesPerPoll; + + private volatile long receiveTimeout = 1000; + + private volatile TaskExecutor taskExecutor; + + private volatile PlatformTransactionManager transactionManager; + + private volatile TransactionDefinition transactionDefinition; + + private volatile ConfigurableBeanFactory beanFactory; + + private volatile MessageEndpoint endpoint; + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + public ConsumerEndpointFactoryBean(MessageConsumer consumer) { + Assert.notNull(consumer, "consumer must not be null"); + this.consumer = consumer; + } + + + public void setInputChannelName(String inputChannelName) { + this.inputChannelName = inputChannelName; + } + + public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { + this.maxMessagesPerPoll = maxMessagesPerPoll; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setTransactionDefinition(TransactionDefinition transactionDefinition) { + this.transactionDefinition = transactionDefinition; + } + + public void setBeanFactory(BeanFactory beanFactory) { + Assert.isInstanceOf(ConfigurableBeanFactory.class, beanFactory, + "a ConfigurableBeanFactory is required"); + this.beanFactory = (ConfigurableBeanFactory) beanFactory; + } + + public void afterPropertiesSet() { + Assert.hasText(this.inputChannelName, "inputChannelName is required"); + } + + public Object getObject() throws Exception { + if (!this.initialized) { + this.initializeEndpoint(); + } + return this.endpoint; + } + + public Class getObjectType() { + if (this.endpoint == null) { + return MessageEndpoint.class; + } + return endpoint.getClass(); + } + + public boolean isSingleton() { + return true; + } + + private void initializeEndpoint() { + synchronized (this.initializationMonitor) { + 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); + } + 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() + "]"); + } + } + this.initialized = true; + } + } + +}