From 176afcf8076f218679b1b0d70b8264544fede6d5 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 21 Oct 2014 16:30:08 +0300 Subject: [PATCH] INT-3519 Add Poll Skip Advice JIRA: https://jira.spring.io/browse/INT-3519 Add a mechanism to allow skipping polls, perhaps because of some downstream condition. INT-3519: Polishing --- .../scheduling/PollSkipAdvice.java | 74 +++++++++++ .../scheduling/PollSkipStrategy.java | 34 +++++ .../endpoint/PollerAdviceTests.java | 124 ++++++++++++++++++ src/reference/docbook/polling-consumer.xml | 9 ++ src/reference/docbook/whats-new.xml | 10 ++ 5 files changed, 251 insertions(+) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipStrategy.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java new file mode 100644 index 0000000000..45e3850864 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipAdvice.java @@ -0,0 +1,74 @@ +/* + * Copyright 2014 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.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +/** + * An advice that can be added to a poller's advice chain that determines + * whether a poll should be skipped or not. May be used to temporarily suspend + * polling when some downstream condition exists in the flow. + * + * @author Gary Russell + * @since 4.1 + * + */ +public class PollSkipAdvice implements MethodInterceptor { + + private static final Log logger = LogFactory.getLog(PollSkipAdvice.class); + + private final PollSkipStrategy pollSkipStrategy; + + public PollSkipAdvice() { + this(new DefaultPollSkipStrategy()); + } + + + public PollSkipAdvice(PollSkipStrategy strategy) { + this.pollSkipStrategy = strategy; + } + + + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + if ("call".equals(invocation.getMethod().getName()) && this.pollSkipStrategy.skipPoll()) { + if (logger.isDebugEnabled()) { + logger.debug("Skipping poll because " + + this.pollSkipStrategy.getClass().getName() + + ".skipPoll() returned true"); + } + return Boolean.FALSE; + } + else { + return invocation.proceed(); + } + } + + + private static class DefaultPollSkipStrategy implements PollSkipStrategy { + + @Override + public boolean skipPoll() { + return false; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipStrategy.java new file mode 100644 index 0000000000..82e9bf7556 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollSkipStrategy.java @@ -0,0 +1,34 @@ +/* + * Copyright 2014 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; + +/** + * Implementations determine whether a particular poll should be skipped. + * + * @author Gary Russell + * @since 4.1 + * + */ +public interface PollSkipStrategy { + + /** + * Return true if this poll should be skipped. + * @return true to skip. + */ + boolean skipPoll(); + +} 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 new file mode 100644 index 0000000000..5068457183 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java @@ -0,0 +1,124 @@ +/* + * Copyright 2014 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 static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; + +import java.util.ArrayList; +import java.util.Date; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.aopalliance.aop.Advice; +import org.junit.Test; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.scheduling.PollSkipAdvice; +import org.springframework.integration.scheduling.PollSkipStrategy; +import org.springframework.messaging.Message; +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; + +/** + * @author Gary Russell + * @since 4.1 + * + */ +public class PollerAdviceTests { + + @Test + public void testDefaultDontSkip() throws Exception { + SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); + final CountDownLatch latch = new CountDownLatch(1); + adapter.setSource(new MessageSource() { + + @Override + public Message receive() { + latch.countDown(); + return null; + } + }); + adapter.setTrigger(new Trigger() { + + @Override + public Date nextExecutionTime(TriggerContext triggerContext) { + return new Date(System.currentTimeMillis() + 10); + } + }); + configure(adapter); + List adviceChain = new ArrayList(); + PollSkipAdvice advice = new PollSkipAdvice(); + adviceChain.add(advice); + adapter.setAdviceChain(adviceChain); + adapter.afterPropertiesSet(); + adapter.start(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + adapter.stop(); + } + + @Test + public void testSkipAll() throws Exception { + SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); + final CountDownLatch latch = new CountDownLatch(1); + adapter.setSource(new MessageSource() { + + @Override + public Message receive() { + latch.countDown(); + return null; + } + }); + adapter.setTrigger(new Trigger() { + + @Override + public Date nextExecutionTime(TriggerContext triggerContext) { + return new Date(System.currentTimeMillis() + 10); + } + }); + configure(adapter); + List adviceChain = new ArrayList(); + PollSkipAdvice advice = new PollSkipAdvice(new PollSkipStrategy() { + + @Override + public boolean skipPoll() { + return true; + } + + }); + adviceChain.add(advice); + adapter.setAdviceChain(adviceChain); + adapter.afterPropertiesSet(); + adapter.start(); + assertFalse(latch.await(1, TimeUnit.SECONDS)); + adapter.stop(); + } + + private void configure(SourcePollingChannelAdapter adapter) { + adapter.setOutputChannel(new NullChannel()); + adapter.setBeanFactory(mock(BeanFactory.class)); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); + scheduler.afterPropertiesSet(); + adapter.setTaskScheduler(scheduler); + } + +} diff --git a/src/reference/docbook/polling-consumer.xml b/src/reference/docbook/polling-consumer.xml index ee1e68499a..176349351b 100644 --- a/src/reference/docbook/polling-consumer.xml +++ b/src/reference/docbook/polling-consumer.xml @@ -59,6 +59,15 @@ Polling internal (pollable) Message Channels Polling internal services (E.g. repeatedly execute methods on a Java class) + + AOP Advice classes can be applied to pollers, in an advice-chain. An example being a + transaction advice to start a transaction. 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. + 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 diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index e947fc4eea..d0bd1573cb 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -34,6 +34,16 @@ for the JSON transformers. See for more information. +
+ PollSkipAdvice + + The PollSkipAdvice is now provided to be used within + <advice-chain> of the <poller> to determine if the current + poll should be suppressed (skipped) by some condition implemented with + PollSkipStrategy. + See for more information. + +
General Changes