From 75ec449c0b8e6ecffb7d98444082d1e311f5d906 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 4 Mar 2015 19:24:00 -0500 Subject: [PATCH] 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 --- .../aop/AbstractMessageSourceAdvice.java | 65 +++++++ .../SimpleActiveIdleMessageSourceAdvice.java | 80 +++++++++ .../endpoint/AbstractPollingEndpoint.java | 38 ++++- .../endpoint/SourcePollingChannelAdapter.java | 32 ++++ .../util/DynamicPeriodicTrigger.java | 161 ++++++++++++++++++ .../endpoint/PollerAdviceTests.java | 130 +++++++++++++- src/reference/docbook/polling-consumer.xml | 72 +++++++- src/reference/docbook/whats-new.xml | 9 + 8 files changed, 581 insertions(+), 6 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aop/AbstractMessageSourceAdvice.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/aop/SimpleActiveIdleMessageSourceAdvice.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/util/DynamicPeriodicTrigger.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 new file mode 100644 index 0000000000..e1934345b4 --- /dev/null +++ b/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); + +} 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 new file mode 100644 index 0000000000..96e40b287e --- /dev/null +++ b/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; + } + +} 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 62e4f69415..cab86ed734 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 @@ -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 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 receiveOnlyAdviceChain = new ArrayList(); + if (!CollectionUtils.isEmpty(adviceChain)) { + for (Advice advice : adviceChain) { + if (isReceiveOnlyAdvice(advice)) { + receiveOnlyAdviceChain.add(advice); + } + } + } Callable pollingTask = new Callable() { @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) proxyFactory.getProxy(this.beanClassLoader); } + if (receiveOnlyAdviceChain.size() > 0) { + applyReceiveOnlyAdviceChain(receiveOnlyAdviceChain); + } return new Poller(pollingTask); } 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 824612855f..cf5cf333af 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 @@ -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 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) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/DynamicPeriodicTrigger.java b/spring-integration-core/src/main/java/org/springframework/integration/util/DynamicPeriodicTrigger.java new file mode 100644 index 0000000000..ce024fe805 --- /dev/null +++ b/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); + } + +} 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 5068457183..66619c34c2 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 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 callOrder = new ArrayList(); + final CountDownLatch latch = new CountDownLatch(4);// advice + advice + source + advice + adapter.setSource(new MessageSource() { + + @Override + public Message 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 adviceChain = new ArrayList(); + + 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 triggerPeriods = new LinkedList(); + final DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(10); + adapter.setSource(new MessageSource() { + + @Override + public Message receive() { + triggerPeriods.add(trigger.getPeriod()); + Message m = null; + if (latch.getCount() % 2 == 0) { + m = new GenericMessage("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)); diff --git a/src/reference/docbook/polling-consumer.xml b/src/reference/docbook/polling-consumer.xml index 176349351b..7fd759d891 100644 --- a/src/reference/docbook/polling-consumer.xml +++ b/src/reference/docbook/polling-consumer.xml @@ -1,7 +1,7 @@
- Poller (Polling Consumer) + Poller (Polling Consumer, Polling Message Source) When Message Endpoints (Channel Adapters) are connected to channels and instantiated, they produce one of the following 2 instances: @@ -66,7 +66,8 @@ of the next poll. The PollSkipAdvice can be used to suppress (skip) a poll, perhaps because there is some downstream condition that would prevent the message to be processed properly. To use this advice, you have to provide it with an implementation - of a PollSkipStrategy. + of a PollSkipStrategy. Version 4.2 added + more flexibility in this area - see . This chapter is meant to only give a high-level overview regarding Polling Consumers @@ -76,5 +77,72 @@ Messaging Endpoints in general and Polling Consumers in particular, please see . +
+ Conditional Pollers + Background + + Advice objects, in an advice-chain on a poller, advise + the whole polling task (message retrieval and processing). These "around advice" objects do not + have access to any context for the poll, just the poll itself. This is fine for requirements + such as making a task transactional, or skipping a poll due to some external condition + as discussed above. 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? + + "Smart" Polling + + Version 4.2 introduced the AbstractMessageSourceAdvice. + Any Advice objects in the advice-chain that + subclass this class, are applied to just the receive operation. Such classes implement the + following methods: + + source)]]> + + This method is called before the MessageSource receive() + method. It enables you to examine and or reconfigure the source at this time. Returning + false cancels this poll (similar to the PollSkipAdvice above). + + 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 (null for no message). + You can even return a different message! + + SimpleActiveIdleMessageSourceAdvice + + This advice is a simple implementation of AbstractMessageSourceAdvice, + 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. + + + 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. + + Advice Chain Ordering + + + 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. + + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index 64dd094767..ac826f7e75 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -118,5 +118,14 @@ See for more information. +
+ Conditional Pollers + + Much more flexibility is now provided for dynamic polling. + + + See for more information. + +