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

INT-1451 refactoring part-2

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-09-25 13:19:32 -04:00
parent ac49187a64
commit 176115af57
28 changed files with 296 additions and 565 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -31,7 +31,6 @@ import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollerFactory;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.util.Assert;
@@ -159,13 +158,10 @@ public class ConsumerEndpointFactoryBean
Assert.notNull(this.pollerMetadata, "No poller has been defined for endpoint '" + this.beanName
+ "', and no default poller is available within the context.");
}
pollingConsumer.setTrigger(this.pollerMetadata.getTrigger());
pollingConsumer.setPollerMetadata(this.pollerMetadata);
pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout());
PollerFactory pollerFactory = new PollerFactory(pollerMetadata);
pollerFactory.setBeanFactory(this.beanFactory);
pollerFactory.setBeanClassLoader(this.beanClassLoader);
pollingConsumer.setPollerFactory(pollerFactory);
pollingConsumer.setBeanClassLoader(beanClassLoader);
pollingConsumer.setBeanFactory(beanFactory);
this.endpoint = pollingConsumer;
}
else {

9
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -27,7 +27,6 @@ import org.springframework.context.SmartLifecycle;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.PollerFactory;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.util.Assert;
@@ -126,11 +125,9 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
Assert.notNull(this.pollerMetadata, "No poller has been defined for channel-adapter '"
+ this.beanName + "', and no default poller is available within the context.");
}
spca.setTrigger(this.pollerMetadata.getTrigger());
PollerFactory pollerFactory = new PollerFactory(pollerMetadata);
pollerFactory.setBeanFactory(this.beanFactory);
pollerFactory.setBeanClassLoader(this.beanClassLoader);
spca.setPollerFactory(pollerFactory);
spca.setPollerMetadata(this.pollerMetadata);
spca.setBeanClassLoader(this.beanClassLoader);
spca.setAutoStartup(this.autoStartup);
spca.setBeanName(this.beanName);
spca.setBeanFactory(this.beanFactory);

42
spring-integration-core/src/main/java/org/springframework/integration/config/xml/PollerParser.java
View File

@@ -29,10 +29,10 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.ManagedProperties;
import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
@@ -98,7 +98,7 @@ public class PollerParser extends AbstractBeanDefinitionParser {
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
if (txElement != null) {
configureTransactionAttributes(txElement, metadataBuilder);
configureTransactionAttributes(txElement, metadataBuilder, parserContext);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(metadataBuilder, element, "task-executor");
return metadataBuilder.getBeanDefinition();
@@ -199,22 +199,32 @@ public class PollerParser extends AbstractBeanDefinitionParser {
}
/**
* Parse a "transactional" element and configure the "transactionManager"
* and "transactionDefinition" properties for the target builder.
* Parse a "transactional" element and configure {@link TransactionAttributeSourceAdvisor} with "transactionManager"
* and other "transactionDefinition" properties. This advisor will be applied on Polling Task proxy
* (see {@link AbstractPollingEndpoint}).
*/
private void configureTransactionAttributes(Element txElement, BeanDefinitionBuilder targetBuilder) {
ManagedProperties transactionalProperties = new ManagedProperties();
transactionalProperties.setProperty("transactionManager", txElement.getAttribute("transaction-manager"));
private void configureTransactionAttributes(Element txElement, BeanDefinitionBuilder targetBuilder, ParserContext parserContext) {
String TX_PKG_PREFIX = "org.springframework.transaction.interceptor";
BeanDefinitionBuilder txDefinitionBuilder =
BeanDefinitionBuilder.genericBeanDefinition(TX_PKG_PREFIX + ".DefaultTransactionAttribute");
txDefinitionBuilder.addPropertyValue("propagationBehaviorName", "PROPAGATION_" + txElement.getAttribute("propagation"));
txDefinitionBuilder.addPropertyValue("isolationLevelName", "ISOLATION_" + txElement.getAttribute("isolation"));
txDefinitionBuilder.addPropertyValue("timeout", txElement.getAttribute("timeout"));
txDefinitionBuilder.addPropertyValue("readOnly", txElement.getAttribute("read-only"));
BeanDefinitionBuilder attributeSourceBuilder =
BeanDefinitionBuilder.genericBeanDefinition(TX_PKG_PREFIX + ".MatchAlwaysTransactionAttributeSource");
attributeSourceBuilder.addPropertyValue("transactionAttribute", txDefinitionBuilder.getBeanDefinition());
transactionalProperties.setProperty("PROPAGATION", "PROPAGATION_" + txElement.getAttribute("propagation"));
transactionalProperties.setProperty("ISOLATION", "ISOLATION_" + txElement.getAttribute("isolation"));
transactionalProperties.setProperty("timeout", txElement.getAttribute("timeout"));
transactionalProperties.setProperty("readOnly", txElement.getAttribute("read-only"));
BeanDefinitionBuilder pollingDecoratorBuilder =
BeanDefinitionBuilder.genericBeanDefinition("org.springframework.integration.scheduling.PollerTaskTransactionDecorator");
pollingDecoratorBuilder.addPropertyValue("transactionalProperties", transactionalProperties);
targetBuilder.addPropertyValue("transactionDecorator", pollingDecoratorBuilder.getBeanDefinition());
BeanDefinitionBuilder txInterceptorBuilder =
BeanDefinitionBuilder.genericBeanDefinition(TX_PKG_PREFIX + ".TransactionInterceptor");
txInterceptorBuilder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
txInterceptorBuilder.addPropertyValue("transactionAttributeSource", attributeSourceBuilder.getBeanDefinition());
BeanDefinitionBuilder txAdvisorBuilder = BeanDefinitionBuilder.genericBeanDefinition(TX_PKG_PREFIX + ".TransactionAttributeSourceAdvisor");
txAdvisorBuilder.addConstructorArgValue(txInterceptorBuilder.getBeanDefinition());
String txInterceptorName =
BeanDefinitionReaderUtils.registerWithGeneratedName(txAdvisorBuilder.getBeanDefinition(), parserContext.getRegistry());
targetBuilder.addPropertyReference("transactionAdvisor", txInterceptorName);
}
/**

137
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -15,23 +15,42 @@
*/
package org.springframework.integration.endpoint;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ScheduledFuture;
import org.aopalliance.aop.Advice;
import org.springframework.aop.Advisor;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingException;
import org.springframework.scheduling.Trigger;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
*/
public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements InitializingBean{
public abstract class AbstractPollingEndpoint extends AbstractEndpoint implements InitializingBean, BeanClassLoaderAware, Callable<Boolean>{
private volatile TaskExecutor taskExecutor = new SyncTaskExecutor();
private ErrorHandler errorHandler;
private volatile Trigger trigger;
private PollerFactory pollerFactory;
private volatile PollerMetadata pollerMetadata;
private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
private volatile ScheduledFuture<?> runningTask;
private volatile Runnable poller;
@@ -45,46 +64,56 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
public AbstractPollingEndpoint() {
this.setPhase(Integer.MAX_VALUE);
}
/**
* @param trigger
*/
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
/**
* @param pollerFactory
*/
public void setPollerFactory(PollerFactory pollerFactory) {
this.pollerFactory = pollerFactory;
}
@Override
protected void onInit() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
Assert.notNull(this.trigger, "trigger is required");
Assert.notNull(this.pollerMetadata.getTrigger(), "trigger is required");
Assert.notNull(this.getBeanFactory(), "BeanFactory must be provided");
TaskExecutor executor = pollerMetadata.getTaskExecutor();
if (executor != null){
taskExecutor = executor;
}
if (taskExecutor != null){
if (!(taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (errorHandler == null) {
errorHandler = new MessagePublishingErrorHandler(
new BeanFactoryChannelResolver(getBeanFactory()));
}
taskExecutor = new ErrorHandlingTaskExecutor(taskExecutor, errorHandler);
}
}
try {
this.poller = this.createPoller();
this.initialized = true;
} catch (Exception e) {
throw new MessagingException("Problems creating a poller", e);
throw new MessagingException("Failed to create Poller", e);
}
}
}
@SuppressWarnings("unchecked")
private Runnable createPoller() throws Exception{
Callable<Boolean> pollingTask = new Callable<Boolean>() {
public Boolean call() throws Exception {
return doPoll();
}
};
if (pollerFactory == null){
poller = new Poller(pollingTask);
} else {
poller = pollerFactory.createPoller(pollingTask);
ProxyFactory proxyFactory = new ProxyFactory(this);
// Add Transaction advice first
Advisor transactionAdvice = this.pollerMetadata.getTransactionAdvisor();
if (transactionAdvice != null){
proxyFactory.addAdvisor(transactionAdvice);
}
return poller;
// . . .then add the rest of the advises
List<Advice> adviceChain = this.pollerMetadata.getAdviceChain();
if (!CollectionUtils.isEmpty(adviceChain)){
for (Advice advice : adviceChain) {
proxyFactory.addAdvice(advice);
}
}
return new Poller((Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader));
}
// LifecycleSupport implementation
@@ -96,7 +125,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
}
Assert.state(this.getTaskScheduler() != null,
"unable to start polling, no taskScheduler available");
this.runningTask = this.getTaskScheduler().schedule(this.poller, this.trigger);
this.runningTask = this.getTaskScheduler().schedule(this.poller, this.pollerMetadata.getTrigger());
}
@Override // guarded by super#lifecycleLock
@@ -106,6 +135,52 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
}
this.runningTask = null;
}
public void setPollerMetadata(PollerMetadata pollerMetadata) {
this.pollerMetadata = pollerMetadata;
}
public void setBeanClassLoader(ClassLoader classLoader){
this.beanClassLoader = classLoader;
}
public void setErrorHandler(ErrorHandler errorHandler) {
this.errorHandler = errorHandler;
}
/**
* Default Poller implementation
*/
private class Poller implements Runnable {
private final long maxMessagesPerPoll = pollerMetadata.getMaxMessagesPerPoll();
private final Callable<Boolean> pollingTask;
public Poller(Callable<Boolean> pollingTask){
this.pollingTask = pollingTask;
}
protected abstract boolean doPoll();
public void run() {
taskExecutor.execute(new Runnable() {
public void run() {
int count = 0;
while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) {
try {
if (!pollingTask.call()){
break;
}
count++;
} catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException)e;
} else {
throw new MessageHandlingException(new ErrorMessage(e));
}
}
}
}
});
}
}
}

89
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AsyncInvokerAdvice.java
View File

@@ -1,89 +0,0 @@
/*
* Copyright 2002-2010 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 java.util.concurrent.Executor;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.util.ErrorHandler;
/**
* Simple advise to support async execution of tasks.
* It will simply delegate <code>invocation.proceed()</code> calls to its {@link TaskExecutor}
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public class AsyncInvokerAdvice implements MethodInterceptor, InitializingBean,BeanFactoryAware {
private Executor taskExecutor;
private volatile ErrorHandler errorHandler;
private BeanFactory beanFactory;
/**
* @param taskExecutor
*/
public AsyncInvokerAdvice(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/*
* (non-Javadoc)
* @see org.aopalliance.intercept.MethodInterceptor#invoke(org.aopalliance.intercept.MethodInvocation)
*/
public Object invoke(final MethodInvocation invocation) throws Throwable {
taskExecutor.execute(new Runnable() {
public void run() {
try {
invocation.proceed();
} catch (Throwable e) {
if (e instanceof RuntimeException){
throw (RuntimeException)e;
} else {
throw new MessagingException("Problems during asynchronous invocation of task: " + this, e);
}
}
}
});
return null;
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
if (this.errorHandler == null) {
this.errorHandler = new MessagePublishingErrorHandler(
new BeanFactoryChannelResolver(this.beanFactory));
}
this.taskExecutor = new ErrorHandlingTaskExecutor(taskExecutor, errorHandler);
}
}
/*
* (non-Javadoc)
* @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(org.springframework.beans.factory.BeanFactory)
*/
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
}

76
spring-integration-core/src/main/java/org/springframework/integration/endpoint/Poller.java
View File

@@ -1,76 +0,0 @@
/*
* Copyright 2002-2010 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 java.util.concurrent.Callable;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.message.ErrorMessage;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class Poller implements Runnable {
public static final int MAX_MESSAGES_UNBOUNDED = -1;
private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
private Callable<Boolean> pollingTask;
/**
* @param pollingTask
*/
public Poller(Callable<Boolean> pollingTask){
this.pollingTask = pollingTask;
}
/* (non-Javadoc)
* @see java.lang.Runnable#run()
*/
public void run() {
int count = 0;
while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) {
try {
boolean computed = pollingTask.call();
if (!computed){
break;
}
count++;
} catch (Exception e) {
if (e instanceof RuntimeException) {
throw (RuntimeException)e;
} else {
throw new MessageHandlingException(new ErrorMessage(e));
}
}
}
}
/**
*
* @return
*/
public long getMaxMessagesPerPoll() {
return maxMessagesPerPoll;
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*
* <p>The default is unbounded.
*
* @see #MAX_MESSAGES_UNBOUNDED
*/
public void setMaxMessagesPerPoll(long maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
}

113
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerFactory.java
View File

@@ -1,113 +0,0 @@
/*
* Copyright 2002-2010 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 java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.util.CollectionUtils;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class PollerFactory implements BeanClassLoaderAware, BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile ClassLoader beanClassLoader;
private volatile BeanFactory beanFactory;
private volatile PollerMetadata pollerMetadata;
/**
*
*/
public PollerFactory(){}
/**
*
* @param pollerMetadata
*/
public PollerFactory(PollerMetadata pollerMetadata){
this.pollerMetadata = pollerMetadata;
}
/**
*
* @param pollingTask
* @return
* @throws Exception
*/
@SuppressWarnings("unchecked")
public Runnable createPoller(Callable<Boolean> pollingTask) throws Exception {
if (this.taskDecorationRequired()){
ProxyFactory proxyFactory = new ProxyFactory(pollingTask);
pollingTask = (Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader);
ObjectDecorator transactionDecorator = this.pollerMetadata.getTransactionDecorator();
// take care of TransactionINterceptor first
if (transactionDecorator != null){
pollingTask = (Callable<Boolean>) transactionDecorator.decorate(pollingTask);
logger.info("Polling task has been decorated with TransactionInterceptor to handle transactions");
}
// ... then add more Advises if provided
List<Advice> advices = this.pollerMetadata.getAdviceChain();
if (advices != null){
for (Advice advice : advices) {
((Advised)pollingTask).addAdvice(advice);
logger.info("Polling task has been decorated with " + advice.getClass().getSimpleName());
}
}
}
Runnable poller = new Poller(pollingTask);
if (pollerMetadata != null){
((Poller)poller).setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
}
// Decorate Poller with AsyncInvokerAdvice
Executor taskExecutor = this.pollerMetadata.getTaskExecutor();
if (taskExecutor != null){
ProxyFactory proxyFactory = new ProxyFactory(poller);
AsyncInvokerAdvice asyncInvokerAdvice = new AsyncInvokerAdvice(taskExecutor);
asyncInvokerAdvice.setBeanFactory(this.beanFactory);
asyncInvokerAdvice.afterPropertiesSet();
proxyFactory.addAdvice(asyncInvokerAdvice);
poller = (Runnable) proxyFactory.getProxy(this.beanClassLoader);
logger.info("Poller has been decorated with AsyncInvokerAdvice for async polling");
}
return poller;
}
public void setBeanClassLoader(ClassLoader beanClassLoader) {
this.beanClassLoader = beanClassLoader;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
public void setPollerMetadata(PollerMetadata pollerMetadata) {
this.pollerMetadata = pollerMetadata;
}
private boolean taskDecorationRequired(){
return pollerMetadata != null &&
( this.pollerMetadata.getTransactionDecorator() != null ||
!CollectionUtils.isEmpty(this.pollerMetadata.getAdviceChain()) );
}
}

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java
View File

@@ -47,8 +47,9 @@ public class PollingConsumer extends AbstractPollingEndpoint {
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
@Override
protected boolean doPoll() {
public Boolean call() {
Message<?> message = (this.receiveTimeout >= 0)
? this.inputChannel.receive(this.receiveTimeout)
: this.inputChannel.receive();

5
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -42,7 +42,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
/**
* Specify the source to be polled for Messages.
*/
@@ -85,8 +84,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
super.onInit();
}
@Override
protected boolean doPoll() {
public Boolean call() throws Exception {
Message<?> message = this.source.receive();
if (message != null) {
if (this.shouldTrack) {
@@ -97,5 +95,4 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
}
return false;
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -32,9 +32,9 @@ import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
/**
@@ -281,7 +281,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
else if (this.replyChannel instanceof PollableChannel) {
PollingConsumer endpoint = new PollingConsumer(
(PollableChannel) this.replyChannel, handler);
endpoint.setTrigger(new PeriodicTrigger(10));
endpoint.setPollerMetadata(new PollerMetadata());
endpoint.setBeanFactory(this.getBeanFactory());
endpoint.setReceiveTimeout(this.replyTimeout);
endpoint.afterPropertiesSet();

42
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java
View File

@@ -17,11 +17,12 @@
package org.springframework.integration.scheduling;
import java.util.List;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.aop.Advisor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
* @author Mark Fisher
@@ -29,24 +30,26 @@ import org.springframework.scheduling.Trigger;
*/
public class PollerMetadata {
private volatile Trigger trigger;
public static final int MAX_MESSAGES_UNBOUNDED = -1;
private volatile Trigger trigger = new PeriodicTrigger(10);
private volatile int maxMessagesPerPoll;
private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
private volatile long receiveTimeout = 1000;
private List<Advice> adviceChain;
private volatile Executor taskExecutor;
private volatile TaskExecutor taskExecutor;
private volatile ObjectDecorator transactionDecorator;
public ObjectDecorator getTransactionDecorator() {
return transactionDecorator;
private volatile Advisor transactionAdvice;
public Advisor getTransactionAdvisor() {
return transactionAdvice;
}
public void setTransactionDecorator(ObjectDecorator transactionDecorator) {
this.transactionDecorator = transactionDecorator;
public void setTransactionAdvisor(Advisor transactionAdvice) {
this.transactionAdvice = transactionAdvice;
}
public void setTrigger(Trigger trigger) {
@@ -57,11 +60,20 @@ public class PollerMetadata {
return this.trigger;
}
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*
* <p>The default is unbounded.
*
* @see #MAX_MESSAGES_UNBOUNDED
*/
public void setMaxMessagesPerPoll(long maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public int getMaxMessagesPerPoll() {
public long getMaxMessagesPerPoll() {
return this.maxMessagesPerPoll;
}
@@ -81,11 +93,11 @@ public class PollerMetadata {
return this.adviceChain;
}
public void setTaskExecutor(Executor taskExecutor) {
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public Executor getTaskExecutor() {
public TaskExecutor getTaskExecutor() {
return this.taskExecutor;
}
}

79
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java
View File

@@ -1,79 +0,0 @@
/*
* Copyright 2002-2010 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.scheduling;
import java.util.Properties;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.endpoint.Poller;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
import org.springframework.transaction.interceptor.TransactionAttributeSourceAdvisor;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
/**
* A simple implementation of {@link ObjectDecorator} which will add
* {@link TransactionInterceptor} advice to any instance of {@link Advised}.
* Currently used to decorate {@link Poller}'s <code>pollingTask</code>.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public class PollerTaskTransactionDecorator implements ObjectDecorator, BeanFactoryAware {
private BeanFactory beanFactory;
private Properties transactionalProperties;
/* (non-Javadoc)
* @see org.springframework.integration.util.ObjectDecorator#decorate(java.lang.Object)
*/
public Object decorate(Object advisedPollingTask) {
Assert.isInstanceOf(Advised.class, advisedPollingTask, "'pollingTask' must be an instance of Advised");
PlatformTransactionManager txManager = (PlatformTransactionManager) this.beanFactory.getBean(transactionalProperties.getProperty("transactionManager"));
DefaultTransactionAttribute txDefinition = new DefaultTransactionAttribute();
txDefinition.setPropagationBehaviorName(transactionalProperties.getProperty("PROPAGATION"));
txDefinition.setIsolationLevelName(transactionalProperties.getProperty("ISOLATION"));
txDefinition.setTimeout(Integer.valueOf(transactionalProperties.getProperty("timeout")));
txDefinition.setReadOnly(transactionalProperties.getProperty("readOnly").equalsIgnoreCase("true"));
MatchAlwaysTransactionAttributeSource attributeSource = new MatchAlwaysTransactionAttributeSource();
attributeSource.setTransactionAttribute(txDefinition);
TransactionInterceptor transactionInterceptor = new TransactionInterceptor();
transactionInterceptor.setTransactionManager(txManager);
transactionInterceptor.setTransactionAttributeSource(attributeSource);
transactionInterceptor.afterPropertiesSet();
((Advised)advisedPollingTask).addAdvisor(new TransactionAttributeSourceAdvisor(transactionInterceptor));
return advisedPollingTask;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
public Properties getTransactionalProperties() {
return transactionalProperties;
}
public void setTransactionalProperties(Properties transactionalProperties) {
this.transactionalProperties = transactionalProperties;
}
}

30
spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java
View File

@@ -1,30 +0,0 @@
/*
* Copyright 2002-2010 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.util;
/**
* Base decorator interface defining common behavior for basic decoration.
*
* @author Oleg Zhurakousky
* @since 2.0
*/
public interface ObjectDecorator {
/**
* @param object
* @return
*/
Object decorate(Object object);
}

16
spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java
View File

@@ -20,12 +20,13 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -39,6 +40,7 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
@@ -67,6 +69,8 @@ public class ApplicationContextMessageBusTests {
handler.setBeanFactory(context);
handler.afterPropertiesSet();
PollingConsumer endpoint = new PollingConsumer(sourceChannel, handler);
endpoint.setPollerMetadata(new PollerMetadata());
endpoint.setBeanFactory(mock(BeanFactory.class));
context.registerEndpoint("testEndpoint", endpoint);
context.refresh();
Message<?> result = targetChannel.receive(3000);
@@ -123,7 +127,11 @@ public class ApplicationContextMessageBusTests {
handler1.setOutputChannel(outputChannel1);
handler2.setOutputChannel(outputChannel2);
PollingConsumer endpoint1 = new PollingConsumer(inputChannel, handler1);
endpoint1.setPollerMetadata(new PollerMetadata());
endpoint1.setBeanFactory(mock(BeanFactory.class));
PollingConsumer endpoint2 = new PollingConsumer(inputChannel, handler2);
endpoint2.setPollerMetadata(new PollerMetadata());
endpoint2.setBeanFactory(mock(BeanFactory.class));
context.registerEndpoint("testEndpoint1", endpoint1);
context.registerEndpoint("testEndpoint2", endpoint2);
context.refresh();
@@ -184,7 +192,9 @@ public class ApplicationContextMessageBusTests {
CountDownLatch latch = new CountDownLatch(1);
SourcePollingChannelAdapter channelAdapter = new SourcePollingChannelAdapter();
channelAdapter.setSource(new FailingSource(latch));
channelAdapter.setTrigger(new PeriodicTrigger(1000));
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(1000));
channelAdapter.setPollerMetadata(pollerMetadata);
channelAdapter.setOutputChannel(outputChannel);
context.registerEndpoint("testChannel", channelAdapter);
context.refresh();
@@ -212,6 +222,8 @@ public class ApplicationContextMessageBusTests {
}
};
PollingConsumer endpoint = new PollingConsumer(errorChannel, handler);
endpoint.setPollerMetadata(new PollerMetadata());
endpoint.setBeanFactory(mock(BeanFactory.class));
context.registerEndpoint("testEndpoint", endpoint);
context.refresh();
errorChannel.send(new ErrorMessage(new RuntimeException("test-exception")));

35
spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml
View File

@@ -5,29 +5,36 @@
http://www.springframework.org/schema/beans/spring-beans.xsd">
<bean id="taskScheduler" class="org.springframework.integration.test.util.TestUtils"
factory-method="createTaskScheduler">
<constructor-arg value="10"/>
factory-method="createTaskScheduler">
<constructor-arg value="10" />
</bean>
<bean id="sourceChannel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="sourceChannel" class="org.springframework.integration.channel.QueueChannel" />
<bean id="targetChannel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="targetChannel" class="org.springframework.integration.channel.QueueChannel" />
<bean id="endpoint" class="org.springframework.integration.endpoint.PollingConsumer">
<constructor-arg ref="sourceChannel"/>
<constructor-arg ref="serviceActivator"/>
<property name="trigger">
<bean class="org.springframework.scheduling.support.PeriodicTrigger">
<constructor-arg value="100"/>
<bean id="endpoint"
class="org.springframework.integration.endpoint.PollingConsumer">
<constructor-arg ref="sourceChannel" />
<constructor-arg ref="serviceActivator" />
<property name="pollerMetadata">
<bean class="org.springframework.integration.scheduling.PollerMetadata">
<property name="trigger">
<bean class="org.springframework.scheduling.support.PeriodicTrigger">
<constructor-arg value="100" />
</bean>
</property>
</bean>
</property>
</bean>
<bean id="serviceActivator" class="org.springframework.integration.handler.ServiceActivatingHandler">
<constructor-arg ref="handler"/>
<property name="outputChannel" ref="targetChannel"/>
<bean id="serviceActivator"
class="org.springframework.integration.handler.ServiceActivatingHandler">
<constructor-arg ref="handler" />
<property name="outputChannel" ref="targetChannel" />
</bean>
<bean id="handler" class="org.springframework.integration.message.TestHandlers" factory-method="echoHandler"/>
<bean id="handler" class="org.springframework.integration.message.TestHandlers"
factory-method="echoHandler" />
</beans>

31
spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java
View File

@@ -18,6 +18,8 @@ package org.springframework.integration.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
import java.util.List;
@@ -28,6 +30,10 @@ import org.aopalliance.aop.Advice;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.aop.Advisor;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
@@ -35,7 +41,6 @@ import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.integration.util.ObjectDecorator;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.ClassUtils;
@@ -70,13 +75,13 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
factoryBean.afterPropertiesSet();
context.registerEndpoint("testPollingEndpoint", factoryBean.getObject());
context.refresh();
Message<?> message = outputChannel.receive(30000);
Message<?> message = outputChannel.receive(5000);
assertEquals("test", message.getPayload());
assertTrue("adviceChain was not applied", adviceApplied.get());
}
@Test
public void testTransactionalAdviceChain() throws Exception {
public void testTransactionalAdviceChain() throws Throwable {
SourcePollingChannelAdapterFactoryBean factoryBean = new SourcePollingChannelAdapterFactoryBean();
QueueChannel outputChannel = new QueueChannel();
TestApplicationContext context = TestUtils.createTestApplicationContext();
@@ -96,19 +101,29 @@ public class SourcePollingChannelAdapterFactoryBeanTests {
pollerMetadata.setTrigger(new PeriodicTrigger(5000));
pollerMetadata.setMaxMessagesPerPoll(1);
final AtomicInteger count = new AtomicInteger();
pollerMetadata.setTransactionDecorator(new ObjectDecorator() {
public Object decorate(Object poller) {
count.incrementAndGet();
return poller;
final MethodInterceptor txAdvice = mock(MethodInterceptor.class);
pollerMetadata.setTransactionAdvisor(new Advisor() {
public boolean isPerInstance() {
return false;
}
public Advice getAdvice() {
return txAdvice;
}
});
when(txAdvice.invoke(Mockito.any(MethodInvocation.class))).thenAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) throws Throwable {
count.incrementAndGet();
return ((MethodInvocation)invocation.getArguments()[0]).proceed();
}
});
pollerMetadata.setAdviceChain(adviceChain);
factoryBean.setPollerMetadata(pollerMetadata);
factoryBean.setAutoStartup(true);
factoryBean.afterPropertiesSet();
context.registerEndpoint("testPollingEndpoint", factoryBean.getObject());
context.refresh();
Message<?> message = outputChannel.receive(30000);
Message<?> message = outputChannel.receive(5000);
assertEquals("test", message.getPayload());
assertEquals(1, count.get());
assertTrue("adviceChain was not applied", adviceApplied.get());

9
spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java
View File

@@ -31,7 +31,6 @@ import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.context.support.StaticApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
@@ -39,14 +38,12 @@ import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MapBasedChannelResolver;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.channel.ChannelResolutionException;
import org.springframework.integration.support.converter.SimpleMessageConverter;
@@ -69,7 +66,9 @@ public class MessagingTemplateTests {
this.requestChannel = new QueueChannel();
context.registerChannel("requestChannel", requestChannel);
PollingConsumer endpoint = new PollingConsumer(requestChannel, new TestHandler());
endpoint.setTrigger(new PeriodicTrigger(10));
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(10));
endpoint.setPollerMetadata(pollerMetadata);
context.registerEndpoint("testEndpoint", endpoint);
context.refresh();
}

6
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java
View File

@@ -34,7 +34,6 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.PollerFactory;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -73,9 +72,8 @@ public class PollingTransactionTests {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionTests.xml", this.getClass());
PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class);
PollerFactory pollerFactory = TestUtils.getPropertyValue(advicedPoller, "pollerFactory",PollerFactory.class);
PollerMetadata pollerMetedata = TestUtils.getPropertyValue(pollerFactory, "pollerMetadata",PollerMetadata.class);
PollerMetadata pollerMetedata = TestUtils.getPropertyValue(advicedPoller, "pollerMetadata",PollerMetadata.class);
List<Advice> adviceChain = TestUtils.getPropertyValue(pollerMetedata, "adviceChain",List.class);
assertEquals(2, adviceChain.size());
Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class);

26
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java
View File

@@ -23,6 +23,7 @@ import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.verify;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import java.util.Date;
import java.util.concurrent.CountDownLatch;
@@ -34,6 +35,7 @@ import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.core.MessageHandler;
@@ -75,10 +77,14 @@ public class PollingConsumerEndpointTests {
trigger.reset();
endpoint = new PollingConsumer(channelMock, consumer);
taskScheduler.setPoolSize(5);
taskScheduler.setErrorHandler(errorHandler);
endpoint.setErrorHandler(errorHandler);
endpoint.setTaskScheduler(taskScheduler);
endpoint.setTrigger(trigger);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setBeanFactory(mock(BeanFactory.class));
endpoint.setReceiveTimeout(-1);
endpoint.afterPropertiesSet();
taskScheduler.afterPropertiesSet();
reset(channelMock);
}
@@ -96,7 +102,8 @@ public class PollingConsumerEndpointTests {
replay(channelMock);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.start();
trigger.await();
endpoint.stop();
@@ -110,7 +117,8 @@ public class PollingConsumerEndpointTests {
replay(channelMock);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.start();
trigger.await();
endpoint.stop();
@@ -125,7 +133,8 @@ public class PollingConsumerEndpointTests {
replay(channelMock);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(6);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.start();
trigger.await();
endpoint.stop();
@@ -160,7 +169,9 @@ public class PollingConsumerEndpointTests {
replay(channelMock);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(10);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
//endpoint.setErrorHandler(null);
endpoint.start();
trigger.await();
endpoint.stop();
@@ -190,7 +201,8 @@ public class PollingConsumerEndpointTests {
endpoint.setReceiveTimeout(1);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.start();
trigger.await();
endpoint.stop();

9
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
@@ -24,11 +25,13 @@ import org.springframework.scheduling.support.PeriodicTrigger;
public class PollingEndpointStub extends AbstractPollingEndpoint {
public PollingEndpointStub() {
this.setTrigger(new PeriodicTrigger(500));
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(500));
this.setPollerMetadata(pollerMetadata);
}
@Override
protected boolean doPoll() {
//@Override
public Boolean call() {
throw new RuntimeException("intentional test failure");
}

4
spring-integration-core/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml
View File

@@ -8,9 +8,7 @@
http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
<si:inbound-channel-adapter id="inboundEndpoint" ref="counter" method="next" channel="producerAndConsumerAutoStartupTestChannel">
<si:poller max-messages-per-poll="1">
<si:interval-trigger interval="50"/>
</si:poller>
<si:poller max-messages-per-poll="1" fixed-rate="50"/>
</si:inbound-channel-adapter>
<si:channel id="producerAndConsumerAutoStartupTestChannel"/>

23
spring-integration-core/src/test/java/org/springframework/integration/endpoint/pollingEndpointErrorHandlingTests.xml
View File

@@ -1,28 +1,25 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:si="http://www.springframework.org/schema/integration"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<si:channel id="errorChannel">
<si:queue capacity="10"/>
<si:queue capacity="10" />
</si:channel>
<bean id="testEndpoint" class="org.springframework.integration.endpoint.PollingEndpointStub">
<property name="pollerFactory">
<bean class="org.springframework.integration.endpoint.PollerFactory">
<constructor-arg>
<bean class="org.springframework.integration.scheduling.PollerMetadata">
<property name="taskExecutor" ref="taskExecutor"/>
</bean>
</constructor-arg>
<bean id="testEndpoint"
class="org.springframework.integration.endpoint.PollingEndpointStub">
<property name="pollerMetadata">
<bean class="org.springframework.integration.scheduling.PollerMetadata">
<property name="taskExecutor" ref="taskExecutor" />
</bean>
</property>
</bean>
<task:executor id="taskExecutor" pool-size="1" rejection-policy="CALLER_RUNS"/>
<task:executor id="taskExecutor" pool-size="1"
rejection-policy="CALLER_RUNS" />
</beans>

1
spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java
View File

@@ -33,7 +33,6 @@ import org.easymock.IAnswer;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageDeliveryException;

5
spring-integration-core/src/test/java/org/springframework/integration/message/MethodInvokingMessageHandlerTests.java
View File

@@ -31,6 +31,7 @@ import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.MethodInvokingMessageHandler;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.test.util.TestUtils.TestApplicationContext;
import org.springframework.scheduling.support.PeriodicTrigger;
@@ -82,7 +83,9 @@ public class MethodInvokingMessageHandlerTests {
assertNull(queue.poll());
MethodInvokingMessageHandler handler = new MethodInvokingMessageHandler(testBean, "foo");
PollingConsumer endpoint = new PollingConsumer(channel, handler);
endpoint.setTrigger(new PeriodicTrigger(10));
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(new PeriodicTrigger(10));
endpoint.setPollerMetadata(pollerMetadata);
context.registerEndpoint("testEndpoint", endpoint);
context.refresh();
String result = queue.poll(2000, TimeUnit.MILLISECONDS);

8
spring-integration-core/src/test/java/org/springframework/integration/test/util/TestUtils.java
View File

@@ -41,11 +41,9 @@ import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.NamedComponent;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
@@ -145,12 +143,6 @@ public abstract class TestUtils {
}
public void registerEndpoint(String endpointName, AbstractEndpoint endpoint) {
if (endpoint instanceof AbstractPollingEndpoint) {
DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
if (accessor.getPropertyValue("trigger") == null) {
((AbstractPollingEndpoint) endpoint).setTrigger(new PeriodicTrigger(10));
}
}
registerBean(endpointName, endpoint, this);
}
}

20
spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java
View File

@@ -28,9 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollerFactory;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -66,7 +64,9 @@ public class ByteStreamWritingMessageHandlerTests {
this.endpoint.setTaskScheduler(scheduler);
scheduler.afterPropertiesSet();
trigger.reset();
endpoint.setTrigger(trigger);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
}
@After
@@ -97,7 +97,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void maxMessagesPerTaskSameAsMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(3);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -114,7 +114,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void maxMessagesPerTaskLessThanMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
@@ -130,7 +130,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void maxMessagesPerTaskExceedsMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -147,7 +147,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -172,7 +172,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -196,7 +196,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void testStreamResetBetweenDispatches() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
@@ -220,7 +220,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void testStreamWriteBetweenDispatches() throws IOException {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);

21
spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java
View File

@@ -27,9 +27,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollerFactory;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -66,7 +64,9 @@ public class CharacterStreamWritingMessageHandlerTests {
this.endpoint.setTaskScheduler(scheduler);
scheduler.afterPropertiesSet();
trigger.reset();
endpoint.setTrigger(trigger);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
}
@After
@@ -85,7 +85,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void twoStringsAndNoNewLinesByDefault() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
endpoint.start();
@@ -104,7 +104,8 @@ public class CharacterStreamWritingMessageHandlerTests {
handler.setShouldAppendNewLine(true);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage<String>("foo"), 0);
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
endpoint.start();
trigger.await();
@@ -122,7 +123,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void maxMessagesPerTaskSameAsMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
endpoint.start();
@@ -135,7 +136,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(10);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
handler.setShouldAppendNewLine(true);
channel.send(new GenericMessage<String>("foo"), 0);
@@ -151,7 +152,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void singleNonStringObject() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
TestObject testObject = new TestObject("foo");
channel.send(new GenericMessage<TestObject>(testObject));
endpoint.start();
@@ -165,7 +166,7 @@ public class CharacterStreamWritingMessageHandlerTests {
endpoint.setReceiveTimeout(0);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");
channel.send(new GenericMessage<TestObject>(testObject1), 0);
@@ -182,7 +183,7 @@ public class CharacterStreamWritingMessageHandlerTests {
endpoint.setReceiveTimeout(0);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerFactory(new PollerFactory(pollerMetadata));
endpoint.setPollerMetadata(pollerMetadata);
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");
channel.send(new GenericMessage<TestObject>(testObject1), 0);

6
spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java
View File

@@ -146,12 +146,6 @@ public abstract class TestUtils {
}
public void registerEndpoint(String endpointName, AbstractEndpoint endpoint) {
if (endpoint instanceof AbstractPollingEndpoint) {
DirectFieldAccessor accessor = new DirectFieldAccessor(endpoint);
if (accessor.getPropertyValue("trigger") == null) {
((AbstractPollingEndpoint) endpoint).setTrigger(new PeriodicTrigger(10));
}
}
registerBean(endpointName, endpoint, this);
}
}