diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index b0c68dac43..06e0f04717 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java @@ -97,23 +97,27 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement @SuppressWarnings("unchecked") private Runnable createPoller() throws Exception{ - ProxyFactory proxyFactory = new ProxyFactory(this); - - // Add Transaction advice first + Callable pollingTask = this; Advisor transactionAdvice = this.pollerMetadata.getTransactionAdvisor(); - if (transactionAdvice != null){ - proxyFactory.addAdvisor(transactionAdvice); - } - - // . . .then add the rest of the advises List adviceChain = this.pollerMetadata.getAdviceChain(); - if (!CollectionUtils.isEmpty(adviceChain)){ - for (Advice advice : adviceChain) { - proxyFactory.addAdvice(advice); + if (transactionAdvice != null || !CollectionUtils.isEmpty(adviceChain)){ + ProxyFactory proxyFactory = new ProxyFactory(this); + + // Add Transaction advice first + if (transactionAdvice != null){ + proxyFactory.addAdvisor(transactionAdvice); } + + // . . .then add the rest of the advises + if (!CollectionUtils.isEmpty(adviceChain)){ + for (Advice advice : adviceChain) { + proxyFactory.addAdvice(advice); + } + } + pollingTask = (Callable) proxyFactory.getProxy(this.beanClassLoader); } - return new Poller((Callable) proxyFactory.getProxy(this.beanClassLoader)); + return new Poller(pollingTask); } // LifecycleSupport implementation @@ -134,6 +138,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement this.runningTask.cancel(true); } this.runningTask = null; + this.initialized = false; } public void setPollerMetadata(PollerMetadata pollerMetadata) { @@ -167,7 +172,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement int count = 0; while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) { try { - if (!pollingTask.call()){ + boolean b = pollingTask.call(); + if (!b){ break; } count++; diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java index df2253b3c0..2dcfb6c44a 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java @@ -32,6 +32,7 @@ import org.junit.Before; import org.junit.Test; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; +import org.springframework.beans.factory.support.GenericBeanDefinition; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.Message; @@ -45,6 +46,7 @@ import org.springframework.integration.monitor.IntegrationMBeanExporter; import org.springframework.integration.monitor.LifecycleMessageHandlerMonitor; import org.springframework.integration.monitor.QueueChannelMonitor; import org.springframework.integration.monitor.DirectChannelMonitor; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.jmx.support.MBeanServerFactoryBean; import org.springframework.jmx.support.ObjectNameManager; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -146,7 +148,9 @@ public class ControlBusTests { RootBeanDefinition endpointDef = new RootBeanDefinition(PollingConsumer.class); endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel")); endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(BridgeHandler.class)); - endpointDef.getPropertyValues().add("trigger", new PeriodicTrigger(10000)); + RootBeanDefinition pollerMetaDefinition = new RootBeanDefinition(PollerMetadata.class); + pollerMetaDefinition.getPropertyValues().add("trigger", new PeriodicTrigger(10000)); + endpointDef.getPropertyValues().add("pollerMetadata", pollerMetaDefinition); context.registerBeanDefinition("pollingConsumer", endpointDef); context.registerBeanDefinition("taskScheduler", new RootBeanDefinition(ThreadPoolTaskScheduler.class)); registerControlBus(context, "domain.test4"); diff --git a/spring-integration-stream/pom.xml b/spring-integration-stream/pom.xml index ef26380908..28aad83d50 100644 --- a/spring-integration-stream/pom.xml +++ b/spring-integration-stream/pom.xml @@ -29,5 +29,9 @@ org.springframework spring-test + + org.mockito + mockito-all + 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 7344ce2f5b..7a78b4daf9 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 @@ -17,6 +17,7 @@ package org.springframework.integration.stream; import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; import java.io.ByteArrayOutputStream; import java.io.IOException; @@ -28,6 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; @@ -67,6 +69,7 @@ public class ByteStreamWritingMessageHandlerTests { PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setTrigger(trigger); endpoint.setPollerMetadata(pollerMetadata); + endpoint.setBeanFactory(mock(BeanFactory.class)); } @After @@ -96,6 +99,7 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskSameAsMessageCount() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(3); endpoint.setPollerMetadata(pollerMetadata); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -113,6 +117,7 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskLessThanMessageCount() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(2); endpoint.setPollerMetadata(pollerMetadata); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -129,6 +134,7 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskExceedsMessageCount() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(5); endpoint.setPollerMetadata(pollerMetadata); endpoint.setReceiveTimeout(0); @@ -146,6 +152,7 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(2); endpoint.setPollerMetadata(pollerMetadata); endpoint.setReceiveTimeout(0); @@ -171,6 +178,7 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(5); endpoint.setPollerMetadata(pollerMetadata); endpoint.setReceiveTimeout(0); @@ -196,6 +204,7 @@ public class ByteStreamWritingMessageHandlerTests { public void testStreamResetBetweenDispatches() { PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setMaxMessagesPerPoll(2); + pollerMetadata.setTrigger(trigger); endpoint.setPollerMetadata(pollerMetadata); endpoint.setReceiveTimeout(0); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -219,6 +228,7 @@ public class ByteStreamWritingMessageHandlerTests { @Test public void testStreamWriteBetweenDispatches() throws IOException { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(2); endpoint.setPollerMetadata(pollerMetadata); endpoint.setReceiveTimeout(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 dc93d998de..7e2a0be5af 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 @@ -17,6 +17,7 @@ package org.springframework.integration.stream; import static org.junit.Assert.assertEquals; +import static org.mockito.Mockito.mock; import java.io.StringWriter; import java.util.Date; @@ -27,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.message.GenericMessage; @@ -67,6 +69,7 @@ public class CharacterStreamWritingMessageHandlerTests { PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setTrigger(trigger); endpoint.setPollerMetadata(pollerMetadata); + endpoint.setBeanFactory(mock(BeanFactory.class)); } @After @@ -85,6 +88,7 @@ public class CharacterStreamWritingMessageHandlerTests { public void twoStringsAndNoNewLinesByDefault() { PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setMaxMessagesPerPoll(1); + pollerMetadata.setTrigger(trigger); endpoint.setPollerMetadata(pollerMetadata); channel.send(new GenericMessage("foo"), 0); channel.send(new GenericMessage("bar"), 0); @@ -103,6 +107,7 @@ public class CharacterStreamWritingMessageHandlerTests { public void twoStringsWithNewLines() { handler.setShouldAppendNewLine(true); PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(1); endpoint.setPollerMetadata(pollerMetadata); channel.send(new GenericMessage("foo"), 0); @@ -122,6 +127,7 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskSameAsMessageCount() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(2); endpoint.setPollerMetadata(pollerMetadata); channel.send(new GenericMessage("foo"), 0); @@ -135,6 +141,7 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(10); endpoint.setPollerMetadata(pollerMetadata); endpoint.setReceiveTimeout(0); @@ -151,6 +158,7 @@ public class CharacterStreamWritingMessageHandlerTests { @Test public void singleNonStringObject() { PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(1); endpoint.setPollerMetadata(pollerMetadata); TestObject testObject = new TestObject("foo"); @@ -165,6 +173,7 @@ public class CharacterStreamWritingMessageHandlerTests { public void twoNonStringObjectWithOutNewLines() { endpoint.setReceiveTimeout(0); PollerMetadata pollerMetadata = new PollerMetadata(); + pollerMetadata.setTrigger(trigger); pollerMetadata.setMaxMessagesPerPoll(2); endpoint.setPollerMetadata(pollerMetadata); TestObject testObject1 = new TestObject("foo"); @@ -183,6 +192,7 @@ public class CharacterStreamWritingMessageHandlerTests { endpoint.setReceiveTimeout(0); PollerMetadata pollerMetadata = new PollerMetadata(); pollerMetadata.setMaxMessagesPerPoll(2); + pollerMetadata.setTrigger(trigger); endpoint.setPollerMetadata(pollerMetadata); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); diff --git a/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java b/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java index 001b3bf1b6..80db088230 100644 --- a/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java +++ b/spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java @@ -27,6 +27,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.ws.MarshallingWebServiceOutboundGateway; import org.springframework.integration.ws.SimpleWebServiceOutboundGateway; import org.springframework.oxm.Marshaller; @@ -228,9 +229,11 @@ public class WebServiceOutboundGatewayParserTests { "simpleWebServiceOutboundGatewayParserTests.xml", this.getClass()); AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithPoller"); assertEquals(PollingConsumer.class, endpoint.getClass()); - Object obj = new DirectFieldAccessor(endpoint).getPropertyValue("trigger"); - assertEquals(PeriodicTrigger.class, obj.getClass()); - PeriodicTrigger trigger = (PeriodicTrigger) obj; + Object pollerMetadata = new DirectFieldAccessor(endpoint).getPropertyValue("pollerMetadata"); + assertEquals(PollerMetadata.class, pollerMetadata.getClass()); + Object triggerObject = new DirectFieldAccessor(pollerMetadata).getPropertyValue("trigger"); + assertEquals(PeriodicTrigger.class, triggerObject.getClass()); + PeriodicTrigger trigger = (PeriodicTrigger) triggerObject; DirectFieldAccessor accessor = new DirectFieldAccessor(trigger); accessor = new DirectFieldAccessor(trigger); assertEquals("PeriodicTrigger had wrong period",