From 176115af571aa53f62e65f1abcab26a46d9a2f55 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Sat, 25 Sep 2010 13:19:32 -0400 Subject: [PATCH] INT-1451 refactoring part-2 --- .../config/ConsumerEndpointFactoryBean.java | 10 +- ...ourcePollingChannelAdapterFactoryBean.java | 9 +- .../integration/config/xml/PollerParser.java | 42 ++++-- .../endpoint/AbstractPollingEndpoint.java | 137 ++++++++++++++---- .../endpoint/AsyncInvokerAdvice.java | 89 ------------ .../integration/endpoint/Poller.java | 76 ---------- .../integration/endpoint/PollerFactory.java | 113 --------------- .../integration/endpoint/PollingConsumer.java | 5 +- .../endpoint/SourcePollingChannelAdapter.java | 5 +- .../gateway/MessagingGatewaySupport.java | 4 +- .../scheduling/PollerMetadata.java | 42 ++++-- .../PollerTaskTransactionDecorator.java | 79 ---------- .../integration/util/ObjectDecorator.java | 30 ---- .../ApplicationContextMessageBusTests.java | 16 +- .../integration/bus/messageBusTests.xml | 35 +++-- ...PollingChannelAdapterFactoryBeanTests.java | 31 +++- .../core/MessagingTemplateTests.java | 9 +- .../dispatcher/PollingTransactionTests.java | 6 +- .../PollingConsumerEndpointTests.java | 26 +++- .../endpoint/PollingEndpointStub.java | 9 +- ...cerAndConsumerAutoStartupTests-context.xml | 4 +- .../pollingEndpointErrorHandlingTests.xml | 23 ++- .../gateway/MessagingGatewayTests.java | 1 - .../MethodInvokingMessageHandlerTests.java | 5 +- .../integration/test/util/TestUtils.java | 8 - .../ByteStreamWritingMessageHandlerTests.java | 20 +-- ...acterStreamWritingMessageHandlerTests.java | 21 +-- .../integration/test/util/TestUtils.java | 6 - 28 files changed, 296 insertions(+), 565 deletions(-) delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/AsyncInvokerAdvice.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/Poller.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerFactory.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java delete mode 100644 spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java 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 37f5d3a6bd..87f94f2636 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 @@ -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 { 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 3c6af7b3fd..957fbee0a5 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 @@ -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{ + + 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 pollingTask = new Callable() { - 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 adviceChain = this.pollerMetadata.getAdviceChain(); + if (!CollectionUtils.isEmpty(adviceChain)){ + for (Advice advice : adviceChain) { + proxyFactory.addAdvice(advice); + } + } + + return new Poller((Callable) 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 pollingTask; + + public Poller(Callable 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)); + } + } + } + } + }); + } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AsyncInvokerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AsyncInvokerAdvice.java deleted file mode 100644 index 18290f889f..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AsyncInvokerAdvice.java +++ /dev/null @@ -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 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/Poller.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/Poller.java deleted file mode 100644 index 2ab9ed2916..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/Poller.java +++ /dev/null @@ -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 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/endpoint/PollerFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerFactory.java deleted file mode 100644 index a2c82f2bfe..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollerFactory.java +++ /dev/null @@ -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 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/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 9de73d9168..d7bc584469 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java @@ -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(); 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 4f4cbd9d24..f5a8fb4288 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 @@ -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; } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index f5fcf1a190..865b85e650 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -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(); 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 6795003067..98576f46a1 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 @@ -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 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. + * + *

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; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java deleted file mode 100644 index 1dbf049a3b..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollerTaskTransactionDecorator.java +++ /dev/null @@ -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 pollingTask. - * - * @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; - } -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java deleted file mode 100644 index 2a8452f204..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/ObjectDecorator.java +++ /dev/null @@ -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); -} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java index 543c75a0e0..979f56fe28 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/ApplicationContextMessageBusTests.java @@ -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"))); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index 23a1ce9dfe..e0f4d2b28d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -5,29 +5,36 @@ http://www.springframework.org/schema/beans/spring-beans.xsd"> - + factory-method="createTaskScheduler"> + - + - + - - - - - - + + + + + + + + + + - - - + + + - + 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 bb12239bd3..88d879477b 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 @@ -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()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java index 5f85a73b92..571605c360 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/core/MessagingTemplateTests.java @@ -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(); } 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 8fef13f285..9215d12399 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 @@ -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 adviceChain = TestUtils.getPropertyValue(pollerMetedata, "adviceChain",List.class); assertEquals(2, adviceChain.size()); Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class); 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 9701ec7bdd..82db074094 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 @@ -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(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java index 831c9b9a5c..b56ae2ac43 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingEndpointStub.java @@ -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"); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml index 461183bc0a..6f6932049f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ProducerAndConsumerAutoStartupTests-context.xml @@ -8,9 +8,7 @@ http://www.springframework.org/schema/integration/spring-integration-2.0.xsd"> - - - + 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 391c1c56fa..ec67ab78ed 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 @@ -1,28 +1,25 @@ + http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd"> - + - - - - - - - - + + + + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java b/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java index 79c19dfe3b..459a3e0906 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/gateway/MessagingGatewayTests.java @@ -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; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/MethodInvokingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/MethodInvokingMessageHandlerTests.java index 0bf4bed39e..047824541c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/message/MethodInvokingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/MethodInvokingMessageHandlerTests.java @@ -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); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/test/util/TestUtils.java b/spring-integration-core/src/test/java/org/springframework/integration/test/util/TestUtils.java index 64ba517942..02a76bdda9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/test/util/TestUtils.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/test/util/TestUtils.java @@ -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); } } 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 2d17debbcb..7344ce2f5b 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 @@ -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(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); @@ -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(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); @@ -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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(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(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 f28615222c..dc93d998de 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 @@ -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("foo"), 0); channel.send(new GenericMessage("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("foo"), 0); + endpoint.setPollerMetadata(pollerMetadata); + channel.send(new GenericMessage("foo"), 0); channel.send(new GenericMessage("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("foo"), 0); channel.send(new GenericMessage("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("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)); 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(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(testObject1), 0); diff --git a/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java b/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java index 51dd2b440f..c770cae5b3 100644 --- a/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java +++ b/spring-integration-test/src/main/java/org/springframework/integration/test/util/TestUtils.java @@ -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); } }