INT-3633: Add MessageSourceAdvice

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

INT-3633: Add SimpleActiveIdleMessageSourceAdvice

Also resolve package tangle.

Polishing and Docs

Polishing according PR comments:
* Fix JavaDocs vulnerabilities
* Fix typos in docs
* Remove unnecessary `AopUtils.canApply` check

spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.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.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
+import org.springframework.integration.core.MessageSource;
+import org.springframework.messaging.Message;
+
+/**
+ * Advice for a {@link MessageSource#receive()} method to decide whether a poll
+ * should be ignored and/or take action after the receive.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ */
+public abstract class AbstractMessageSourceAdvice implements MethodInterceptor {
+
+	@Override
+	public final Object invoke(MethodInvocation invocation) throws Throwable {
+		Object target = invocation.getThis();
+		if (!(target instanceof MessageSource)
+				|| invocation.getMethod().getName() != "receive") {
+			return invocation.proceed();
+		}
+
+		Message<?> result = null;
+		if (beforeReceive((MessageSource<?>) target)) {
+			result = (Message<?>) invocation.proceed();
+		}
+		return afterReceive(result, (MessageSource<?>) target);
+	}
+
+	/**
+	 * Subclasses can decide whether to proceed with this poll.
+	 * @param source the message source.
+	 * @return true to proceed.
+	 */
+	public abstract boolean beforeReceive(MessageSource<?> source);
+
+	/**
+	 * Subclasses can take actions based on the result of the poll; e.g.
+	 * adjust the {@code trigger}. The message can also be replaced with a new one.
+	 * @param result the received message.
+	 * @param source the message source.
+	 * @return a message to continue to process the result, null to discard whatever the poll returned.
+	 */
+	public abstract Message<?> afterReceive(Message<?> result, MessageSource<?> source);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java

@@ -0,0 +1,80 @@
+/*
+ * 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.DynamicPeriodicTrigger;
+import org.springframework.messaging.Message;
+
+/**
+ * A simple advice that polls at one rate when messages exist and another when
+ * there are no messages.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ * @see DynamicPeriodicTrigger
+ */
+public class SimpleActiveIdleMessageSourceAdvice extends AbstractMessageSourceAdvice {
+
+	private final DynamicPeriodicTrigger trigger;
+
+	private volatile long idlePollPeriod;
+
+	private volatile long activePollPeriod;
+
+
+	public SimpleActiveIdleMessageSourceAdvice(DynamicPeriodicTrigger trigger) {
+		this.trigger = trigger;
+		this.idlePollPeriod = trigger.getPeriod();
+		this.activePollPeriod = trigger.getPeriod();
+	}
+
+	/**
+	 * Set the poll period when messages are not returned. Defaults to the
+	 * trigger's period.
+	 * @param idlePollPeriod the period in milliseconds.
+	 */
+	public void setIdlePollPeriod(long idlePollPeriod) {
+		this.idlePollPeriod = idlePollPeriod;
+	}
+
+	/**
+	 * Set the poll period when messages are returned. Defaults to the
+	 * trigger's period.
+	 * @param activePollPeriod the period in milliseconds.
+	 */
+	public void setActivePollPeriod(long activePollPeriod) {
+		this.activePollPeriod = activePollPeriod;
+	}
+
+	@Override
+	public boolean beforeReceive(MessageSource<?> source) {
+		return true;
+	}
+
+	@Override
+	public Message<?> afterReceive(Message<?> result, MessageSource<?> aource) {
+		if (result == null) {
+			this.trigger.setPeriod(this.idlePollPeriod);
+		}
+		else {
+			this.trigger.setPeriod(this.activePollPeriod);
+		}
+		return result;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.endpoint;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
@@ -109,6 +111,27 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 		this.transactionSynchronizationFactory = transactionSynchronizationFactory;
 	}
 
+	protected ClassLoader getBeanClassLoader() {
+		return beanClassLoader;
+	}
+
+	/**
+	 * Return true if this advice should be applied only to the {@link #receiveMessage()} operation
+	 * rather than the whole poll.
+	 * @param advice The advice.
+	 * @return true to only advise the receive operation.
+	 */
+	protected boolean isReceiveOnlyAdvice(Advice advice) {
+		return false;
+	}
+
+	/**
+	 * Add the advice chain to the component that responds to {@link #receiveMessage()} calls.
+	 * @param chain the advice chain {@code Collection}.
+	 */
+	protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
+	}
+
 	@Override
 	protected void onInit() {
 		synchronized (this.initializationMonitor) {
@@ -138,6 +161,14 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 
 	@SuppressWarnings("unchecked")
 	private Runnable createPoller() throws Exception {
+		List<Advice> receiveOnlyAdviceChain = new ArrayList<Advice>();
+		if (!CollectionUtils.isEmpty(adviceChain)) {
+			for (Advice advice : adviceChain) {
+				if (isReceiveOnlyAdvice(advice)) {
+					receiveOnlyAdviceChain.add(advice);
+				}
+			}
+		}
 
 		Callable<Boolean> pollingTask = new Callable<Boolean>() {
 			@Override
@@ -151,11 +182,16 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 			ProxyFactory proxyFactory = new ProxyFactory(pollingTask);
 			if (!CollectionUtils.isEmpty(adviceChain)) {
 				for (Advice advice : adviceChain) {
-					proxyFactory.addAdvice(advice);
+					if (!isReceiveOnlyAdvice(advice)) {
+						proxyFactory.addAdvice(advice);
+					}
 				}
 			}
 			pollingTask = (Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader);
 		}
+		if (receiveOnlyAdviceChain.size() > 0) {
+			applyReceiveOnlyAdviceChain(receiveOnlyAdviceChain);
+		}
 		return new Poller(pollingTask);
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java

@@ -16,7 +16,16 @@
 
 package org.springframework.integration.endpoint;
 
+import java.util.Collection;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.aop.framework.Advised;
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.AopUtils;
+import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
 import org.springframework.context.Lifecycle;
+import org.springframework.integration.aop.AbstractMessageSourceAdvice;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.core.MessagingTemplate;
 import org.springframework.integration.history.MessageHistory;
@@ -92,6 +101,29 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 				((NamedComponent) this.source).getComponentType() : "inbound-channel-adapter";
 	}
 
+	@Override
+	protected boolean isReceiveOnlyAdvice(Advice advice) {
+		return advice instanceof AbstractMessageSourceAdvice;
+	}
+
+	@Override
+	protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
+		if (AopUtils.isAopProxy(this.source)) {
+			for (Advice advice : chain) {
+				NameMatchMethodPointcutAdvisor sourceAdvice = new NameMatchMethodPointcutAdvisor(advice);
+				sourceAdvice.addMethodName("receive");
+				((Advised) this.source).addAdvice(advice);
+			}
+		}
+		else {
+			ProxyFactory proxyFactory = new ProxyFactory(this.source);
+			for (Advice advice : chain) {
+				proxyFactory.addAdvice(advice);
+			}
+			this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
+		}
+	}
+
 	@Override
 	protected void doStart() {
 		if (this.source instanceof Lifecycle) {

spring-integration-core/src/main/java/org/springframework/integration/util/DynamicPeriodicTrigger.java

@@ -0,0 +1,161 @@
+/*
+ * Copyright 2002-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 java.util.concurrent.TimeUnit;
+
+import org.springframework.scheduling.Trigger;
+import org.springframework.scheduling.TriggerContext;
+import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.util.Assert;
+
+
+/**
+ * This is a dynamically changeable {@link Trigger}. It is based on the
+ * {@link PeriodicTrigger} implementations. However, the fields of this dynamic
+ * trigger are not final and the properties can be inspected and set via
+ * explicit getters and setters.
+ *
+ * @author Gunnar Hillert
+ * @since 4.2
+ */
+public class DynamicPeriodicTrigger implements Trigger {
+
+	private volatile long period;
+
+	private volatile TimeUnit timeUnit;
+
+	private volatile long initialDelay = 0;
+
+	private volatile boolean fixedRate = false;
+
+	/**
+	 * Create a trigger with the given period in milliseconds. The underlying
+	 * {@link TimeUnit} will be initialized to TimeUnit.MILLISECONDS.
+	 * @param period Must not be negative
+	 */
+	public DynamicPeriodicTrigger(long period) {
+		this(period, TimeUnit.MILLISECONDS);
+	}
+
+	/**
+	 * Create a trigger with the given period and time unit. The time unit will
+	 * apply not only to the period but also to any 'initialDelay' value, if
+	 * configured on this Trigger later via {@link #setInitialDelay(long)}.
+	 * @param period Must not be negative
+	 * @param timeUnit Must not be null
+	 */
+	public DynamicPeriodicTrigger(long period, TimeUnit timeUnit) {
+		Assert.isTrue(period >= 0, "period must not be negative");
+		Assert.notNull(timeUnit, "timeUnit must not be null");
+
+		this.timeUnit = timeUnit;
+		this.period = this.timeUnit.toMillis(period);
+	}
+
+	/**
+	 * Specify the delay for the initial execution. It will be evaluated in
+	 * terms of this trigger's {@link TimeUnit}. If no time unit was explicitly
+	 * provided upon instantiation, the default is milliseconds.
+	 * @param initialDelay the initial delay in milliseconds.
+	 */
+	public void setInitialDelay(long initialDelay) {
+		Assert.isTrue(initialDelay >= 0, "initialDelay must not be negative");
+		this.initialDelay = this.timeUnit.toMillis(initialDelay);
+	}
+
+	/**
+	 * Specify whether the periodic interval should be measured between the
+	 * scheduled start times rather than between actual completion times.
+	 * The latter, "fixed delay" behavior, is the default.
+	 * @param fixedRate the fixed rate {@code boolean} flag.
+	 */
+	public void setFixedRate(boolean fixedRate) {
+		this.fixedRate = fixedRate;
+	}
+
+	/**
+	 * Return the time after which a task should run again.
+	 * @param triggerContext the trigger context to determine the previous state of schedule.
+	 * @return the the next schedule date.
+	 */
+	@Override
+	public Date nextExecutionTime(TriggerContext triggerContext) {
+		if (triggerContext.lastScheduledExecutionTime() == null) {
+			return new Date(System.currentTimeMillis() + this.initialDelay);
+		}
+		else if (this.fixedRate) {
+			return new Date(triggerContext.lastScheduledExecutionTime().getTime() + this.period);
+		}
+		return new Date(triggerContext.lastCompletionTime().getTime() + this.period);
+	}
+
+	public long getPeriod() {
+		return period;
+	}
+
+	/**
+	 * Specify the period of the trigger. It will be evaluated in
+	 * terms of this trigger's {@link TimeUnit}. If no time unit was explicitly
+	 * provided upon instantiation, the default is milliseconds.
+	 * @param period Must not be negative
+	 */
+	public void setPeriod(long period) {
+		Assert.isTrue(period >= 0, "period must not be negative");
+		this.period = this.timeUnit.toMillis(period);
+	}
+
+	public TimeUnit getTimeUnit() {
+		return timeUnit;
+	}
+
+	public void setTimeUnit(TimeUnit timeUnit) {
+		Assert.notNull(timeUnit, "timeUnit must not be null");
+		this.timeUnit = timeUnit;
+	}
+
+	public long getInitialDelay() {
+		return initialDelay;
+	}
+
+	public boolean isFixedRate() {
+		return fixedRate;
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (this == obj) {
+			return true;
+		}
+		if (!(obj instanceof DynamicPeriodicTrigger)) {
+			return false;
+		}
+		DynamicPeriodicTrigger other = (DynamicPeriodicTrigger) obj;
+		return this.fixedRate == other.fixedRate
+				&& this.initialDelay == other.initialDelay
+				&& this.period == other.period;
+	}
+
+	@Override
+	public int hashCode() {
+		return (this.fixedRate ? 14 : 41) +
+				(int) (38 * this.period) +
+				(int) (43 * this.initialDelay);
+	}
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014 the original author or authors.
+ * Copyright 2014-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.
@@ -16,25 +16,37 @@
 
 package org.springframework.integration.endpoint;
 
+import static org.hamcrest.Matchers.contains;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Date;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 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.integration.aop.AbstractMessageSourceAdvice;
+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.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;
@@ -46,6 +58,8 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
  */
 public class PollerAdviceTests {
 
+	public Message<?> receiveAdviceResult;
+
 	@Test
 	public void testDefaultDontSkip() throws Exception {
 		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
@@ -60,9 +74,13 @@ public class PollerAdviceTests {
 		});
 		adapter.setTrigger(new Trigger() {
 
+			private boolean done;
+
 			@Override
 			public Date nextExecutionTime(TriggerContext triggerContext) {
-				return new Date(System.currentTimeMillis() + 10);
+				Date date = done ? null : new Date(System.currentTimeMillis() + 10);
+				done = true;
+				return date;
 			}
 		});
 		configure(adapter);
@@ -90,9 +108,13 @@ public class PollerAdviceTests {
 		});
 		adapter.setTrigger(new Trigger() {
 
+			private boolean done;
+
 			@Override
 			public Date nextExecutionTime(TriggerContext triggerContext) {
-				return new Date(System.currentTimeMillis() + 10);
+				Date date = done ? null : new Date(System.currentTimeMillis() + 10);
+				done = true;
+				return date;
 			}
 		});
 		configure(adapter);
@@ -113,6 +135,108 @@ public class PollerAdviceTests {
 		adapter.stop();
 	}
 
+	@Test
+	public void testMixedAdvice() throws Exception {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		final List<String> callOrder = new ArrayList<String>();
+		final CountDownLatch latch = new CountDownLatch(4);// advice + advice + source + advice
+		adapter.setSource(new MessageSource<Object>() {
+
+			@Override
+			public Message<Object> receive() {
+				callOrder.add("c");
+				latch.countDown();
+				return null;
+			}
+		});
+		adapter.setTrigger(new Trigger() {
+
+			private boolean done;
+
+			@Override
+			public Date nextExecutionTime(TriggerContext triggerContext) {
+				Date date = done ? null : new Date(System.currentTimeMillis() + 10);
+				done = true;
+				return date;
+			}
+		});
+		configure(adapter);
+		List<Advice> adviceChain = new ArrayList<Advice>();
+
+		class TestGeneralAdvice implements MethodInterceptor {
+
+			@Override
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				callOrder.add("a");
+				latch.countDown();
+				return invocation.proceed();
+			}
+
+		}
+		adviceChain.add(new TestGeneralAdvice());
+
+		class TestSourceAdvice extends AbstractMessageSourceAdvice {
+
+			@Override
+			public boolean beforeReceive(MessageSource<?> target) {
+				callOrder.add("b");
+				latch.countDown();
+				return true;
+			}
+
+			@Override
+			public Message<?> afterReceive(Message<?> result, MessageSource<?> target) {
+				callOrder.add("d");
+				latch.countDown();
+				return result;
+			}
+
+		}
+		adviceChain.add(new TestSourceAdvice());
+
+		adapter.setAdviceChain(adviceChain);
+		adapter.afterPropertiesSet();
+		adapter.start();
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		assertThat(callOrder, contains("a", "b", "c", "d")); // advice + advice + source + advice
+		adapter.stop();
+	}
+
+	@Test
+	public void testActiveIdleAdvice() throws Exception {
+		SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
+		final CountDownLatch latch = new CountDownLatch(5);
+		final LinkedList<Long> triggerPeriods = new LinkedList<Long>();
+		final DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(10);
+		adapter.setSource(new MessageSource<Object>() {
+
+			@Override
+			public Message<Object> receive() {
+				triggerPeriods.add(trigger.getPeriod());
+				Message<Object> m = null;
+				if (latch.getCount() % 2 == 0) {
+					m = new GenericMessage<Object>("foo");
+				}
+				latch.countDown();
+				return m;
+			}
+		});
+		QueueChannel channel = new QueueChannel();
+		SimpleActiveIdleMessageSourceAdvice toggling = new SimpleActiveIdleMessageSourceAdvice(trigger);
+		toggling.setActivePollPeriod(11);
+		toggling.setIdlePollPeriod(12);
+		adapter.setAdviceChain(Collections.singletonList(toggling));
+		configure(adapter);
+		adapter.afterPropertiesSet();
+		adapter.start();
+		assertTrue(latch.await(10, TimeUnit.SECONDS));
+		adapter.stop();
+		while (triggerPeriods.size() > 5) {
+			triggerPeriods.removeLast();
+		}
+		assertThat(triggerPeriods, Matchers.contains(10L, 12L, 11L, 12L, 11L));
+	}
+
 	private void configure(SourcePollingChannelAdapter adapter) {
 		adapter.setOutputChannel(new NullChannel());
 		adapter.setBeanFactory(mock(BeanFactory.class));