diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index d7576da57a..ef70d76563 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -13,9 +13,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.springframework.integration.config; +import org.springframework.beans.factory.BeanClassLoaderAware; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanNameAware; @@ -32,6 +32,7 @@ import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.scheduling.PollerFactory; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -42,7 +43,7 @@ import org.springframework.util.StringUtils; * @author Josh Long */ public class ConsumerEndpointFactoryBean - implements FactoryBean, BeanFactoryAware, BeanNameAware, InitializingBean, SmartLifecycle { + implements FactoryBean, BeanFactoryAware, BeanNameAware, BeanClassLoaderAware, InitializingBean, SmartLifecycle { private volatile MessageHandler handler; @@ -51,12 +52,14 @@ public class ConsumerEndpointFactoryBean private volatile String inputChannelName; private volatile PollerMetadata pollerMetadata; - + private volatile boolean autoStartup = true; private volatile MessageChannel inputChannel; private volatile ConfigurableBeanFactory beanFactory; + + private volatile ClassLoader beanClassLoader; private volatile AbstractEndpoint endpoint; @@ -86,6 +89,10 @@ public class ConsumerEndpointFactoryBean public void setPollerMetadata(PollerMetadata pollerMetadata) { this.pollerMetadata = pollerMetadata; } + + public void setBeanClassLoader(ClassLoader classLoader) { + this.beanClassLoader = classLoader; + } public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; @@ -153,11 +160,12 @@ public class ConsumerEndpointFactoryBean + "', and no default poller is available within the context."); } pollingConsumer.setTrigger(this.pollerMetadata.getTrigger()); - pollingConsumer.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll()); pollingConsumer.setReceiveTimeout(this.pollerMetadata.getReceiveTimeout()); - pollingConsumer.setTaskExecutor(this.pollerMetadata.getTaskExecutor()); - pollingConsumer.setPollingDecorator(this.pollerMetadata.getPollingDecorator()); - pollingConsumer.setAdviceChain(this.pollerMetadata.getAdviceChain()); + + PollerFactory pollerFactory = new PollerFactory(pollerMetadata); + pollerFactory.setBeanFactory(this.beanFactory); + pollerFactory.setBeanClassLoader(this.beanClassLoader); + pollingConsumer.setPollerFactory(pollerFactory); this.endpoint = pollingConsumer; } else { @@ -205,5 +213,4 @@ public class ConsumerEndpointFactoryBean this.endpoint.stop(callback); } } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/Poller.java b/spring-integration-core/src/main/java/org/springframework/integration/config/Poller.java new file mode 100644 index 0000000000..5862e11c05 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/Poller.java @@ -0,0 +1,76 @@ +/* + * 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.config; + +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 pollingTask; + /** + * @param pollingTask + */ + public Poller(Callable 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. + * + *

The default is unbounded. + * + * @see #MAX_MESSAGES_UNBOUNDED + */ + public void setMaxMessagesPerPoll(long maxMessagesPerPoll) { + this.maxMessagesPerPoll = maxMessagesPerPoll; + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java index 39e0e04a00..ce736fc142 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java @@ -28,6 +28,7 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; +import org.springframework.integration.scheduling.PollerFactory; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.util.Assert; @@ -126,14 +127,13 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean adviceChain = new CopyOnWriteArrayList(); - - private volatile ClassLoader classLoader = ClassUtils.getDefaultClassLoader(); + private PollerFactory pollerFactory; private volatile ScheduledFuture runningTask; @@ -68,52 +41,24 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement private volatile boolean initialized; private final Object initializationMonitor = new Object(); - - + /** + * + */ public AbstractPollingEndpoint() { this.setPhase(Integer.MAX_VALUE); } - - + /** + * @param trigger + */ public void setTrigger(Trigger trigger) { this.trigger = trigger; } - /** - * 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. - * - *

The default is unbounded. - * - * @see #MAX_MESSAGES_UNBOUNDED + * @param pollerFactory */ - public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { - this.maxMessagesPerPoll = maxMessagesPerPoll; + public void setPollerFactory(PollerFactory pollerFactory) { + this.pollerFactory = pollerFactory; } - - public void setTaskExecutor(Executor taskExecutor) { - this.taskExecutor = taskExecutor; - } - - public void setErrorHandler(ErrorHandler errorHandler){ - this.errorHandler = errorHandler; - } - - public void setBeanClassLoader(ClassLoader classLoader) { - Assert.notNull(classLoader, "ClassLoader must not be null"); - this.classLoader = classLoader; - } -// - public void setAdviceChain(List adviceChain) { - synchronized (this.adviceChain) { - this.adviceChain.clear(); - if (adviceChain != null) { - this.adviceChain.addAll(adviceChain); - } - } - } - @Override protected void onInit() { synchronized (this.initializationMonitor) { @@ -121,40 +66,29 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement return; } Assert.notNull(this.trigger, "trigger is required"); - if (this.taskExecutor != null && !(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) { - if (this.errorHandler == null) { - this.errorHandler = new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(getBeanFactory())); - } - this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler); + try { + this.poller = this.createPoller(); + this.initialized = true; + } catch (Exception e) { + throw new MessagingException("Problems creating a poller", e); } - this.poller = this.createPoller(); - this.initialized = true; } } - private Runnable createPoller() { - Runnable poller = new Poller(); - if (pollingDecorator != null){ - poller = (Runnable) pollingDecorator.decorate(poller); - } - if (poller instanceof Advised){ - Advised advised = (Advised) poller; - for (Advice advice : adviceChain) { - advised.addAdvice(advice); + private Runnable createPoller() throws Exception{ + Callable pollingTask = new Callable() { + public Boolean call() throws Exception { + return doPoll(); } + }; + if (pollerFactory == null){ + poller = new Poller(pollingTask); } else { - if (adviceChain.size() > 0){ - ProxyFactory proxyFactory = new ProxyFactory(poller); - for (Advice advice : adviceChain) { - proxyFactory.addAdvice(advice); - } - poller = (Runnable) proxyFactory.getProxy(this.classLoader); - } + poller = pollerFactory.createPoller(pollingTask); } return poller; } - // LifecycleSupport implementation @Override // guarded by super#lifecycleLock @@ -175,37 +109,5 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.runningTask = null; } - protected abstract boolean doPoll(); - - - private class Poller implements Runnable { - - public void run() { - if (taskExecutor != null) { - taskExecutor.execute(new Runnable() { - public void run() { - poll(); - } - }); - } - else { - poll(); - } - } - - private void poll() { - int count = 0; - while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) { - if (!innerPoll()) { - break; - } - count++; - } - } - - private boolean innerPoll() { - return doPoll(); - } - } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 058adbdee0..4f4cbd9d24 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -30,6 +30,7 @@ import org.springframework.util.Assert; * {@link MessageSource} to a {@link MessageChannel}. * * @author Mark Fisher + * @author Oleg Zhurakousky */ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent { @@ -81,11 +82,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme protected void onInit() { Assert.notNull(this.source, "source must not be null"); Assert.notNull(this.outputChannel, "outputChannel must not be null"); - if (this.maxMessagesPerPoll < 0) { - // the default is 1 since a source might return - // a non-null value every time it is invoked - this.setMaxMessagesPerPoll(1); - } super.onInit(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerFactory.java new file mode 100644 index 0000000000..2cde1a42df --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerFactory.java @@ -0,0 +1,114 @@ +/* + * 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.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.config.Poller; +import org.springframework.integration.util.AsyncInvokerAdvice; +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 pollingTask) throws Exception { + if (this.taskDecorationRequired()){ + ProxyFactory proxyFactory = new ProxyFactory(pollingTask); + pollingTask = (Callable) proxyFactory.getProxy(this.beanClassLoader); + ObjectDecorator transactionDecorator = this.pollerMetadata.getTransactionDecorator(); + // take care of TransactionINterceptor first + if (transactionDecorator != null){ + pollingTask = (Callable) transactionDecorator.decorate(pollingTask); + logger.info("Polling task has been decorated with TransactionInterceptor to handle transactions"); + } + // ... then add more Advises if provided + List 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()) ); + } +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java index 514af3f62e..6795003067 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerMetadata.java @@ -20,7 +20,7 @@ import java.util.List; import java.util.concurrent.Executor; import org.aopalliance.aop.Advice; -import org.springframework.integration.endpoint.PollerCallbackDecorator; +import org.springframework.integration.util.ObjectDecorator; import org.springframework.scheduling.Trigger; /** @@ -39,14 +39,14 @@ public class PollerMetadata { private volatile Executor taskExecutor; - private PollerCallbackDecorator pollingDecorator; + private volatile ObjectDecorator transactionDecorator; - public PollerCallbackDecorator getPollingDecorator() { - return pollingDecorator; + public ObjectDecorator getTransactionDecorator() { + return transactionDecorator; } - public void setPollingDecorator(PollerCallbackDecorator pollingDecorator) { - this.pollingDecorator = pollingDecorator; + public void setTransactionDecorator(ObjectDecorator transactionDecorator) { + this.transactionDecorator = transactionDecorator; } public void setTrigger(Trigger trigger) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/TransactionalCallbackDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java similarity index 62% rename from spring-integration-core/src/main/java/org/springframework/integration/config/TransactionalCallbackDecorator.java rename to spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java index a18fd86809..62b5588fde 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/TransactionalCallbackDecorator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java @@ -13,40 +13,41 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.config; +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.PollerCallbackDecorator; +import org.springframework.integration.config.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.TransactionProxyFactoryBean; +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 pollingTask. + * * @author Oleg Zhurakousky * @since 2.0 */ -class TransactionalCallbackDecorator implements PollerCallbackDecorator, BeanFactoryAware { +public class PollerTaskTransactionDecorator implements ObjectDecorator, BeanFactoryAware { private BeanFactory beanFactory; - private Properties transactionalProperties; - public Properties getTransactionalProperties() { - return transactionalProperties; - } - - public void setTransactionalProperties(Properties transactionalProperties) { - this.transactionalProperties = transactionalProperties; - } - - public Object decorate(Object pollingCallback){ - TransactionProxyFactoryBean txFactoryBean = new TransactionProxyFactoryBean(); - txFactoryBean.setBeanFactory(beanFactory); + /* (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")); - txFactoryBean.setTransactionManager(txManager); DefaultTransactionAttribute txDefinition = new DefaultTransactionAttribute(); txDefinition.setPropagationBehaviorName(transactionalProperties.getProperty("PROPAGATION")); txDefinition.setIsolationLevelName(transactionalProperties.getProperty("ISOLATION")); @@ -54,13 +55,25 @@ class TransactionalCallbackDecorator implements PollerCallbackDecorator, BeanFac txDefinition.setReadOnly(transactionalProperties.getProperty("readOnly").equalsIgnoreCase("true")); MatchAlwaysTransactionAttributeSource attributeSource = new MatchAlwaysTransactionAttributeSource(); attributeSource.setTransactionAttribute(txDefinition); - txFactoryBean.setTransactionAttributeSource(attributeSource); - txFactoryBean.setTarget(pollingCallback); - txFactoryBean.afterPropertiesSet(); - return txFactoryBean.getObject(); + + 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; + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/AsyncInvokerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/util/AsyncInvokerAdvice.java new file mode 100644 index 0000000000..9803445de3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/AsyncInvokerAdvice.java @@ -0,0 +1,88 @@ +/* + * 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; + +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.util.ErrorHandler; + +/** + * Simple advise to support async execution of tasks. + * It will simply delegate invocation.proceed() 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; + } +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java similarity index 74% rename from spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java rename to spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java index c274797852..2a8452f204 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerCallbackDecorator.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java @@ -13,11 +13,18 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.endpoint; +package org.springframework.integration.util; + /** + * Base decorator interface defining common behavior for basic decoration. + * * @author Oleg Zhurakousky * @since 2.0 */ -public interface PollerCallbackDecorator { - Object decorate(Object poller); +public interface ObjectDecorator { + /** + * @param object + * @return + */ + Object decorate(Object object); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java index 9260860552..bb12239bd3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBeanTests.java @@ -31,11 +31,11 @@ import org.junit.Test; import org.springframework.integration.Message; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.endpoint.PollerCallbackDecorator; 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; @@ -96,7 +96,7 @@ public class SourcePollingChannelAdapterFactoryBeanTests { pollerMetadata.setTrigger(new PeriodicTrigger(5000)); pollerMetadata.setMaxMessagesPerPoll(1); final AtomicInteger count = new AtomicInteger(); - pollerMetadata.setPollingDecorator(new PollerCallbackDecorator() { + pollerMetadata.setTransactionDecorator(new ObjectDecorator() { public Object decorate(Object poller) { count.incrementAndGet(); return poller; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java index c4b2c8ee07..fe0e698153 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.java @@ -22,6 +22,7 @@ import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import java.util.List; +import java.util.concurrent.Callable; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; @@ -35,6 +36,8 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.scheduling.PollerFactory; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.TestTransactionManager; import org.springframework.transaction.IllegalTransactionStateException; @@ -65,18 +68,22 @@ public class PollingTransactionTests { } @Test + @SuppressWarnings("unchecked") public void transactionWithCommitAndAdvices() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "transactionTests.xml", this.getClass()); PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class); - @SuppressWarnings("unchecked") - List adviceChain = TestUtils.getPropertyValue(advicedPoller, "adviceChain",List.class); + + PollerFactory pollerFactory = TestUtils.getPropertyValue(advicedPoller, "pollerFactory",PollerFactory.class); + PollerMetadata pollerMetedata = TestUtils.getPropertyValue(pollerFactory, "pollerMetadata",PollerMetadata.class); + List adviceChain = TestUtils.getPropertyValue(pollerMetedata, "adviceChain",List.class); assertEquals(2, adviceChain.size()); Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class); - assertTrue("Poller is not Advised", poller instanceof Advised); - Advisor[] advisors = ((Advised)poller).getAdvisors(); + Callable pollingTask = TestUtils.getPropertyValue(poller, "pollingTask", Callable.class); + assertTrue("Poller is not Advised", pollingTask instanceof Advised); + Advisor[] advisors = ((Advised)pollingTask).getAdvisors(); assertEquals(3, advisors.length); - // System.err.println(Arrays.asList(advisors)); + assertTrue("First advisor is not TX", advisors[0] instanceof TransactionAttributeSourceAdvisor); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); MessageChannel input = (MessageChannel) context.getBean("goodInputWithAdvice"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java index 6e814de260..26f4264606 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java @@ -39,6 +39,8 @@ import org.springframework.integration.MessageRejectedException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.scheduling.PollerFactory; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -93,7 +95,9 @@ public class PollingConsumerEndpointTests { expect(channelMock.receive()).andReturn(message); expectLastCall(); replay(channelMock); - endpoint.setMaxMessagesPerPoll(1); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(1); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.start(); trigger.await(); endpoint.stop(); @@ -105,7 +109,9 @@ public class PollingConsumerEndpointTests { public void multipleMessages() { expect(channelMock.receive()).andReturn(message).times(5); replay(channelMock); - endpoint.setMaxMessagesPerPoll(5); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(5); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.start(); trigger.await(); endpoint.stop(); @@ -118,7 +124,9 @@ public class PollingConsumerEndpointTests { expect(channelMock.receive()).andReturn(message).times(5); expect(channelMock.receive()).andReturn(null); replay(channelMock); - endpoint.setMaxMessagesPerPoll(6); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(6); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.start(); trigger.await(); endpoint.stop(); @@ -151,7 +159,9 @@ public class PollingConsumerEndpointTests { public void droppedMessage_onePerPoll() throws Throwable { expect(channelMock.receive()).andReturn(badMessage).times(1); replay(channelMock); - endpoint.setMaxMessagesPerPoll(10); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(10); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.start(); trigger.await(); endpoint.stop(); @@ -179,7 +189,9 @@ public class PollingConsumerEndpointTests { expectLastCall(); replay(channelMock); endpoint.setReceiveTimeout(1); - endpoint.setMaxMessagesPerPoll(1); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(1); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.start(); trigger.await(); endpoint.stop(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/pollingEndpointErrorHandlingTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/pollingEndpointErrorHandlingTests.xml index f49f51b4e1..02917f66d6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/pollingEndpointErrorHandlingTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/pollingEndpointErrorHandlingTests.xml @@ -12,7 +12,15 @@ - + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java index 232d5f53d0..0610d5cc54 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInvokingMessageHandlerTests.java @@ -141,13 +141,6 @@ public class GatewayInvokingMessageHandlerTests { @Test public void validateGatewayWithErrorAsyncAndMaper() { - try { - gatewayWithErrorAsync.sendRecieve("echoWithErrorAsyncChannel"); - Assert.fail(); - } catch (Exception e) { - Assert.assertTrue(e instanceof MessageHandlingException); - } - try { Object result = gatewayWithErrorAsyncAndMapper.sendRecieve("echoWithErrorAsyncChannel"); Assert.assertEquals("Error happened in message: echoWithErrorAsyncChannel", result); diff --git a/spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java b/spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java index f53ae6cc48..3dc0b03408 100644 --- a/spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java +++ b/spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java @@ -32,6 +32,8 @@ import org.junit.Test; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.scheduling.PollerFactory; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -93,7 +95,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskSameAsMessageCount() { - endpoint.setMaxMessagesPerPoll(3); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(3); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); @@ -108,7 +112,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskLessThanMessageCount() { - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); @@ -122,7 +128,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskExceedsMessageCount() { - endpoint.setMaxMessagesPerPoll(5); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(5); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.setReceiveTimeout(0); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -137,7 +145,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() { - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.setReceiveTimeout(0); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -160,7 +170,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() { - endpoint.setMaxMessagesPerPoll(5); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(5); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.setReceiveTimeout(0); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -182,7 +194,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testStreamResetBetweenDispatches() { - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.setReceiveTimeout(0); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); @@ -204,7 +218,9 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testStreamWriteBetweenDispatches() throws IOException { - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.setReceiveTimeout(0); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); diff --git a/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java b/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java index 45c24ad282..53b8231b40 100644 --- a/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java +++ b/spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java @@ -31,6 +31,8 @@ import org.junit.Test; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.scheduling.PollerFactory; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -81,7 +83,9 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void twoStringsAndNoNewLinesByDefault() { - endpoint.setMaxMessagesPerPoll(1); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(1); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage("foo"), 0); channel.send(new GenericMessage("bar"), 0); endpoint.start(); @@ -98,8 +102,9 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void twoStringsWithNewLines() { handler.setShouldAppendNewLine(true); - endpoint.setMaxMessagesPerPoll(1); - channel.send(new GenericMessage("foo"), 0); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(1); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage("foo"), 0); channel.send(new GenericMessage("bar"), 0); endpoint.start(); trigger.await(); @@ -115,7 +120,9 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskSameAsMessageCount() { - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); channel.send(new GenericMessage("foo"), 0); channel.send(new GenericMessage("bar"), 0); endpoint.start(); @@ -126,7 +133,9 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() { - endpoint.setMaxMessagesPerPoll(10); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(10); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); endpoint.setReceiveTimeout(0); handler.setShouldAppendNewLine(true); channel.send(new GenericMessage("foo"), 0); @@ -140,7 +149,9 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void singleNonStringObject() { - endpoint.setMaxMessagesPerPoll(1); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(1); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); endpoint.start(); @@ -152,7 +163,9 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void twoNonStringObjectWithOutNewLines() { endpoint.setReceiveTimeout(0); - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); @@ -167,7 +180,9 @@ public class CharacterStreamWritingMessageHandlerTests { public void twoNonStringObjectWithNewLines() { handler.setShouldAppendNewLine(true); endpoint.setReceiveTimeout(0); - endpoint.setMaxMessagesPerPoll(2); + PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setMaxMessagesPerPoll(2); + endpoint.setPollerFactory(new PollerFactory(pollerMetadata)); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0);