diff --git a/org.springframework.integration.adapter/.classpath b/org.springframework.integration.adapter/.classpath index 8c16497102..b3fa2c313c 100644 --- a/org.springframework.integration.adapter/.classpath +++ b/org.springframework.integration.adapter/.classpath @@ -26,6 +26,5 @@ - diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java index f83d22ca51..6b07cc68e9 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/event/MessagingEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/StubSession.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/StubSession.java index 55f8a16c09..dc6088e76e 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/StubSession.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/StubSession.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -42,7 +42,7 @@ import javax.jms.TopicSubscriber; */ public class StubSession implements Session { - private String messageText; + private final String messageText; public StubSession(String messageText) { diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml index 89be9fd48c..074ff76621 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/jms/config/jmsSourceEndpoint.xml @@ -9,7 +9,11 @@ - + + + + + diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml index 1d83a97169..ce8720ae82 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/filecopy/fileCopyDemo-common.xml @@ -11,13 +11,13 @@ - + + + - - + ref="exclaimer" method="exclaim"/> diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml index c2cb4efa16..afeb4f3569 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/oddeven/oddEvenDemo.xml @@ -13,11 +13,11 @@ - - - + - + + + diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java index 0264ce4a28..840eba411a 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/QuoteService.java @@ -22,13 +22,11 @@ import java.util.Random; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; -import org.springframework.integration.annotation.Poller; /** * @author Mark Fisher */ @MessageEndpoint(input="tickers", output="quotes") -@Poller(period=300) public class QuoteService { @Handler diff --git a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java index 756a84ab8f..72ff8da5de 100644 --- a/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java +++ b/org.springframework.integration.samples/src/main/java/org/springframework/integration/samples/quote/TickerStream.java @@ -20,11 +20,13 @@ import java.util.Random; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.Pollable; +import org.springframework.integration.annotation.Poller; /** * @author Mark Fisher */ @ChannelAdapter("tickers") +@Poller(period = 300) public class TickerStream { @Pollable diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java index 7a85c514f7..a68535a727 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java @@ -39,7 +39,7 @@ import org.springframework.integration.scheduling.PollingSchedule; @Documented public @interface Poller { - int period() default 0; + int period(); long initialDelay() default PollingSchedule.DEFAULT_INITIAL_DELAY; @@ -47,6 +47,6 @@ public @interface Poller { TimeUnit timeUnit() default TimeUnit.MILLISECONDS; - int maxMessagesPerPoll() default 1; + int maxMessagesPerPoll() default -1; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java deleted file mode 100644 index fe561e0b12..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/AbstractChannelAdapter.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * 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.channel; - -import org.springframework.integration.message.BlockingTarget; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; -import org.springframework.util.Assert; - -/** - * The base class for Channel Adapters. - * - * @author Mark Fisher - */ -public class AbstractChannelAdapter extends AbstractMessageChannel { - - private final MessageTarget target; - - - public AbstractChannelAdapter(String name, MessageTarget target) { - Assert.notNull(name, "name must not be null"); - this.setBeanName(name); - this.target = target; - } - - - @Override - protected boolean doSend(Message message, long timeout) { - if (this.target == null) { - return false; - } - return (timeout >= 0 && this.target instanceof BlockingTarget) - ? ((BlockingTarget) this.target).send(message, timeout) - : this.target.send(message); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java deleted file mode 100644 index aa892ea62c..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannelAdapter.java +++ /dev/null @@ -1,84 +0,0 @@ -/* - * 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.channel; - -import java.util.List; - -import org.springframework.integration.message.BlockingSource; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageDeliveryAware; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.message.selector.MessageSelector; - -/** - * Channel Adapter implementation for a {@link PollableSource}. - * - * @author Mark Fisher - */ -public class PollableChannelAdapter extends AbstractChannelAdapter implements PollableChannel, MessageDeliveryAware { - - private final PollableSource source; - - - public PollableChannelAdapter(String name, PollableSource source, MessageTarget target) { - super(name, target); - this.source = source; - } - - - public Message receive() { - return this.receive(-1); - } - - public Message receive(long timeout) { - if (this.source != null) { - return (timeout >= 0 && this.source instanceof BlockingSource) - ? ((BlockingSource) this.source).receive(timeout) - : this.source.receive(); - } - return null; - } - - public List> clear() { - if (this.source != null && this.source instanceof PollableChannel) { - return ((PollableChannel) this.source).clear(); - } - return null; - } - - public List> purge(MessageSelector selector) { - if (this.source != null && this.source instanceof PollableChannel) { - return ((PollableChannel) this.source).purge(selector); - } - return null; - } - - public void onSend(Message sentMessage) { - if (this.source != null && this.source instanceof MessageDeliveryAware) { - ((MessageDeliveryAware) this.source).onSend(sentMessage); - } - } - - public void onFailure(MessagingException exception) { - if (this.source != null && this.source instanceof MessageDeliveryAware) { - ((MessageDeliveryAware) this.source).onFailure(exception); - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java deleted file mode 100644 index 429a4c402b..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/SubscribableChannelAdapter.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.channel; - -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.SubscribableSource; -import org.springframework.util.Assert; - -/** - * Channel Adapter implementation for a {@link SubscribableSource}. - * - * @author Mark Fisher - */ -public class SubscribableChannelAdapter extends AbstractChannelAdapter implements SubscribableSource { - - private final SubscribableSource source; - - - public SubscribableChannelAdapter(String name, SubscribableSource source, MessageTarget target) { - super(name, target); - Assert.notNull(source, "source must not be null"); - this.source = source; - } - - - public boolean subscribe(MessageTarget target) { - return this.source.subscribe(target); - } - - public boolean unsubscribe(MessageTarget target) { - return this.source.unsubscribe(target); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java deleted file mode 100644 index c9aa56f493..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBean.java +++ /dev/null @@ -1,110 +0,0 @@ -/* - * 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.channel.config; - -import java.util.List; - -import org.springframework.beans.factory.BeanNameAware; -import org.springframework.beans.factory.FactoryBean; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.channel.ChannelInterceptor; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PollableChannelAdapter; -import org.springframework.integration.channel.SubscribableChannelAdapter; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.message.SubscribableSource; - -/** - * @author Mark Fisher - */ -public class ChannelAdapterFactoryBean implements FactoryBean, BeanNameAware { - - private volatile String name; - - private volatile MessageSource source; - - private volatile MessageTarget target; - - private volatile AbstractMessageChannel channel; - - private List interceptors; - - private volatile boolean initialized; - - private final Object initializationMonitor = new Object(); - - - public void setBeanName(String name) { - this.name = name; - } - - public void setSource(MessageSource source) { - this.source = source; - } - - public void setTarget(MessageTarget target) { - this.target = target; - } - - public void setInterceptors(List interceptors) { - this.interceptors = interceptors; - } - - public Object getObject() throws Exception { - if (!this.initialized) { - this.initializeChannel(); - } - return this.channel; - } - - public Class getObjectType() { - if (!this.initialized) { - return MessageChannel.class; - } - return this.channel.getClass(); - } - - public boolean isSingleton() { - return true; - } - - private void initializeChannel() { - synchronized (this.initializationMonitor) { - if (this.initialized) { - return; - } - if (this.source == null || source instanceof PollableSource) { - this.channel = new PollableChannelAdapter( - this.name, (PollableSource) this.source, this.target); - } - else if (this.source instanceof SubscribableSource) { - this.channel = new SubscribableChannelAdapter( - this.name, (SubscribableSource) this.source, this.target); - } - else { - throw new ConfigurationException("source must be either a PollableSource or SubscribableSource"); - } - if (this.interceptors != null) { - this.channel.setInterceptors(this.interceptors); - } - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index f7f53545e9..737aae1773 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -18,24 +18,98 @@ package org.springframework.integration.config; import org.w3c.dom.Element; +import org.springframework.beans.factory.BeanDefinitionStoreException; +import org.springframework.beans.factory.config.BeanDefinitionHolder; +import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.integration.channel.config.ChannelAdapterFactoryBean; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.endpoint.InboundChannelAdapter; +import org.springframework.integration.endpoint.OutboundChannelAdapter; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; /** - * Parser for the element. + * Parser for the <channel-adapter/> element. * * @author Mark Fisher */ -public class ChannelAdapterParser extends AbstractChannelParser { +public class ChannelAdapterParser extends AbstractBeanDefinitionParser { - protected final Class getBeanClass(Element element) { - return ChannelAdapterFactoryBean.class; + @Override + protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext) throws BeanDefinitionStoreException { + String id = element.getAttribute("id"); + if (!element.hasAttribute("channel")) { + // the created channel will get the 'id', so the adapter's bean name includes a suffix + id = id + ".adapter"; + } + else if (!StringUtils.hasText(id)) { + id = parserContext.getReaderContext().generateBeanName(definition); + } + return id; } @Override - protected void postProcess(BeanDefinitionBuilder builder, Element element) { - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "source"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "target"); + protected AbstractBeanDefinition parseInternal(Element element, ParserContext parserContext) { + String source = element.getAttribute("source"); + String target = element.getAttribute("target"); + String channelName = element.getAttribute("channel"); + Element pollerElement = DomUtils.getChildElementByTagName(element, "poller"); + BeanDefinitionBuilder adapterBuilder = null; + if (StringUtils.hasText(source)) { + if (StringUtils.hasText(target)) { + throw new ConfigurationException("both 'source' and 'target' are not allowed, provide only one"); + } + adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(InboundChannelAdapter.class); + if (pollerElement != null) { + String pollerBeanName = IntegrationNamespaceUtils.parsePoller(source, pollerElement, parserContext); + adapterBuilder.addPropertyReference("source", pollerBeanName); + } + else { + adapterBuilder.addPropertyReference("source", source); + } + if (StringUtils.hasText(channelName)) { + adapterBuilder.addPropertyReference("target", channelName); + } + else { + adapterBuilder.addPropertyReference("target", + this.createDirectChannel(element, parserContext)); + } + } + else if (StringUtils.hasText(target)) { + adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(OutboundChannelAdapter.class); + adapterBuilder.addPropertyReference("target", target); + if (pollerElement != null) { + if (!StringUtils.hasText(channelName)) { + throw new ConfigurationException("outbound channel-adapter with a 'poller' requires a 'channel' to poll"); + } + String pollerBeanName = IntegrationNamespaceUtils.parsePoller(channelName, pollerElement, parserContext); + adapterBuilder.addPropertyReference("source", pollerBeanName); + } + else { + adapterBuilder.addPropertyReference("source", + this.createDirectChannel(element, parserContext)); + } + } + else { + throw new ConfigurationException("either 'source' or 'target' is required"); + } + return adapterBuilder.getBeanDefinition(); + } + + private String createDirectChannel(Element element, ParserContext parserContext) { + String channelId = element.getAttribute("id"); + if (!StringUtils.hasText(channelId)) { + throw new ConfigurationException("The channel-adapter's 'id' attribute is required when no 'channel' " + + "reference has been provided, because that 'id' would be used for the created channel."); + } + BeanDefinitionBuilder channelBuilder = BeanDefinitionBuilder.genericBeanDefinition(DirectChannel.class); + BeanDefinitionHolder holder = new BeanDefinitionHolder(channelBuilder.getBeanDefinition(), channelId); + BeanDefinitionReaderUtils.registerBeanDefinition(holder, parserContext.getRegistry()); + return channelId; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java index acf3d3da09..9b48d9ed80 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -25,12 +25,7 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.core.Conventions; -import org.springframework.integration.dispatcher.PollingDispatcher; -import org.springframework.integration.dispatcher.SimpleDispatcher; -import org.springframework.integration.message.AsyncMessageExchangeTemplate; -import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.transaction.support.DefaultTransactionDefinition; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; @@ -145,7 +140,7 @@ public abstract class IntegrationNamespaceUtils { * @return the name of the poller bean definition */ public static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcher.class); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcherFactoryBean.class); Long period = Long.valueOf(element.getAttribute("period")); PollingSchedule schedule = new PollingSchedule(period); String initialDelay = element.getAttribute("initial-delay"); @@ -158,42 +153,21 @@ public abstract class IntegrationNamespaceUtils { else { schedule.setFixedRate(false); } - String templateBeanName = parseMessageExhangeTemplate(element, parserContext); - builder.addConstructorArgReference(sourceBeanName); - builder.addConstructorArgValue(schedule); - builder.addConstructorArgValue(new SimpleDispatcher()); - builder.addConstructorArgReference(templateBeanName); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor"); + Element txElement = DomUtils.getChildElementByTagName(element, "transactional"); + if (txElement != null) { + builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager")); + builder.addPropertyValue("propagationBehaviorName", txElement.getAttribute("propagation")); + builder.addPropertyValue("isolationLevelName", txElement.getAttribute("isolation")); + builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout")); + builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only")); + } + builder.addPropertyReference("source", sourceBeanName); + builder.addPropertyValue("schedule", schedule); setValueIfAttributeDefined(builder, element, "receive-timeout"); setValueIfAttributeDefined(builder, element, "send-timeout"); setValueIfAttributeDefined(builder, element, "max-messages-per-poll"); return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); } - private static String parseMessageExhangeTemplate(Element element, ParserContext parserContext) { - String taskExecutorRef = element.getAttribute("task-executor"); - Element txElement = DomUtils.getChildElementByTagName(element, "transactional"); - Class beanClass = (StringUtils.hasText(taskExecutorRef)) ? - AsyncMessageExchangeTemplate.class : MessageExchangeTemplate.class; - BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass); - if (StringUtils.hasText(taskExecutorRef)) { - builder.addConstructorArgReference(taskExecutorRef); - } - if (txElement != null) { - builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager")); - builder.addPropertyValue("propagationBehaviorName", DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation")); - builder.addPropertyValue("isolationLevelName", DefaultTransactionDefinition.PREFIX_ISOLATION + txElement.getAttribute("isolation")); - builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout")); - builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only")); - } - String receiveTimeout = element.getAttribute("receive-timeout"); - if (StringUtils.hasText(receiveTimeout)) { - builder.addPropertyValue("receiveTimeout", Long.parseLong(receiveTimeout)); - } - String sendTimeout = element.getAttribute("send-timeout"); - if (StringUtils.hasText(sendTimeout)) { - builder.addPropertyValue("sendTimeout", Long.parseLong(sendTimeout)); - } - return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry()); - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java new file mode 100644 index 0000000000..d7e4c9694a --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java @@ -0,0 +1,174 @@ +/* + * 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.config; + +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.dispatcher.SimpleDispatcher; +import org.springframework.integration.message.AsyncMessageExchangeTemplate; +import org.springframework.integration.message.MessageExchangeTemplate; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.PollableSource; +import org.springframework.integration.scheduling.PollingSchedule; +import org.springframework.integration.scheduling.Schedule; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.support.DefaultTransactionDefinition; + +/** + * @author Mark Fisher + */ +public class PollingDispatcherFactoryBean implements FactoryBean, InitializingBean { + + private volatile PollingDispatcher poller; + + private volatile MessageSource source; + + private volatile Schedule schedule; + + private volatile long receiveTimeout = -1; + + private volatile long sendTimeout = -1; + + private volatile int maxMessagesPerPoll = -1; + + private volatile TaskExecutor taskExecutor; + + private volatile PlatformTransactionManager transactionManager; + + private volatile String propagationBehaviorName; + + private volatile String isolationLevelName; + + private volatile int transactionTimeout; + + private volatile boolean transactionReadOnly; + + private volatile boolean validated; + + private volatile boolean initialized; + + private final Object initializationMonitor = new Object(); + + + public void setSource(MessageSource source) { + this.source = source; + } + + public void setSchedule(Schedule schedule) { + this.schedule = schedule; + } + + public void setReceiveTimeout(long receiveTimeout) { + this.receiveTimeout = receiveTimeout; + } + + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + + public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { + this.maxMessagesPerPoll = maxMessagesPerPoll; + } + + public void setTaskExecutor(TaskExecutor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + public void setTransactionManager(PlatformTransactionManager transactionManager) { + this.transactionManager = transactionManager; + } + + public void setPropagationBehaviorName(String propagationBehaviorName) { + this.propagationBehaviorName = propagationBehaviorName; + } + + public void setIsolationLevelName(String isolationLevelName) { + this.isolationLevelName = isolationLevelName; + } + + public void setTransactionTimeout(int transactionTimeout) { + this.transactionTimeout = transactionTimeout; + } + + public void setTransactionReadOnly(boolean transactionReadOnly) { + this.transactionReadOnly = transactionReadOnly; + } + + public void afterPropertiesSet() { + synchronized (this.initializationMonitor) { + if (this.source == null) { + throw new ConfigurationException("source is required"); + } + if (!(this.source instanceof PollableSource)) { + throw new BeanCreationException("Poller requires a PollableSource, but actual type of '" + + this.source + "' is [" + this.source.getClass() + "]"); + } + this.validated = true; + } + } + + public Object getObject() throws Exception { + if (!this.initialized) { + this.initPoller(); + } + return this.poller; + } + + public Class getObjectType() { + return PollingDispatcher.class; + } + + public boolean isSingleton() { + return true; + } + + private void initPoller() { + synchronized (this.initializationMonitor) { + if (this.initialized) { + return; + } + if (!this.validated) { + this.afterPropertiesSet(); + } + if (this.schedule == null) { + this.schedule = new PollingSchedule(0); + } + MessageExchangeTemplate template = this.createMessageExchangeTemplate(); + this.poller = new PollingDispatcher((PollableSource) this.source, this.schedule, new SimpleDispatcher(), template); + this.poller.setMaxMessagesPerPoll(this.maxMessagesPerPoll); + this.initialized = true; + } + } + + private MessageExchangeTemplate createMessageExchangeTemplate() { + MessageExchangeTemplate template = (this.taskExecutor != null) ? + new AsyncMessageExchangeTemplate(this.taskExecutor) : new MessageExchangeTemplate(); + template.setTransactionManager(this.transactionManager); + template.setPropagationBehaviorName(DefaultTransactionDefinition.PREFIX_PROPAGATION + this.propagationBehaviorName); + template.setIsolationLevelName(DefaultTransactionDefinition.PREFIX_ISOLATION + this.isolationLevelName); + template.setTransactionTimeout(this.transactionTimeout); + template.setTransactionReadOnly(this.transactionReadOnly); + template.setReceiveTimeout(this.receiveTimeout); + template.setSendTimeout(this.sendTimeout); + return template; + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java index aba8e9f3ca..5c1b21bd1f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/PollableAnnotationPostProcessor.java @@ -24,11 +24,17 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.Pollable; +import org.springframework.integration.annotation.Poller; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.PollableChannelAdapter; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.endpoint.InboundChannelAdapter; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MethodInvokingSource; +import org.springframework.integration.message.PollableSource; +import org.springframework.integration.scheduling.PollingSchedule; /** * Post-processor for methods annotated with {@link Pollable @Pollable}. @@ -48,9 +54,39 @@ public class PollableAnnotationPostProcessor extends AbstractAnnotationMethodPos source.setMethod(method); ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class); if (channelAdapterAnnotation != null) { + Poller pollerAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), Poller.class); + if (pollerAnnotation == null) { + throw new ConfigurationException("The @Poller annotation is required (at class-level) " + + "when using the @ChannelAdapter annotation with a @Pollable method annotation."); + } + PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period()); + schedule.setInitialDelay(pollerAnnotation.initialDelay()); + schedule.setFixedRate(pollerAnnotation.fixedRate()); + schedule.setTimeUnit(pollerAnnotation.timeUnit()); + PollingDispatcher poller = new PollingDispatcher((PollableSource) source, schedule); + int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll(); + if (maxMessagesPerPoll == -1) { + // the default is 1 since a MethodInvokingSource might return a non-null value + // every time it is invoked, thus producing an infinite number of messages per poll + maxMessagesPerPoll = 1; + } + poller.setMaxMessagesPerPoll(maxMessagesPerPoll); + InboundChannelAdapter adapter = new InboundChannelAdapter(); + adapter.setSource(poller); String channelName = channelAdapterAnnotation.value(); - PollableChannelAdapter adapter = new PollableChannelAdapter(channelName, source, null); - this.getMessageBus().registerChannel(adapter); + MessageChannel channel = this.getMessageBus().lookupChannel(channelName); + if (channel == null) { + adapter.setBeanName(channelName + ".adapter"); + DirectChannel directChannel = new DirectChannel(); + directChannel.setBeanName(channelName); + this.getMessageBus().registerChannel(directChannel); + channel = directChannel; + } + else { + adapter.setBeanName(channelName); + } + adapter.setTarget(channel); + this.getMessageBus().registerEndpoint(adapter); } return source; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java index 280f1b31b4..39b5b44f72 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TargetAnnotationPostProcessor.java @@ -24,10 +24,15 @@ import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.ConfigurationException; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.PollableChannelAdapter; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.endpoint.MessageEndpoint; +import org.springframework.integration.endpoint.OutboundChannelAdapter; import org.springframework.integration.handler.MethodInvokingTarget; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.PollableSource; +import org.springframework.integration.scheduling.PollingSchedule; /** * Post-processor for classes annotated with {@link MessageTarget @MessageTarget}. @@ -47,9 +52,29 @@ public class TargetAnnotationPostProcessor extends AbstractAnnotationMethodPostP target.setMethod(method); ChannelAdapter channelAdapterAnnotation = AnnotationUtils.findAnnotation(bean.getClass(), ChannelAdapter.class); if (channelAdapterAnnotation != null) { + OutboundChannelAdapter adapter = new OutboundChannelAdapter(); String channelName = channelAdapterAnnotation.value(); - PollableChannelAdapter adapter = new PollableChannelAdapter(channelName, null, target); - this.getMessageBus().registerChannel(adapter); + MessageChannel channel = this.getMessageBus().lookupChannel(channelName); + if (channel == null) { + adapter.setBeanName(channelName + ".adapter"); + DirectChannel directChannel = new DirectChannel(); + directChannel.setBeanName(channelName); + this.getMessageBus().registerChannel(directChannel); + channel = directChannel; + } + else { + adapter.setBeanName(channelName); + } + if (channel instanceof PollableSource) { + // TODO: add poller config if period, etc is provided (add to @Pollable) + PollingDispatcher poller = new PollingDispatcher((PollableSource) channel, new PollingSchedule(0)); + adapter.setSource(poller); + } + else { + adapter.setSource(channel); + } + adapter.setTarget(target); + this.getMessageBus().registerEndpoint(adapter); } return target; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd index 65f20b374f..6dfac88838 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd @@ -162,12 +162,13 @@ MessageTarget. Therefore, either "source" or "target" should be provided (but never both). - - - - - - + + + + + + + diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index cc5a40ae70..95b8cd9b14 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -20,10 +20,13 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.BeanNameAware; -import org.springframework.integration.message.CompositeMessage; import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.MessagingException; +import org.springframework.integration.scheduling.Schedule; import org.springframework.integration.util.ErrorHandler; /** @@ -37,15 +40,17 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware private volatile String name; + private MessageSource source; + + private MessageTarget target; + + private volatile Schedule schedule; + private volatile ErrorHandler errorHandler; - private volatile boolean requiresReply = false; + private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate(); - public void setBeanName(String name) { - this.name = name; - } - /** * Return the name of this endpoint. */ @@ -53,12 +58,36 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware return this.name; } - /** - * Specify whether this endpoint should throw an Exception when - * it returns an invalid reply Message after handling the request. - */ - public void setRequiresReply(boolean requiresReply) { - this.requiresReply = requiresReply; + public void setBeanName(String name) { + this.name = name; + } + + public MessageSource getSource() { + return this.source; + } + + public void setSource(MessageSource source) { + this.source = source; + } + + public MessageTarget getTarget() { + return this.target; + } + + public void setTarget(MessageTarget target) { + this.target = target; + } + + public Schedule getSchedule() { + return this.schedule; + } + + public void setSchedule(Schedule schedule) { + this.schedule = schedule; + } + + protected MessageExchangeTemplate getMessageExchangeTemplate() { + return this.messageExchangeTemplate; } /** @@ -71,49 +100,29 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware this.errorHandler = errorHandler; } - public final boolean send(Message requestMessage) { - if (requestMessage == null || requestMessage.getPayload() == null) { + public final boolean send(Message message) { + if (message == null || message.getPayload() == null) { throw new IllegalArgumentException("Message and its payload must not be null"); } if (this.logger.isDebugEnabled()) { - this.logger.debug("endpoint '" + this + "' handling message: " + requestMessage); + this.logger.debug("endpoint '" + this + "' processing message: " + message); } try { - Message replyMessage = this.handleRequestMessage(requestMessage); - if (!this.isValidReplyMessage(replyMessage)) { - if (this.requiresReply) { - throw new MessageHandlingException(requestMessage, - "endpoint requires reply but none was received"); - } - } - else if (replyMessage instanceof CompositeMessage) { - for (Message nextReply : (CompositeMessage) replyMessage) { - this.sendReplyMessage(nextReply, requestMessage); - } - } - else { - this.sendReplyMessage(replyMessage, requestMessage); - } - return true; + return this.sendInternal(message); } catch (Exception e) { if (e instanceof MessagingException) { this.handleException((MessagingException) e); } else { - this.handleException(new MessageHandlingException(requestMessage, + this.handleException(new MessageHandlingException(message, "failure occurred in endpoint's send operation", e)); } return false; } } - protected abstract Message handleRequestMessage(Message requestMessage); - - protected abstract boolean isValidReplyMessage(Message replyMessage); - - protected abstract void sendReplyMessage(Message replyMessage, Message requestMessage); - + protected abstract boolean sendInternal(Message message); private void handleException(MessagingException exception) { if (this.errorHandler == null) { @@ -125,7 +134,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, BeanNameAware this.errorHandler.handle(exception); } - public String toString() { return (this.name != null) ? this.name : super.toString(); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java new file mode 100644 index 0000000000..a00eaa9faf --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractRequestReplyEndpoint.java @@ -0,0 +1,104 @@ +/* + * 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; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.CompositeMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageSource; +import org.springframework.integration.message.MessageTarget; + +/** + * @author Mark Fisher + */ +public abstract class AbstractRequestReplyEndpoint extends AbstractEndpoint { + + private volatile String inputChannelName; + + private volatile String outputChannelName; + + private volatile boolean requiresReply = false; + + + public String getInputChannelName() { + return this.inputChannelName; + } + + public void setInputChannelName(String inputChannelName) { + this.inputChannelName = inputChannelName; + } + + public String getOutputChannelName() { + return this.outputChannelName; + } + + public void setOutputChannelName(String outputChannelName) { + this.outputChannelName = outputChannelName; + } + + /** + * Specify whether this endpoint should throw an Exception when + * it returns an invalid reply Message after handling the request. + */ + public void setRequiresReply(boolean requiresReply) { + this.requiresReply = requiresReply; + } + + + protected boolean sendInternal(Message requestMessage) { + Message replyMessage = this.handleRequestMessage(requestMessage); + if (!this.isValidReplyMessage(replyMessage)) { + if (this.requiresReply) { + throw new MessageHandlingException(requestMessage, + "endpoint requires reply but none was received"); + } + } + else if (replyMessage instanceof CompositeMessage) { + for (Message nextReply : (CompositeMessage) replyMessage) { + this.sendReplyMessage(nextReply, requestMessage); + } + } + else { + this.sendReplyMessage(replyMessage, requestMessage); + } + return true; + } + + protected abstract Message handleRequestMessage(Message requestMessage); + + protected abstract boolean isValidReplyMessage(Message replyMessage); + + protected abstract void sendReplyMessage(Message replyMessage, Message requestMessage); + + @Override + public void setSource(MessageSource source) { + if (source instanceof MessageChannel) { + this.setInputChannelName(((MessageChannel) source).getName()); + } + super.setSource(source); + } + + @Override + public void setTarget(MessageTarget target) { + if (target instanceof MessageChannel) { + this.setOutputChannelName(((MessageChannel) target).getName()); + } + super.setTarget(target); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java index 0119172ee6..3b0a26adc3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/DefaultEndpoint.java @@ -28,14 +28,11 @@ import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.CompositeMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.MessageHeaders; import org.springframework.integration.message.MessageRejectedException; -import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; import org.springframework.integration.message.selector.MessageSelector; -import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; /** @@ -60,20 +57,16 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class DefaultEndpoint extends AbstractEndpoint implements ChannelRegistryAware { +public class DefaultEndpoint extends AbstractRequestReplyEndpoint implements ChannelRegistryAware { private final T handler; - private volatile MessageChannel outputChannel; - private volatile ChannelRegistry channelRegistry; private volatile MessageSelector selector; private final List interceptors = new ArrayList(); - private final MessageExchangeTemplate messageExchangeTemplate = new MessageExchangeTemplate(); - /** * Create an endpoint for the given handler. @@ -83,15 +76,6 @@ public class DefaultEndpoint extends AbstractEndpoint this.handler = handler; } - - /** - * Specify the channel where reply Messages should be sent if - * no 'nextTarget' header value is available on the reply Message. - */ - public void setOutputChannel(MessageChannel outputChannel) { - this.outputChannel = outputChannel; - } - protected T getHandler() { return this.handler; } @@ -127,7 +111,7 @@ public class DefaultEndpoint extends AbstractEndpoint * target. The default value indicates an indefinite timeout. */ public void setReplyTimeout(long replyTimeout) { - this.messageExchangeTemplate.setSendTimeout(replyTimeout); + this.getMessageExchangeTemplate().setSendTimeout(replyTimeout); } @Override @@ -170,7 +154,7 @@ public class DefaultEndpoint extends AbstractEndpoint replyMessage = MessageBuilder.fromMessage(replyMessage) .copyHeadersIfAbsent(requestMessage.getHeaders()) .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, requestMessage.getHeaders().getId()).build(); - if (!this.messageExchangeTemplate.send(replyMessage, replyTarget)) { + if (!this.getMessageExchangeTemplate().send(replyMessage, replyTarget)) { throw new MessageEndpointReplyException(replyMessage, requestMessage, "failed to send reply to '" + replyTarget + "'"); } @@ -221,7 +205,7 @@ public class DefaultEndpoint extends AbstractEndpoint private MessageTarget resolveReplyTarget(Message replyMessage, MessageHeaders requestHeaders) { MessageTarget replyTarget = this.resolveTargetAttribute(replyMessage.getHeaders().getNextTarget()); if (replyTarget == null) { - replyTarget = this.outputChannel; + replyTarget = this.getTarget(); } if (replyTarget == null) { replyTarget = this.resolveTargetAttribute(requestHeaders.getReturnAddress()); @@ -245,54 +229,17 @@ public class DefaultEndpoint extends AbstractEndpoint return replyTarget; } - /* TODO: the following properties/methods are candidates for removal from the MessageEndpoint interface. */ + // TODO: remove - private volatile String inputChannelName; - private volatile String outputChannelName; - private volatile MessageSource source; - private volatile Schedule schedule; - - public String getInputChannelName() { - return this.inputChannelName; + public void setReturnAddressOverrides(boolean returnAddressOverrides) { } - public void setInputChannelName(String inputChannelName) { - this.inputChannelName = inputChannelName; - } - - public String getOutputChannelName() { - return this.outputChannelName; - } - - public void setOutputChannelName(String outputChannelName) { - this.outputChannelName = outputChannelName; - } - - public void setReturnAddressOverrides(boolean b) { - } - - public Schedule getSchedule() { - return this.schedule; - } - - public void setSchedule(Schedule schedule) { - this.schedule = schedule; - } - - public MessageSource getSource() { - return this.source; - } - - public MessageTarget getTarget() { - return this.outputChannel; - } - - public void setSource(MessageSource source) { - this.source = source; - } - - public void setTarget(MessageTarget target) { - this.outputChannel = (MessageChannel) target; + /** + * Specify the channel where reply Messages should be sent if + * no 'nextTarget' header value is available on the reply Message. + */ + public void setOutputChannel(MessageChannel outputChannel) { + this.setTarget(outputChannel); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java new file mode 100644 index 0000000000..0ba5faa18d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/InboundChannelAdapter.java @@ -0,0 +1,67 @@ +/* + * 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; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageDeliveryAware; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.MessagingException; + +/** + * A Channel Adapter implementation for connecting a + * {@link org.springframework.integration.message.MessageSource} + * to a {@link MessageChannel}. + * + * @author Mark Fisher + */ +public class InboundChannelAdapter extends AbstractEndpoint { + + @Override + protected boolean sendInternal(Message message) { + try { + boolean sent = this.getMessageExchangeTemplate().send(message, this.getTarget()); + if (sent && this.getSource() instanceof MessageDeliveryAware) { + ((MessageDeliveryAware) this.getSource()).onSend(message); + } + return sent; + } + catch (Exception e) { + MessagingException exception = (e instanceof MessagingException) ? (MessagingException) e + : new MessageDeliveryException(message, "channel-adapter failed to send message to target"); + if (this.getSource() instanceof MessageDeliveryAware) { + ((MessageDeliveryAware) this.getSource()).onFailure(exception); + } + return false; + } + } + + public String getInputChannelName() { + return null; + } + + public String getOutputChannelName() { + if (this.getTarget() instanceof MessageChannel) { + return ((MessageChannel) this.getTarget()).getName(); + } + return null; + } + + public void setInputChannelName(String inputChannelName) { + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java new file mode 100644 index 0000000000..71b1f989c0 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/OutboundChannelAdapter.java @@ -0,0 +1,49 @@ +/* + * 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; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; + +/** + * A Channel Adapter implementation for connecting a {@link MessageChannel} + * to a {@link org.springframework.integration.message.MessageTarget}. + * + * @author Mark Fisher + */ +public class OutboundChannelAdapter extends AbstractEndpoint { + + @Override + protected boolean sendInternal(Message message) { + return this.getMessageExchangeTemplate().send(message, this.getTarget()); + } + + public String getInputChannelName() { + if (this.getSource() instanceof MessageChannel) { + return ((MessageChannel) this.getSource()).getName(); + } + return null; + } + + public String getOutputChannelName() { + return null; + } + + public void setInputChannelName(String inputChannelName) { + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java index 0938ea8eea..491a7065bd 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageExchangeTemplate.java @@ -32,6 +32,7 @@ import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; +import org.springframework.util.Assert; /** * This is the central class for invoking message exchange operations @@ -192,6 +193,7 @@ public class MessageExchangeTemplate implements InitializingBean { private boolean doSend(Message message, MessageTarget target) { + Assert.notNull(target, "target must not be null"); long timeout = this.sendTimeout; boolean sent = (timeout >= 0 && target instanceof BlockingTarget) ? ((BlockingTarget) target).send(message, timeout) @@ -203,6 +205,7 @@ public class MessageExchangeTemplate implements InitializingBean { } private Message doReceive(PollableSource source) { + Assert.notNull(source, "source must not be null"); long timeout = this.receiveTimeout; Message message = (timeout >= 0 && source instanceof BlockingSource) ? ((BlockingSource) source).receive(timeout) diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java index 9ac9bd569b..efaeeaed03 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java @@ -30,10 +30,10 @@ import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.channel.PollableChannelAdapter; import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.DefaultEndpoint; +import org.springframework.integration.endpoint.InboundChannelAdapter; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; import org.springframework.integration.message.GenericMessage; @@ -223,17 +223,10 @@ public class DefaultMessageBusTests { public void testErrorChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new DefaultMessageBus(); CountDownLatch latch = new CountDownLatch(1); - PollableChannelAdapter channelAdapter = new PollableChannelAdapter( - "testChannel", new FailingSource(latch), null); - MessageHandler handler = new MessageHandler() { - public Message handle(Message message) { - return message; - } - }; - DefaultEndpoint endpoint = new DefaultEndpoint(handler); - endpoint.setBeanName("testEndpoint"); - endpoint.setSource(channelAdapter); - bus.registerEndpoint(endpoint); + InboundChannelAdapter channelAdapter = new InboundChannelAdapter(); + channelAdapter.setSource(new FailingSource(latch)); + channelAdapter.setBeanName("testChannel"); + bus.registerEndpoint(channelAdapter); bus.start(); latch.await(2000, TimeUnit.MILLISECONDS); Message message = ((PollableChannel) bus.getErrorChannel()).receive(5000); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanContextTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanContextTests.java deleted file mode 100644 index abad2054b5..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanContextTests.java +++ /dev/null @@ -1,51 +0,0 @@ -/* - * 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.channel.config; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; - -import org.junit.Test; - -import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.StringMessage; - -/** - * @author Mark Fisher - */ -public class ChannelAdapterFactoryBeanContextTests { - - @Test - public void testPollableSourceWithoutTarget() throws Exception { - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( - "channelAdapterFactoryBeanTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) - context.getBean("adapterWithPollableSourceAndNoTarget"); - assertTrue(channel instanceof PollableChannel); - assertEquals("adapterWithPollableSourceAndNoTarget", channel.getName()); - Message reply = ((PollableChannel) channel).receive(); - assertNotNull(reply); - assertEquals("test", reply.getPayload()); - assertFalse(channel.send(new StringMessage("no target"))); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanTests.java deleted file mode 100644 index 8de12aa7c1..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/ChannelAdapterFactoryBeanTests.java +++ /dev/null @@ -1,99 +0,0 @@ -/* - * 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.channel.config; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - -import java.util.concurrent.atomic.AtomicReference; - -import org.junit.Test; - -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageTarget; -import org.springframework.integration.message.StringMessage; -import org.springframework.integration.message.SubscribableSource; - -/** - * @author Mark Fisher - */ -public class ChannelAdapterFactoryBeanTests { - - @Test - public void testPollableSourceWithoutTarget() throws Exception { - ChannelAdapterFactoryBean factoryBean = new ChannelAdapterFactoryBean(); - factoryBean.setBeanName("testChannel"); - factoryBean.setSource(new TestPollableSource()); - Object bean = factoryBean.getObject(); - assertTrue(bean instanceof PollableChannel); - PollableChannel channel = (PollableChannel) bean; - Message reply = channel.receive(); - assertNotNull(reply); - assertEquals("test", reply.getPayload()); - assertFalse(channel.send(new StringMessage("no target"))); - } - - @Test - public void testSubscribableSourceWithoutTarget() throws Exception { - ChannelAdapterFactoryBean factoryBean = new ChannelAdapterFactoryBean(); - factoryBean.setBeanName("testChannel"); - TestSubscribableSource source = new TestSubscribableSource(); - factoryBean.setSource(source); - Object bean = factoryBean.getObject(); - assertTrue(bean instanceof MessageChannel); - assertTrue(bean instanceof SubscribableSource); - MessageChannel channel = (MessageChannel) bean; - final AtomicReference> messageReference = new AtomicReference>(); - ((SubscribableSource) bean).subscribe(new MessageTarget() { - public boolean send(Message message) { - messageReference.set(message); - return true; - } - }); - source.publishMessage(new StringMessage("foo")); - assertNotNull(messageReference.get()); - assertEquals("foo", (messageReference.get()).getPayload()); - assertFalse(channel.send(new StringMessage("no target"))); - } - - @Test - public void testTargetOnly() throws Exception { - ChannelAdapterFactoryBean factoryBean = new ChannelAdapterFactoryBean(); - factoryBean.setBeanName("testChannel"); - final AtomicReference> messageRef = new AtomicReference>(); - factoryBean.setTarget(new MessageTarget() { - public boolean send(Message message) { - messageRef.set(message); - return true; - } - }); - Object bean = factoryBean.getObject(); - assertTrue(bean instanceof PollableChannel); - PollableChannel channel = (PollableChannel) bean; - Message reply = channel.receive(0); - assertNull(reply); - assertTrue(channel.send(new StringMessage("hello"))); - assertNotNull(messageRef.get()); - assertEquals("hello", messageRef.get().getPayload()); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelAdapterFactoryBeanTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelAdapterFactoryBeanTests.xml deleted file mode 100644 index 550a1ef223..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/config/channelAdapterFactoryBeanTests.xml +++ /dev/null @@ -1,23 +0,0 @@ - - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml index da8b1bd3bc..a9335910a3 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml @@ -9,9 +9,7 @@ - - - + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java index 8a1a0398df..726ac1f086 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java @@ -16,15 +16,18 @@ package org.springframework.integration.config; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; import org.junit.Test; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.channel.PollableChannelAdapter; -import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.endpoint.OutboundChannelAdapter; +import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests; @@ -37,24 +40,20 @@ public class ChannelAdapterParserTests extends AbstractJUnit4SpringContextTests @Test public void targetOnly() { - Object bean = this.applicationContext.getBean("targetOnly"); - assertTrue(bean instanceof MessageChannel); - assertTrue(bean instanceof PollableChannel); - assertTrue(bean instanceof PollableChannelAdapter); - MessageChannel channel = (MessageChannel) bean; - assertTrue(channel.send(new StringMessage("test"))); - } - - @Test - public void targetWithDatatypeAccepts() { - MessageChannel channel = (MessageChannel) this.applicationContext.getBean("targetWithDatatype"); - assertTrue(channel.send(new StringMessage("test"))); - } - - @Test(expected = MessageDeliveryException.class) - public void targetWithDatatypeRejects() { - MessageChannel channel = (MessageChannel) this.applicationContext.getBean("targetWithDatatype"); - channel.send(new GenericMessage(123)); + String beanName = "outboundWithImplicitChannel"; + Object channel = this.applicationContext.getBean(beanName); + assertTrue(channel instanceof DirectChannel); + MessageBus bus = (MessageBus) this.applicationContext.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + assertNotNull(bus.lookupChannel(beanName)); + Object adapter = bus.lookupEndpoint(beanName + ".adapter"); + assertNotNull(adapter); + assertTrue(adapter instanceof OutboundChannelAdapter); + TestTarget target = (TestTarget) this.applicationContext.getBean("target"); + assertNull(target.getLastMessage()); + Message message = new StringMessage("test"); + assertTrue(((MessageChannel) channel).send(message)); + assertNotNull(target.getLastMessage()); + assertEquals(message, target.getLastMessage()); } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestTarget.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestTarget.java index ac4ee00821..7e0bad3ce4 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/TestTarget.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/TestTarget.java @@ -24,7 +24,15 @@ import org.springframework.integration.message.MessageTarget; */ public class TestTarget implements MessageTarget { + private volatile Message lastMessage; + + + public Message getLastMessage() { + return this.lastMessage; + } + public boolean send(Message message) { + this.lastMessage = message; return true; } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index fb0f001b76..a751b21e68 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -29,12 +29,12 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.context.ApplicationContext; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.core.annotation.Order; @@ -52,6 +52,7 @@ import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.bus.MessageBus; 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.QueueChannel; @@ -91,22 +92,24 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testSimpleHandlerWithContext() { - ApplicationContext context = new ClassPathXmlApplicationContext( + AbstractApplicationContext context = new ClassPathXmlApplicationContext( "handlerAnnotationPostProcessorTests.xml", this.getClass()); MessageHandler handler = (MessageHandler) context.getBean("simpleHandler"); Message reply = handler.handle(new StringMessage("world")); assertEquals("hello world", reply.getPayload()); + context.stop(); } @Test public void testSimpleHandlerEndpointWithContext() { - ApplicationContext context = new ClassPathXmlApplicationContext( + AbstractApplicationContext context = new ClassPathXmlApplicationContext( "handlerAnnotationPostProcessorTests.xml", this.getClass()); MessageChannel inputChannel = (MessageChannel) context.getBean("inputChannel"); PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel"); inputChannel.send(new StringMessage("foo")); Message reply = outputChannel.receive(1000); assertEquals("hello foo", reply.getPayload()); + context.stop(); } @Test @@ -381,16 +384,27 @@ public class MessagingAnnotationPostProcessorTests { } @Test - public void testChannelAdapterAnnotation() { + public void testChannelAdapterAnnotation() throws InterruptedException { MessageBus messageBus = new DefaultMessageBus(); MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); postProcessor.afterPropertiesSet(); ChannelAdapterAnnotationTestBean testBean = new ChannelAdapterAnnotationTestBean(); postProcessor.postProcessAfterInitialization(testBean, "testBean"); messageBus.start(); - PollableChannel testChannel = (PollableChannel) messageBus.lookupChannel("testChannel"); - Message message = testChannel.receive(1000); - assertEquals("test", message.getPayload()); + DirectChannel testChannel = (DirectChannel) messageBus.lookupChannel("testChannel"); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference> receivedMessage = new AtomicReference>(); + testChannel.subscribe(new org.springframework.integration.message.MessageTarget() { + public boolean send(Message message) { + receivedMessage.set(message); + latch.countDown(); + return false; + } + }); + latch.await(3, TimeUnit.SECONDS); + assertEquals(0, latch.getCount()); + assertNotNull(receivedMessage.get()); + assertEquals("test", receivedMessage.get().getPayload()); messageBus.stop(); } @@ -534,6 +548,7 @@ public class MessagingAnnotationPostProcessorTests { @ChannelAdapter("testChannel") + @Poller(period = 1000, initialDelay = 0, maxMessagesPerPoll = 1) private static class ChannelAdapterAnnotationTestBean { @Pollable diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java deleted file mode 100644 index a302b63dcb..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MethodInvokingAdapterTests.java +++ /dev/null @@ -1,60 +0,0 @@ -/* - * 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.handler; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; - -import java.io.IOException; - -import org.junit.Test; - -import org.springframework.context.support.AbstractApplicationContext; -import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.PollableChannel; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.StringMessage; - -/** - * @author Mark Fisher - */ -public class MethodInvokingAdapterTests { - - @Test - public void testMethodInvokingSourceChannelAdapter() throws IOException, InterruptedException { - AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTests.xml", this.getClass()); - PollableChannel channel = (PollableChannel) context.getBean("sourceChannelAdapter"); - Message message = channel.receive(1000); - assertNotNull(message); - assertEquals("foo", message.getPayload()); - } - - @Test - public void testMethodInvokingTargetChannelAdapter() throws IOException, InterruptedException { - AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTests.xml", this.getClass()); - MessageChannel channel = (MessageChannel) context.getBean("targetChannelAdapter"); - TestSink sink = (TestSink) context.getBean("sink"); - assertNull(sink.get()); - channel.send(new StringMessage("foo")); - String result = sink.get(); - assertNotNull(result); - assertEquals("foo", result); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml deleted file mode 100644 index 14a1a45c5d..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTests.xml +++ /dev/null @@ -1,29 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTestsWithNamespace.xml b/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTestsWithNamespace.xml deleted file mode 100644 index 3c33549674..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/adapterTestsWithNamespace.xml +++ /dev/null @@ -1,24 +0,0 @@ - - - - - - - - - - - - - - - - - -