diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java new file mode 100644 index 0000000000..4d14608fdf --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java @@ -0,0 +1,65 @@ +/* + * 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.aop; + +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.util.CompoundTrigger; +import org.springframework.messaging.Message; +import org.springframework.scheduling.Trigger; +import org.springframework.util.Assert; + +/** + * An {@link AbstractMessageSourceAdvice} that uses a {@link CompoundTrigger} to adjust + * the poller - when a message is present, the compound trigger's primary trigger is + * used to determine the next poll. When no message is present, the override trigger is + * used. + *

+ * The poller advised by this class must be configured to use the same + * {@link CompoundTrigger} instance and must not use a task executor. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class CompoundTriggerAdvice extends AbstractMessageSourceAdvice { + + private final CompoundTrigger compoundTrigger; + + private final Trigger override; + + public CompoundTriggerAdvice(CompoundTrigger compoundTrigger, Trigger overrideTrigger) { + Assert.notNull(compoundTrigger, "'compoundTrigger' cannot be null"); + this.compoundTrigger = compoundTrigger; + this.override = overrideTrigger; + } + + @Override + public boolean beforeReceive(MessageSource source) { + return true; + } + + @Override + public Message afterReceive(Message result, MessageSource source) { + if (result == null) { + this.compoundTrigger.setOverride(this.override); + } + else { + this.compoundTrigger.setOverride(null); + } + return result; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/CompoundTrigger.java b/spring-integration-core/src/main/java/org/springframework/integration/util/CompoundTrigger.java new file mode 100644 index 0000000000..f35bce3859 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/CompoundTrigger.java @@ -0,0 +1,77 @@ +/* + * 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.util; + +import java.util.Date; + +import org.springframework.scheduling.Trigger; +import org.springframework.scheduling.TriggerContext; + +import reactor.core.support.Assert; + +/** + * A {@link Trigger} that delegates the {@link #nextExecutionTime(TriggerContext)} + * to one of two Triggers. If the {@link #setOverride(Trigger) override} trigger is + * {@code null}, the primary trigger is invoked; otherwise the override trigger is + * invoked. + * + * @author Gary Russell + * @since 4.3 + * + */ +public class CompoundTrigger implements Trigger { + + private volatile Trigger primary; + + private volatile Trigger override; + + /** + * Construct a compound trigger with the supplied primary trigger. + * @param primary the primary trigger. + */ + public CompoundTrigger(Trigger primary) { + setPrimary(primary); + } + + /** + * Set the primary trigger. + * @param primary the trigger. + */ + public final void setPrimary(Trigger primary) { + Assert.notNull(primary, "'primary' cannot be null"); + this.primary = primary; + } + + /** + * Set the override trigger; set to null to revert to using the + * primary trigger. + * @param override the override trigger, or null. + */ + public void setOverride(Trigger override) { + this.override = override; + } + + @Override + public Date nextExecutionTime(TriggerContext triggerContext) { + if (this.override != null) { + return this.override.nextExecutionTime(triggerContext); + } + else { + return this.primary.nextExecutionTime(triggerContext); + } + } + +} 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 66619c34c2..0bd00e4f8a 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 @@ -18,9 +18,14 @@ package org.springframework.integration.endpoint; import static org.hamcrest.Matchers.contains; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.atLeast; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; import java.util.ArrayList; import java.util.Collections; @@ -33,23 +38,27 @@ import java.util.concurrent.TimeUnit; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; -import org.hamcrest.Matchers; import org.junit.Test; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aop.AbstractMessageSourceAdvice; +import org.springframework.integration.aop.CompoundTriggerAdvice; import org.springframework.integration.aop.SimpleActiveIdleMessageSourceAdvice; import org.springframework.integration.channel.NullChannel; -import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageSource; import org.springframework.integration.scheduling.PollSkipAdvice; import org.springframework.integration.scheduling.PollSkipStrategy; +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.support.GenericMessage; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.TriggerContext; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.scheduling.support.PeriodicTrigger; /** * @author Gary Russell @@ -221,11 +230,11 @@ public class PollerAdviceTests { return m; } }); - QueueChannel channel = new QueueChannel(); SimpleActiveIdleMessageSourceAdvice toggling = new SimpleActiveIdleMessageSourceAdvice(trigger); toggling.setActivePollPeriod(11); toggling.setIdlePollPeriod(12); adapter.setAdviceChain(Collections.singletonList(toggling)); + adapter.setTrigger(trigger); configure(adapter); adapter.afterPropertiesSet(); adapter.start(); @@ -234,7 +243,42 @@ public class PollerAdviceTests { while (triggerPeriods.size() > 5) { triggerPeriods.removeLast(); } - assertThat(triggerPeriods, Matchers.contains(10L, 12L, 11L, 12L, 11L)); + assertThat(triggerPeriods, contains(10L, 12L, 11L, 12L, 11L)); + } + + @Test + public void testCompoundTriggerAdvice() throws Exception { + SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); + final CountDownLatch latch = new CountDownLatch(5); + final LinkedList overridePresent = new LinkedList(); + final CompoundTrigger compoundTrigger = new CompoundTrigger(new PeriodicTrigger(10)); + Trigger override = spy(new PeriodicTrigger(5)); + final CompoundTriggerAdvice advice = new CompoundTriggerAdvice(compoundTrigger, override); + adapter.setSource(new MessageSource() { + + @Override + public Message receive() { + overridePresent.add(TestUtils.getPropertyValue(compoundTrigger, "override")); + Message m = null; + if (latch.getCount() % 2 == 0) { + m = new GenericMessage("foo"); + } + latch.countDown(); + return m; + } + }); + adapter.setAdviceChain(Collections.singletonList(advice)); + adapter.setTrigger(compoundTrigger); + configure(adapter); + adapter.afterPropertiesSet(); + adapter.start(); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + adapter.stop(); + while (overridePresent.size() > 5) { + overridePresent.removeLast(); + } + assertThat(overridePresent, contains(null, override, null, override, null)); + verify(override, atLeast(2)).nextExecutionTime(any(TriggerContext.class)); } private void configure(SourcePollingChannelAdapter adapter) { @@ -245,4 +289,29 @@ public class PollerAdviceTests { adapter.setTaskScheduler(scheduler); } + @Test + public void testCompoundAdviceXML() throws Exception { + ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext("compound-trigger-context.xml", + getClass()); + SourcePollingChannelAdapter adapter = ctx.getBean(SourcePollingChannelAdapter.class); + Source source = ctx.getBean(Source.class); + adapter.start(); + assertTrue(source.latch.await(10, TimeUnit.SECONDS)); + assertNotNull(TestUtils.getPropertyValue(adapter, "trigger.override")); + adapter.stop(); + ctx.close(); + } + + public static class Source implements MessageSource { + + private final CountDownLatch latch = new CountDownLatch(5); + + @Override + public Message receive() { + latch.countDown(); + return null; + } + + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/compound-trigger-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/compound-trigger-context.xml new file mode 100644 index 0000000000..d0d01cdc52 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/compound-trigger-context.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index 83fa85e06a..a42dae01c3 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -84,6 +84,16 @@ Message afterReceive(Message result, MessageSource source) This method is called after the `receive()` method; again, you can reconfigure the source, or take any action perhaps depending on the result (which can be `null` if there was no message created by the source). You can even return a different message! +.Advice Chain Ordering +[IMPORTANT] +===== +It is important to understand how the advice chain is processed during initialization. +`Advice` objects that do not extend `AbstractMessageSourceAdvice` are applied to the whole poll process and are all invoked first, in order, before any `AbstractMessageSourceAdvice`; then `AbstractMessageSourceAdvice` objects are invoked in order around the `MessageSource` `receive()` method. +If you have, say `Advice` objects `a, b, c, d`, where `b` and `d` are `AbstractMessageSourceAdvice`, they will be applied in the order `a, c, b, d`. +Also, if a `MessageSource` is already a `Proxy`, the `AbstractMessageSourceAdvice` will be invoked after any existing `Advice` objects. +If you wish to change the order, you should wire up the proxy yourself. +===== + ===== SimpleActiveIdleMessageSourceAdvice This advice is a simple implementation of `AbstractMessageSourceAdvice`, when used in conjunction with a `DynamicPeriodicTrigger`, it adjuststhe polling frequency depending on whether or not the previous poll resulted in a message or not. @@ -95,12 +105,52 @@ This will only work if the advice is called on the poller thread. It will *not* work if the poller has a `task-executor`. To use this advice where you wish to use async operations after the result of a poll, do the async handoff later, perhaps by using an `ExecutorChannel`. -.Advice Chain Ordering -[IMPORTANT] -===== -It is important to understand how the advice chain is processed during initialization. -`Advice` objects that do not extend `AbstractMessageSourceAdvice` are applied to the whole poll process and are all invoked first, in order, before any `AbstractMessageSourceAdvice`; then `AbstractMessageSourceAdvice` objects are invoked in order around the `MessageSource` `receive()` method. -If you have, say `Advice` objects `a, b, c, d`, where `b` and `d` are `AbstractMessageSourceAdvice`, they will be applied in the order `a, c, b, d`. -Also, if a `MessageSource` is already a `Proxy`, the `AbstractMessageSourceAdvice` will be invoked after any existing `Advice` objects. -If you wish to change the order, you should wire up the proxy yourself. -===== +===== CompoundTriggerAdvice + +This advice allows the selection of one of two triggers based on whether a poll returns a message or not. +Consider a poller that uses a `CronTrigger`; `CronTrigger` s are immutable so cannot be altered once constructed. +Consider a use case where we want to use a cron expression to trigger a poll once each hour but, if no message is +received, poll once per minute and, when a message is retrieved, revert to using the cron expression. + +The advice (and poller) use a `CompoundTrigger` for this purpose. +The trigger's `primary` trigger can be a `CronTrigger`. +When the advice detects that no message is received, it adds the secondary trigger to the `CompoundTrigger`. +When the `CompoundTrigger` 's `nextExecutionTime` method is invoked, it will delegate to the secondary trigger, if +present; otherwise the primary trigger. + +The poller must also have a reference to the same `CompoundTrigger`. + +The following shows the configuration for the hourly cron expression with fall-back to every minute... + +[source, xml] +---- + + + + + + + + + + + + + + + + + + + + + + + +---- + +.Important: Async Handoff +IMPORTANT: This advice modifies the trigger based on the `receive()` result. +This will only work if the advice is called on the poller thread. +It will *not* work if the poller has a `task-executor`. +To use this advice where you wish to use async operations after the result of a poll, do the async handoff later, perhaps by using an `ExecutorChannel`.