INT-3914: Add CompoundTriggerAdvice

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

Author: garyrussell

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.
+ * <p>
+ * The poller advised by this class must be configured to use the same
+ * {@link CompoundTrigger} instance and must <b>not</b> 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;
+	}
+
+}

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

spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollerAdviceTests.java

@@ -18,9 +18,14 @@
 
 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 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 Message<Object> receive() {
 				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 Message<Object> receive() {
 		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<Object> overridePresent = new LinkedList<Object>();
+		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<Object>() {
+
+			@Override
+			public Message<Object> receive() {
+				overridePresent.add(TestUtils.getPropertyValue(compoundTrigger, "override"));
+				Message<Object> m = null;
+				if (latch.getCount() % 2 == 0) {
+					m = new GenericMessage<Object>("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 @@ private void configure(SourcePollingChannelAdapter adapter) {
 		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<Object> {
+
+		private final CountDownLatch latch = new CountDownLatch(5);
+
+		@Override
+		public Message<Object> receive() {
+			latch.countDown();
+			return null;
+		}
+
+	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/endpoint/compound-trigger-context.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
+
+	<int:inbound-channel-adapter channel="nullChannel" auto-startup="false">
+		<bean class="org.springframework.integration.endpoint.PollerAdviceTests.Source" />
+		<int:poller trigger="compoundTrigger">
+			<int:advice-chain>
+				<bean class="org.springframework.integration.aop.CompoundTriggerAdvice">
+					<constructor-arg ref="compoundTrigger"/>
+					<constructor-arg ref="secondary"/>
+				</bean>
+			</int:advice-chain>
+		</int:poller>
+	</int:inbound-channel-adapter>
+
+	<bean id="compoundTrigger" class="org.springframework.integration.util.CompoundTrigger">
+		<constructor-arg ref="primary" />
+	</bean>
+
+	<bean id="primary" class="org.springframework.scheduling.support.CronTrigger">
+		<constructor-arg value="*/1 * * * * *" />
+	</bean>
+
+	<bean id="secondary" class="org.springframework.scheduling.support.PeriodicTrigger">
+		<constructor-arg value="10" />
+	</bean>
+
+</beans>

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]
+----
+<int:inbound-channel-adapter channel="nullChannel" auto-startup="false">
+    <bean class="org.springframework.integration.endpoint.PollerAdviceTests.Source" />
+    <int:poller trigger="compoundTrigger">
+        <int:advice-chain>
+            <bean class="org.springframework.integration.aop.CompoundTriggerAdvice">
+                <constructor-arg ref="compoundTrigger"/>
+                <constructor-arg ref="secondary"/>
+            </bean>
+        </int:advice-chain>
+    </int:poller>
+</int:inbound-channel-adapter>
+
+<bean id="compoundTrigger" class="org.springframework.integration.util.CompoundTrigger">
+    <constructor-arg ref="primary" />
+</bean>
+
+<bean id="primary" class="org.springframework.scheduling.support.CronTrigger">
+    <constructor-arg value="0 0 * * * *" /> <!-- top of every hour -->
+</bean>
+
+<bean id="secondary" class="org.springframework.scheduling.support.PeriodicTrigger">
+    <constructor-arg value="60000" />
+</bean>
+----
+
+.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`.