From 2d9a5f60f48330e03e5a24ddc887560d6d02b514 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 28 Apr 2020 13:03:30 -0400 Subject: [PATCH] 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` --- .../aop/AbstractMessageSourceAdvice.java | 26 ++---- .../aop/CompoundTriggerAdvice.java | 21 ++++- .../integration/aop/MessageSourceMutator.java | 37 +++++++-- .../integration/aop/ReceiveMessageAdvice.java | 72 +++++++++++++++++ .../SimpleActiveIdleMessageSourceAdvice.java | 6 +- .../SimpleActiveIdleReceiveMessageAdvice.java | 80 +++++++++++++++++++ .../endpoint/AbstractPollingEndpoint.java | 48 ++++++++++- .../integration/endpoint/PollingConsumer.java | 16 +++- .../endpoint/SourcePollingChannelAdapter.java | 58 +++----------- .../endpoint/PollerAdviceTests.java | 80 +++++++++++++++---- .../file/remote/aop/RotatingServerAdvice.java | 13 ++- ...InboundChannelAdapterIntegrationTests.java | 14 +--- src/reference/asciidoc/ftp.adoc | 4 +- src/reference/asciidoc/polling-consumer.adoc | 37 ++++++--- src/reference/asciidoc/sftp.adoc | 4 +- src/reference/asciidoc/whats-new.adoc | 6 ++ 16 files changed, 394 insertions(+), 128 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aop/ReceiveMessageAdvice.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleReceiveMessageAdvice.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java index 81a02714bb..61ffa02181 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java +++ b/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 { } 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 index 6f6c6103f6..7171403552 100644 --- 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 @@ -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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessageSourceMutator.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessageSourceMutator.java index 66b6ff83c5..eb60dbdcd4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessageSourceMutator.java +++ b/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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/ReceiveMessageAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/ReceiveMessageAdvice.java new file mode 100644 index 0000000000..e6054147e5 --- /dev/null +++ b/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); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java index d7dae0ded2..7e468ddeb6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java +++ b/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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleReceiveMessageAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleReceiveMessageAdvice.java new file mode 100644 index 0000000000..9d2f506711 --- /dev/null +++ b/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; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java index 937f0ff5eb..6e89c8f1d8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java +++ b/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 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 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java index 8744edb630..1986a5c43b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingConsumer.java +++ b/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 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 && diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index caebab68c9..ef4286350b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/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 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 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 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 d554b87b7d..5e08c409bf 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 @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-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. @@ -35,8 +35,7 @@ import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.Joinpoint; import org.aopalliance.intercept.MethodInterceptor; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.aop.Advisor; import org.springframework.aop.framework.Advised; @@ -48,9 +47,9 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.aop.AbstractMessageSourceAdvice; import org.springframework.integration.aop.CompoundTriggerAdvice; -import org.springframework.integration.aop.SimpleActiveIdleMessageSourceAdvice; +import org.springframework.integration.aop.ReceiveMessageAdvice; +import org.springframework.integration.aop.SimpleActiveIdleReceiveMessageAdvice; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.NullChannel; import org.springframework.integration.config.EnableIntegration; @@ -64,14 +63,14 @@ import org.springframework.integration.util.CompoundTrigger; import org.springframework.integration.util.DynamicPeriodicTrigger; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; 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; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Gary Russell @@ -80,8 +79,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @since 4.1 * */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class PollerAdviceTests { @@ -194,10 +192,10 @@ public class PollerAdviceTests { }); final AtomicInteger count = new AtomicInteger(); - class TestSourceAdvice extends AbstractMessageSourceAdvice { + class TestSourceAdvice implements ReceiveMessageAdvice { @Override - public boolean beforeReceive(MessageSource target) { + public boolean beforeReceive(Object target) { count.incrementAndGet(); callOrder.add("b"); latch.get().countDown(); @@ -205,7 +203,7 @@ public class PollerAdviceTests { } @Override - public Message afterReceive(Message result, MessageSource target) { + public Message afterReceive(Message result, Object target) { callOrder.add("d"); latch.get().countDown(); return result; @@ -267,7 +265,7 @@ public class PollerAdviceTests { latch.countDown(); return m; }); - SimpleActiveIdleMessageSourceAdvice toggling = new SimpleActiveIdleMessageSourceAdvice(trigger); + SimpleActiveIdleReceiveMessageAdvice toggling = new SimpleActiveIdleReceiveMessageAdvice(trigger); toggling.setActivePollPeriod(11); toggling.setIdlePollPeriod(12); adapter.setAdviceChain(Collections.singletonList(toggling)); @@ -282,6 +280,56 @@ public class PollerAdviceTests { } } + @Test + public void testActiveIdleAdviceOnQueueChannel() throws Exception { + final CountDownLatch latch = new CountDownLatch(5); + final LinkedList triggerPeriods = new LinkedList<>(); + final DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(10); + + PollingConsumer pollingConsumer = + new PollingConsumer(new PollableChannel() { + + @Override + public Message receive() { + synchronized (triggerPeriods) { + triggerPeriods.add(trigger.getDuration().toMillis()); + } + Message m = null; + if (latch.getCount() % 2 == 0) { + m = new GenericMessage<>("foo"); + } + latch.countDown(); + return m; + } + + @Override + public Message receive(long timeout) { + return receive(); + } + + @Override + public boolean send(Message message, long timeout) { + return false; + } + + }, m -> { }); + + SimpleActiveIdleReceiveMessageAdvice toggling = new SimpleActiveIdleReceiveMessageAdvice(trigger); + toggling.setActivePollPeriod(11); + toggling.setIdlePollPeriod(12); + pollingConsumer.setAdviceChain(Collections.singletonList(toggling)); + pollingConsumer.setTrigger(trigger); + pollingConsumer.setBeanFactory(this.beanFactory); + pollingConsumer.setTaskScheduler(this.threadPoolTaskScheduler); + pollingConsumer.afterPropertiesSet(); + pollingConsumer.start(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + pollingConsumer.stop(); + synchronized (triggerPeriods) { + assertThat(triggerPeriods.subList(0, 5)).containsExactly(10L, 12L, 11L, 12L, 11L); + } + } + @Test public void testCompoundTriggerAdvice() throws Exception { SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); @@ -361,18 +409,18 @@ public class PollerAdviceTests { } - public static class OtherAdvice extends AbstractMessageSourceAdvice { + public static class OtherAdvice implements ReceiveMessageAdvice { private int calls; @Override - public boolean beforeReceive(MessageSource source) { + public boolean beforeReceive(Object source) { this.calls++; return true; } @Override - public Message afterReceive(Message result, MessageSource source) { + public Message afterReceive(Message result, Object source) { return result; } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java index 1f637535e3..47088d9ea5 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/aop/RotatingServerAdvice.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. @@ -18,9 +18,10 @@ package org.springframework.integration.file.remote.aop; import java.util.List; -import org.springframework.integration.aop.AbstractMessageSourceAdvice; +import org.springframework.integration.aop.MessageSourceMutator; import org.springframework.integration.core.MessageSource; import org.springframework.integration.file.remote.session.DelegatingSessionFactory; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -35,7 +36,10 @@ import org.springframework.util.Assert; * @since 5.0.7 * */ -public class RotatingServerAdvice extends AbstractMessageSourceAdvice { +@SuppressWarnings("deprecation") +public class RotatingServerAdvice + extends org.springframework.integration.aop.AbstractMessageSourceAdvice + implements MessageSourceMutator { private final RotationPolicy rotationPolicy; @@ -79,7 +83,8 @@ public class RotatingServerAdvice extends AbstractMessageSourceAdvice { } @Override - public Message afterReceive(Message result, MessageSource source) { + @Nullable + public Message afterReceive(@Nullable Message result, MessageSource source) { this.rotationPolicy.afterReceive(result != null, source); return result; } diff --git a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java index b4b8877276..79f1258b0d 100644 --- a/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.java +++ b/spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/config/MongoDbInboundChannelAdapterIntegrationTests.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. @@ -31,9 +31,8 @@ import org.springframework.data.mongodb.core.MongoOperations; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; -import org.springframework.integration.aop.AbstractMessageSourceAdvice; +import org.springframework.integration.aop.ReceiveMessageAdvice; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.mongodb.rules.MongoDbAvailable; import org.springframework.integration.mongodb.rules.MongoDbAvailableTests; @@ -278,15 +277,10 @@ public class MongoDbInboundChannelAdapterIntegrationTests extends MongoDbAvailab } - public static final class TestMessageSourceAdvice extends AbstractMessageSourceAdvice { + public static final class TestMessageSourceAdvice implements ReceiveMessageAdvice { @Override - public boolean beforeReceive(MessageSource source) { - return true; - } - - @Override - public Message afterReceive(Message result, MessageSource source) { + public Message afterReceive(Message result, Object source) { return result; } diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 4983ce66f7..89b8230eae 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/src/reference/asciidoc/ftp.adoc @@ -263,7 +263,7 @@ The following example shows how to declare a delegating session factory: IMPORTANT: When you use session caching (see <>), each of the delegates should be cached. You cannot cache the `DelegatingSessionFactory` itself. -Starting with _version 5.0.7_, the `DelegatingSessionFactory` can be used in conjunction with a `RotatingServerAdvice` to poll multiple servers; see <>. +Starting with version 5.0.7, the `DelegatingSessionFactory` can be used in conjunction with a `RotatingServerAdvice` to poll multiple servers; see <>. [[ftp-inbound]] === FTP Inbound Channel Adapter @@ -704,7 +704,7 @@ Notice that, in this example, the message handler downstream of the transformer [[ftp-rotating-server-advice]] === Inbound Channel Adapters: Polling Multiple Servers and Directories -Starting with _version 5.0.7_, the `RotatingServerAdvice` is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories. +Starting with version 5.0.7, the `RotatingServerAdvice` is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories. Configure the advice and add it to the poller's advice chain as normal. A `DelegatingSessionFactory` is used to select the server see <> for more information. The advice configuration consists of a list of `RotationPolicy.KeyDirectory` objects. diff --git a/src/reference/asciidoc/polling-consumer.adoc b/src/reference/asciidoc/polling-consumer.adoc index d1160d837a..e6303e194e 100644 --- a/src/reference/asciidoc/polling-consumer.adoc +++ b/src/reference/asciidoc/polling-consumer.adoc @@ -145,18 +145,21 @@ These "`around advice`" methods do not have access to any context for the poll - This is fine for requirements such as making a task transactional or skipping a poll due to some external condition, as discussed earlier. What if we wish to take some action depending on the result of the `receive` part of the poll or if we want to adjust the poller depending on conditions? For those instances, Spring Integration offers "`Smart`" Polling. +[[smart-polling]] ===== "`Smart`" Polling -Version 4.2 introduced the `AbstractMessageSourceAdvice`. -Any `Advice` objects in the `advice-chain` that subclass this class are applied only to the receive operation. +Version 5.3 introduced the `ReceiveMessageAdvice` interface. +(The `AbstractMessageSourceAdvice` has been deprecated in favor of `default` methods in the `MessageSourceMutator`.) +Any `Advice` objects in the `advice-chain` that implement this interface are applied only to the receive operation - `MessageSource.receive()` and `PollableChannel.receive(timeout)`. +Therefore they can be applied only for the `SourcePollingChannelAdapter` or `PollingConsumer`. Such classes implement the following methods: -* `beforeReceive(MessageSource source)` -This method is called before the `MessageSource.receive()` method. +* `beforeReceive(Object source)` +This method is called before the `Object.receive()` method. It lets you examine and reconfigure the source. Returning `false` cancels this poll (similar to the `PollSkipAdvice` mentioned earlier). -* `Message afterReceive(Message result, MessageSource source)` +* `Message afterReceive(Message result, Object 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 @@ -164,7 +167,7 @@ You can even return a different message .Thread safety [IMPORTANT] ==== -If an advice mutates the `MessageSource`, you should not configure the poller with a `TaskExecutor`. +If an advice mutates the the, you should not configure the poller with a `TaskExecutor`. If an advice mutates the source, such mutations are not thread safe and could cause unexpected results, especially with high frequency pollers. If you need to process poll results concurrently, consider using a downstream `ExecutorChannel` instead of adding an executor to the poller. ==== @@ -173,21 +176,22 @@ If you need to process poll results concurrently, consider using a downstream `E [IMPORTANT] ===== You should 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, for example, `Advice` objects `a, b, c, d`, where `b` and `d` are `AbstractMessageSourceAdvice`, the objects are applied in the following order: `a, c, b, d`. -Also, if a `MessageSource` is already a `Proxy`, the `AbstractMessageSourceAdvice` is invoked after any existing `Advice` objects. +`Advice` objects that do not implement `ReceiveMessageAdvice` are applied to the whole poll process and are all invoked first, in order, before any `ReceiveMessageAdvice`. +Then `ReceiveMessageAdvice` objects are invoked in order around the source `receive()` method. +If you have, for example, `Advice` objects `a, b, c, d`, where `b` and `d` are `ReceiveMessageAdvice`, the objects are applied in the following order: `a, c, b, d`. +Also, if a source is already a `Proxy`, the `ReceiveMessageAdvice` is invoked after any existing `Advice` objects. If you wish to change the order, you must wire up the proxy yourself. ===== -===== `SimpleActiveIdleMessageSourceAdvice` +===== `SimpleActiveIdleReceiveMessageAdvice` -This advice is a simple implementation of `AbstractMessageSourceAdvice`. +(The previous `SimpleActiveIdleMessageSourceAdvice` for only `MessageSource` is deprecated.) +This advice is a simple implementation of `ReceiveMessageAdvice`. When used in conjunction with a `DynamicPeriodicTrigger`, it adjusts the polling frequency, depending on whether or not the previous poll resulted in a message or not. The poller must also have a reference to the same `DynamicPeriodicTrigger`. .Important: Async Handoff -IMPORTANT: `SimpleActiveIdleMessageSourceAdvice` modifies the trigger based on the `receive()` result. +IMPORTANT: `SimpleActiveIdleReceiveMessageAdvice` modifies the trigger based on the `receive()` result. This works only if the advice is called on the poller thread. It does 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`. @@ -241,3 +245,10 @@ IMPORTANT: `CompoundTriggerAdvice` modifies the trigger based on the `receive()` This works only if the advice is called on the poller thread. It does 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`. + +===== MessageSource-only Advices + +Some advices might be applied only for the `MessageSource.receive()` and they don't make sense for `PollableChannel`. +For this purpose a `MessageSourceMutator` interface (an extension of the `ReceiveMessageAdvice`) is still present. +With `default` methods it fully replaces already deprecated `AbstractMessageSourceAdvice` and should be used in those implementations where only `MessageSource` proxying is expected. +See <<./ftp.adoc#ftp-rotating-server-advice,Inbound Channel Adapters: Polling Multiple Servers and Directories>> for more information. diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index d553c45aec..737d149efc 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/src/reference/asciidoc/sftp.adoc @@ -230,7 +230,7 @@ We added convenience methods so that you can more easily do so from a message fl IMPORTANT: When using session caching (see <>), each of the delegates should be cached. You cannot cache the `DelegatingSessionFactory` itself. -Starting with _version 5.0.7_, the `DelegatingSessionFactory` can be used in conjunction with a `RotatingServerAdvice` to poll multiple servers; see <>. +Starting with version 5.0.7, the `DelegatingSessionFactory` can be used in conjunction with a `RotatingServerAdvice` to poll multiple servers; see <>. [[sftp-session-caching]] === SFTP Session Caching @@ -703,7 +703,7 @@ Notice that, in this example, the message handler downstream of the transformer [[sftp-rotating-server-advice]] === Inbound Channel Adapters: Polling Multiple Servers and Directories -Starting with _version 5.0.7_, the `RotatingServerAdvice` is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories. +Starting with version 5.0.7, the `RotatingServerAdvice` is available; when configured as a poller advice, the inbound adapters can poll multiple servers and directories. Configure the advice and add it to the poller's advice chain as normal. A `DelegatingSessionFactory` is used to select the server see <<./ftp.adoc#ftp-dsf,Delegating Session Factory>> for more information. The advice configuration consists of a list of `RotationPolicy.KeyDirectory` objects. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 99e3601188..b24d71edf7 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -66,6 +66,12 @@ The `spring-integration-mongodb` module now provides channel adapter implementat Also, a reactive implementation for MongoDb change stream support is present with the `MongoDbChangeStreamMessageProducer`. See <<./mongodb.adoc#mongodb,MongoDB Support>> for more information. +[[x5.3-receive-message-advice]] +==== ReceiveMessageAdvice + +A special `ReceiveMessageAdvice` has been introduced to proxy exactly `MessageSource.receive()` or `PollableChannel.receive()`. +See <<./polling-consumer.adoc#smart-polling,Smart Polling>> for more information. + [[x5.3-general]] === General Changes