INT-3921: SimplePollSkipAdvice

JIRA: https://jira.spring.io/browse/INT-3921

Polishing - PR Comments

Also increased some timeouts in JsonPathTests (and fixed a loop counter).

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;
+	}
+
+}

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<Object>() {
+		class LocalSource implements MessageSource<Object> {
+
+			private final CountDownLatch latch;
+
+			public LocalSource(CountDownLatch latch) {
+				this.latch = latch;
+			}
 
 			@Override
 			public Message<Object> 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<Advice> adviceChain = new ArrayList<Advice>();
-		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<String>("@skipper.skipPolls()"));
+		assertTrue(this.skipper.skipPoll());
+		this.control.send(new GenericMessage<String>("@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();
+		}
+
+	}
+
 }

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<String> 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));

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 <<conditional-pollers>>.
 
 This chapter is meant to only give a high-level overview regarding Polling Consumers and how they fit into the concept of message channels - <<channel>> and channel adapters - <<channel-adapter>>.