Introduce a ReceiveMessageAdvice (#3265)

* Introduce a `ReceiveMessageAdvice`

* Deprecate an `AbstractMessageSourceAdvice` in favor of
`default` method in the `MessageSourceMutator`
* Move a `applyReceiveOnlyAdviceChain()` logic into the `AbstractPollingEndpoint`:
now both `PollingConsumer` and `SourcePollingChannelAdapter` can use
`ReceiveMessageAdvice`
* Introduce a `SimpleActiveIdleReceiveMessageAdvice` based already
on the `ReceiveMessageAdvice` and deprecate a `SimpleActiveIdleMessageSourceAdvice`
which is fully replaceable with newly introduced `SimpleActiveIdleReceiveMessageAdvice`
* Add `@SuppressWarnings("deprecation")` for those out-of-the-box `ReceiveMessageAdvice`
implementation which still use an `AbstractMessageSourceAdvice` for
backward compatibility
* Document a new feature and give the `MessageSourceMutator` a new meaning

* * Fix language in the `polling-consumer.adoc`

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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,34 +16,20 @@
 
 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
+ * @author Artem Bilan
  *
  * @since 4.2
+ *
+ * @deprecated since 5.3 in favor of {@link MessageSourceMutator}.
  */
-public abstract class AbstractMessageSourceAdvice implements MethodInterceptor, MessageSourceMutator {
-
-	@Override
-	public final Object invoke(MethodInvocation invocation) throws Throwable {
-		Object target = invocation.getThis();
-		if (!(target instanceof MessageSource)) {
-			return invocation.proceed();
-		}
-
-		Message<?> result = null;
-		if (beforeReceive((MessageSource<?>) target)) {
-			result = (Message<?>) invocation.proceed();
-		}
-		return afterReceive(result, (MessageSource<?>) target);
-	}
+@Deprecated
+public abstract class AbstractMessageSourceAdvice implements MessageSourceMutator {
 
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.aop;
 
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.util.CompoundTrigger;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.scheduling.Trigger;
 import org.springframework.util.Assert;
@@ -35,7 +36,10 @@ import org.springframework.util.Assert;
  * @since 4.3
  *
  */
-public class CompoundTriggerAdvice extends AbstractMessageSourceAdvice {
+@SuppressWarnings("deprecation")
+public class CompoundTriggerAdvice
+		extends AbstractMessageSourceAdvice
+		implements ReceiveMessageAdvice {
 
 	private final CompoundTrigger compoundTrigger;
 
@@ -47,8 +51,21 @@ public class CompoundTriggerAdvice extends AbstractMessageSourceAdvice {
 		this.override = overrideTrigger;
 	}
 
+	/**
+	 * @param result the received message.
+	 * @param source the message source.
+	 * @return the message or null
+	 * @deprecated since 5.3 in favor of {@link #afterReceive(Message, Object)}
+	 */
 	@Override
+	@Deprecated
 	public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
+		return afterReceive(result, (Object) source);
+	}
+
+	@Override
+	@Nullable
+	public Message<?> afterReceive(@Nullable Message<?> result, Object source) {
 		if (result == null) {
 			this.compoundTrigger.setOverride(this.override);
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -17,19 +17,31 @@
 package org.springframework.integration.aop;
 
 import org.springframework.integration.core.MessageSource;
+import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 
 /**
- * An object that can mutate a {@link MessageSource} before and/or after
+ * A {@link ReceiveMessageAdvice} extension that can mutate a {@link MessageSource} before and/or after
  * {@link MessageSource#receive()} is called.
  *
  * @author Gary Russell
+ * @author Artem Bilan
  *
- * @since 5.0.7.
- *
+ * @since 5.0.7
  */
 @FunctionalInterface
-public interface MessageSourceMutator {
+public interface MessageSourceMutator extends ReceiveMessageAdvice {
+
+	@Override
+	default boolean beforeReceive(Object source) {
+		if (source instanceof MessageSource<?>) {
+			return beforeReceive((MessageSource<?>) source);
+		}
+		else {
+			throw new IllegalArgumentException(
+					"The 'MessageSourceMutator' supports only a 'MessageSource' in the before/after hooks: " + source);
+		}
+	}
 
 	/**
 	 * Subclasses can decide whether to proceed with this poll.
@@ -40,6 +52,18 @@ public interface MessageSourceMutator {
 		return true;
 	}
 
+	@Override
+	@Nullable
+	default Message<?> afterReceive(@Nullable Message<?> result, Object source) {
+		if (source instanceof MessageSource<?>) {
+			return afterReceive(result, (MessageSource<?>) source);
+		}
+		else {
+			throw new IllegalArgumentException(
+					"The 'MessageSourceMutator' supports only a 'MessageSource' in the before/after hooks: " + 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.
@@ -47,6 +71,7 @@ public interface MessageSourceMutator {
 	 * @param source the message source.
 	 * @return a message to continue to process the result, null to discard whatever the poll returned.
 	 */
-	Message<?> afterReceive(Message<?> result, MessageSource<?> source);
+	@Nullable
+	Message<?> afterReceive(@Nullable Message<?> result, MessageSource<?> source);
 
 }

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

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.PollableChannel;
+
+/**
+ * An AOP advice to perform hooks before and/or after a {@code receive()} contract is called.
+ *
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ */
+@FunctionalInterface
+public interface ReceiveMessageAdvice extends MethodInterceptor {
+
+	/**
+	 * Subclasses can decide whether to {@link MethodInvocation#proceed()} or not.
+	 * @param source the source of the message to receive.
+	 * @return true to proceed (default).
+	 */
+	default boolean beforeReceive(Object source) {
+		return true;
+	}
+
+	@Override
+	@Nullable
+	default Object invoke(MethodInvocation invocation) throws Throwable {
+		Object target = invocation.getThis();
+		if (!(target instanceof MessageSource) && !(target instanceof PollableChannel)) {
+			return invocation.proceed();
+		}
+
+		Message<?> result = null;
+		if (beforeReceive(target)) {
+			result = (Message<?>) invocation.proceed();
+		}
+		return afterReceive(result, target);
+	}
+
+	/**
+	 * Subclasses can take actions based on the result of the {@link MethodInvocation#proceed()}; e.g.
+	 * adjust the {@code trigger}. The message can also be replaced with a new one.
+	 * @param result the received message.
+	 * @param source the source of the message to receive.
+	 * @return a message to continue to process the result, null to discard whatever
+	 * the {@link MethodInvocation#proceed()} returned.
+	 */
+	@Nullable
+	Message<?> afterReceive(@Nullable Message<?> result, Object source);
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -31,7 +31,11 @@ import org.springframework.messaging.Message;
  * @since 4.2
  *
  * @see DynamicPeriodicTrigger
+ *
+ * @deprecated since 5.3 in favor of {@link SimpleActiveIdleReceiveMessageAdvice} with the same
+ * (but more common) functionality.
  */
+@Deprecated
 public class SimpleActiveIdleMessageSourceAdvice extends AbstractMessageSourceAdvice {
 
 	private final DynamicPeriodicTrigger trigger;

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

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2020 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
+ *
+ *      https://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 java.time.Duration;
+
+import org.springframework.integration.util.DynamicPeriodicTrigger;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ A simple advice that polls at one rate when messages exist and another when
+ * there are no messages.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 5.3
+ *
+ * @see DynamicPeriodicTrigger
+ */
+public class SimpleActiveIdleReceiveMessageAdvice implements ReceiveMessageAdvice {
+
+	private final DynamicPeriodicTrigger trigger;
+
+	private volatile Duration idlePollPeriod;
+
+	private volatile Duration activePollPeriod;
+
+	public SimpleActiveIdleReceiveMessageAdvice(DynamicPeriodicTrigger trigger) {
+		Assert.notNull(trigger, "'trigger' must not be null");
+		this.trigger = trigger;
+		this.idlePollPeriod = trigger.getDuration();
+		this.activePollPeriod = trigger.getDuration();
+	}
+
+	/**
+	 * 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 = Duration.ofMillis(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 = Duration.ofMillis(activePollPeriod);
+	}
+
+	@Override
+	public Message<?> afterReceive(Message<?> result, Object source) {
+		if (result == null) {
+			this.trigger.setDuration(this.idlePollPeriod);
+		}
+		else {
+			this.trigger.setDuration(this.activePollPeriod);
+		}
+		return result;
+	}
+
+}

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

@@ -19,6 +19,7 @@ package org.springframework.integration.endpoint;
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Date;
+import java.util.HashSet;
 import java.util.List;
 import java.util.concurrent.Callable;
 import java.util.concurrent.Executor;
@@ -28,10 +29,14 @@ import java.util.stream.Collectors;
 import org.aopalliance.aop.Advice;
 import org.reactivestreams.Subscription;
 
+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.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.core.task.SyncTaskExecutor;
+import org.springframework.integration.aop.ReceiveMessageAdvice;
 import org.springframework.integration.channel.ChannelUtils;
 import org.springframework.integration.channel.MessagePublishingErrorHandler;
 import org.springframework.integration.support.MessagingExceptionWrapper;
@@ -72,6 +77,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 	 */
 	public static final long DEFAULT_POLLING_PERIOD = 10;
 
+	private final Collection<Advice> appliedAdvices = new HashSet<>();
+
 	private final Object initializationMonitor = new Object();
 
 	private Executor taskExecutor = new SyncTaskExecutor();
@@ -173,7 +180,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 	 * @return true to only advise the receive operation.
 	 */
 	protected boolean isReceiveOnlyAdvice(Advice advice) {
-		return false;
+		return advice instanceof ReceiveMessageAdvice;
 	}
 
 	/**
@@ -181,6 +188,37 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 	 * @param chain the advice chain {@code Collection}.
 	 */
 	protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
+		if (!CollectionUtils.isEmpty(chain)) {
+			Object source = getReceiveMessageSource();
+			if (source != null) {
+				if (AopUtils.isAopProxy(source)) {
+					Advised advised = (Advised) source;
+					this.appliedAdvices.forEach(advised::removeAdvice);
+					chain.forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
+				}
+				else {
+					ProxyFactory proxyFactory = new ProxyFactory(source);
+					chain.forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
+					source = proxyFactory.getProxy(getBeanClassLoader());
+				}
+				this.appliedAdvices.clear();
+				this.appliedAdvices.addAll(chain);
+				if (!(isSyncExecutor()) && logger.isWarnEnabled()) {
+					logger.warn(getComponentName() + ": A task executor is supplied and " + chain.size()
+							+ "ReceiveMessageAdvice(s) is/are provided. If an advice mutates the source, such "
+							+ "mutations are not thread safe and could cause unexpected results, especially with "
+							+ "high frequency pollers. Consider using a downstream ExecutorChannel instead of "
+							+ "adding an executor to the poller");
+				}
+				setReceiveMessageSource(source);
+			}
+		}
+	}
+
+	private NameMatchMethodPointcutAdvisor adviceToReceiveAdvisor(Advice advice) {
+		NameMatchMethodPointcutAdvisor sourceAdvisor = new NameMatchMethodPointcutAdvisor(advice);
+		sourceAdvisor.addMethodName("receive");
+		return sourceAdvisor;
 	}
 
 	protected boolean isReactive() {
@@ -191,6 +229,14 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 		return this.pollingFlux;
 	}
 
+	protected Object getReceiveMessageSource() {
+		return null;
+	}
+
+	protected void setReceiveMessageSource(Object source) {
+
+	}
+
 	@Override
 	protected void onInit() {
 		synchronized (this.initializationMonitor) {

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

@@ -57,13 +57,13 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat
 	 */
 	public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
 
-	private final PollableChannel inputChannel;
-
 	private final MessageHandler handler;
 
 	private final List<ChannelInterceptor> channelInterceptors;
 
-	private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
+	private PollableChannel inputChannel;
+
+	private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
 
 	public PollingConsumer(PollableChannel inputChannel, MessageHandler handler) {
 		Assert.notNull(inputChannel, "inputChannel must not be null");
@@ -110,6 +110,16 @@ public class PollingConsumer extends AbstractPollingEndpoint implements Integrat
 		return this.handler;
 	}
 
+	@Override
+	protected Object getReceiveMessageSource() {
+		return this.inputChannel;
+	}
+
+	@Override
+	protected void setReceiveMessageSource(Object source) {
+		this.inputChannel = (PollableChannel) source;
+	}
+
 	@Override
 	protected boolean isReactive() {
 		return getOutputChannel() instanceof ReactiveStreamsSubscribableChannel &&

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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,21 +16,12 @@
 
 package org.springframework.integration.endpoint;
 
-import java.util.Collection;
-import java.util.HashSet;
-
-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.beans.factory.BeanCreationException;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AckUtils;
 import org.springframework.integration.acks.AcknowledgmentCallback;
-import org.springframework.integration.aop.MessageSourceMutator;
 import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
 import org.springframework.integration.context.ExpressionCapable;
 import org.springframework.integration.core.MessageSource;
@@ -43,7 +34,6 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
-import org.springframework.util.CollectionUtils;
 
 /**
  * A Channel Adapter implementation for connecting a
@@ -59,9 +49,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 
 	private final MessagingTemplate messagingTemplate = new MessagingTemplate();
 
-	private final Collection<Advice> appliedAdvices = new HashSet<>();
-
-	private volatile MessageSource<?> originalSource;
+	private MessageSource<?> originalSource;
 
 	private volatile MessageSource<?> source;
 
@@ -136,45 +124,19 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 				((NamedComponent) this.source).getComponentType() : "inbound-channel-adapter";
 	}
 
-	@Override
-	protected boolean isReceiveOnlyAdvice(Advice advice) {
-		return advice instanceof MessageSourceMutator;
-	}
-
-	@Override
-	protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
-		if (!CollectionUtils.isEmpty(chain)) {
-			if (AopUtils.isAopProxy(this.source)) {
-				Advised advised = (Advised) this.source;
-				this.appliedAdvices.forEach(advised::removeAdvice);
-				chain.forEach(advice -> advised.addAdvisor(adviceToReceiveAdvisor(advice)));
-			}
-			else {
-				ProxyFactory proxyFactory = new ProxyFactory(this.source);
-				chain.forEach(advice -> proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice)));
-				this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
-			}
-			this.appliedAdvices.clear();
-			this.appliedAdvices.addAll(chain);
-			if (!(isSyncExecutor()) && logger.isWarnEnabled()) {
-				logger.warn(getComponentName() + ": A task executor is supplied and " + chain.size()
-						+ "MessageSourceMutator(s) is/are provided. If an advice mutates the source, such "
-						+ "mutations are not thread safe and could cause unexpected results, especially with "
-						+ "high frequency pollers. Consider using a downstream ExecutorChannel instead of "
-						+ "adding an executor to the poller");
-			}
-		}
-	}
-
 	@Override
 	protected boolean isReactive() {
 		return getOutputChannel() instanceof ReactiveStreamsSubscribableChannel;
 	}
 
-	private NameMatchMethodPointcutAdvisor adviceToReceiveAdvisor(Advice advice) {
-		NameMatchMethodPointcutAdvisor sourceAdvisor = new NameMatchMethodPointcutAdvisor(advice);
-		sourceAdvisor.addMethodName("receive");
-		return sourceAdvisor;
+	@Override
+	protected Object getReceiveMessageSource() {
+		return getMessageSource();
+	}
+
+	@Override
+	protected final void setReceiveMessageSource(Object source) {
+		this.source = (MessageSource<?>) source;
 	}
 
 	@Override