diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimplePollSkipStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimplePollSkipStrategy.java new file mode 100644 index 0000000000..fb925be8a2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/SimplePollSkipStrategy.java @@ -0,0 +1,56 @@ +/* + * Copyright 2015 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 org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; + +/** + * A simple {@link PollSkipStrategy} to be used with a {@code PollSkipAdvice}. + * Invoke {@link #skipPolls()} to start skipping polls; invoke {@link #reset()} + * to resume polling. + * + * @author Gary Russell + * @since 4.2.5 + * + */ +@ManagedResource +public class SimplePollSkipStrategy implements PollSkipStrategy { + + private volatile boolean skip; + + @Override + public boolean skipPoll() { + return this.skip; + } + + /** + * Skip future polls. + */ + @ManagedOperation + public void skipPolls() { + this.skip = true; + } + + /** + * Resume polling at the next {@code Trigger} event. + */ + @ManagedOperation + public void reset() { + this.skip = false; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java index 0bd00e4f8a..3158af9c6a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java @@ -39,36 +39,57 @@ import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.junit.Test; +import org.junit.runner.RunWith; import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.aop.AbstractMessageSourceAdvice; import org.springframework.integration.aop.CompoundTriggerAdvice; import org.springframework.integration.aop.SimpleActiveIdleMessageSourceAdvice; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.ExpressionControlBusFactoryBean; import org.springframework.integration.core.MessageSource; import org.springframework.integration.scheduling.PollSkipAdvice; -import org.springframework.integration.scheduling.PollSkipStrategy; +import org.springframework.integration.scheduling.SimplePollSkipStrategy; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.CompoundTrigger; import org.springframework.integration.util.DynamicPeriodicTrigger; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.scheduling.support.PeriodicTrigger; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell * @since 4.1 * */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +@DirtiesContext public class PollerAdviceTests { public Message receiveAdviceResult; + @Autowired + private MessageChannel control; + + @Autowired + private SimplePollSkipStrategy skipper; + @Test public void testDefaultDontSkip() throws Exception { SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); @@ -104,18 +125,26 @@ public class PollerAdviceTests { } @Test - public void testSkipAll() throws Exception { + public void testSkipSimple() throws Exception { SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); - final CountDownLatch latch = new CountDownLatch(1); - adapter.setSource(new MessageSource() { + class LocalSource implements MessageSource { + + private final CountDownLatch latch; + + public LocalSource(CountDownLatch latch) { + this.latch = latch; + } @Override public Message receive() { latch.countDown(); return null; } - }); - adapter.setTrigger(new Trigger() { + + }; + CountDownLatch latch = new CountDownLatch(1); + adapter.setSource(new LocalSource(latch)); + class OneAndDone10msTrigger implements Trigger { private boolean done; @@ -125,23 +154,34 @@ public class PollerAdviceTests { done = true; return date; } - }); + }; + adapter.setTrigger(new OneAndDone10msTrigger()); configure(adapter); List adviceChain = new ArrayList(); - PollSkipAdvice advice = new PollSkipAdvice(new PollSkipStrategy() { - - @Override - public boolean skipPoll() { - return true; - } - - }); + SimplePollSkipStrategy skipper = new SimplePollSkipStrategy(); + skipper.skipPolls(); + PollSkipAdvice advice = new PollSkipAdvice(skipper); adviceChain.add(advice); adapter.setAdviceChain(adviceChain); adapter.afterPropertiesSet(); adapter.start(); assertFalse(latch.await(1, TimeUnit.SECONDS)); adapter.stop(); + skipper.reset(); + latch = new CountDownLatch(1); + adapter.setSource(new LocalSource(latch)); + adapter.setTrigger(new OneAndDone10msTrigger()); + adapter.start(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + adapter.stop(); + } + + @Test + public void testSkipSimpleControlBus() { + this.control.send(new GenericMessage("@skipper.skipPolls()")); + assertTrue(this.skipper.skipPoll()); + this.control.send(new GenericMessage("@skipper.reset()")); + assertFalse(this.skipper.skipPoll()); } @Test @@ -314,4 +354,26 @@ public class PollerAdviceTests { } + @Configuration + @EnableIntegration + public static class Config { + + @Bean + public SimplePollSkipStrategy skipper() { + return new SimplePollSkipStrategy(); + } + + @Bean + public MessageChannel control() { + return new DirectChannel(); + } + + @Bean + @ServiceActivator(inputChannel = "control") + public ExpressionControlBusFactoryBean controlBus() { + return new ExpressionControlBusFactoryBean(); + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonPathTests.java b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonPathTests.java index 5eee0db750..f894596840 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/json/JsonPathTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/json/JsonPathTests.java @@ -119,7 +119,7 @@ public class JsonPathTests { @Test public void testInt3139JsonPathTransformer() throws IOException { this.transformerInput.send(testMessage); - Message receive = this.output.receive(1000); + Message receive = this.output.receive(10000); assertNotNull(receive); assertEquals("Nigel Rees", receive.getPayload()); @@ -142,23 +142,23 @@ public class JsonPathTests { @Test public void testInt3139JsonPathFilter() { this.filterInput1.send(testMessage); - Message receive = this.output.receive(1000); + Message receive = this.output.receive(10000); assertNotNull(receive); assertEquals(JSON, receive.getPayload()); this.filterInput2.send(testMessage); - receive = this.output.receive(1000); + receive = this.output.receive(10000); assertNotNull(receive); Message message = MessageBuilder.withPayload(JSON) .setHeader("price", 10) .build(); this.filterInput3.send(message); - receive = this.output.receive(1000); + receive = this.output.receive(10000); assertNotNull(receive); this.filterInput4.send(testMessage); - receive = this.output.receive(1000); + receive = this.output.receive(10000); assertNotNull(receive); try { @@ -171,7 +171,7 @@ public class JsonPathTests { receive = this.output.receive(0); assertNull(receive); - receive = this.discardChannel.receive(1000); + receive = this.discardChannel.receive(10000); assertNotNull(receive); } @@ -179,8 +179,8 @@ public class JsonPathTests { @Test public void testInt3139JsonPathSplitter() { this.splitterInput.send(testMessage); - for(int i = 0; i < 3; i++) { - Message receive = this.splitterOutput.receive(1000); + for(int i = 0; i < 4; i++) { + Message receive = this.splitterOutput.receive(10000); assertNotNull(receive); assertTrue(receive.getPayload() instanceof Map); } @@ -192,7 +192,7 @@ public class JsonPathTests { .setHeader("jsonPath", "$.store.book[0].category") .build(); this.routerInput.send(message); - Message receive = this.routerOutput1.receive(1000); + Message receive = this.routerOutput1.receive(10000); assertNotNull(receive); assertEquals(JSON, receive.getPayload()); assertNull(this.routerOutput2.receive(10)); @@ -201,7 +201,7 @@ public class JsonPathTests { .setHeader("jsonPath", "$.store.book[2].category") .build(); this.routerInput.send(message); - receive = this.routerOutput2.receive(1000); + receive = this.routerOutput2.receive(10000); assertNotNull(receive); assertEquals(JSON, receive.getPayload()); assertNull(this.routerOutput1.receive(10)); diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index a42dae01c3..ff0bdf9189 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -50,6 +50,10 @@ Starting with _version 4.1_ a `PollSkipAdvice` is provided. Pollers use triggers to determine the time of the next poll. The `PollSkipAdvice` can be used to suppress (skip) a poll, perhaps because there is some downstream condition that would prevent the message to be processed properly. To use this advice, you have to provide it with an implementation of a `PollSkipStrategy`. +Startng with version 4.2.5, a `SimplePollSkipStrategy` is provided. +Add an instance as a bean to the application context, inject it into a `PollSkipAdvice` and add that to the poller's +advice chain. +To skip polling, call `skipPolls()`, to resume polling, call `reset()`. _Version 4.2_ added more flexibility in this area - see <>. This chapter is meant to only give a high-level overview regarding Polling Consumers and how they fit into the concept of message channels - <> and channel adapters - <>.