Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3914: Add CompoundTriggerAdvice

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3914
This commit is contained in:
Gary Russell
2015-12-15 12:04:56 -05:00
committed by Artem Bilan
parent b5ec73b638
commit 758a6bdab1
5 changed files with 306 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

65
spring-integration-core/src/main/java/org/springframework/integration/aop/CompoundTriggerAdvice.java Normal file
View File

@@ -0,0 +1,65 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.aop;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.util.CompoundTrigger;
import org.springframework.messaging.Message;
import org.springframework.scheduling.Trigger;
import org.springframework.util.Assert;
/**
* An {@link AbstractMessageSourceAdvice} that uses a {@link CompoundTrigger} to adjust
* the poller - when a message is present, the compound trigger's primary trigger is
* used to determine the next poll. When no message is present, the override trigger is
* used.
* <p>
* The poller advised by this class must be configured to use the same
* {@link CompoundTrigger} instance and must <b>not</b> use a task executor.
*
* @author Gary Russell
* @since 4.3
*
*/
public class CompoundTriggerAdvice extends AbstractMessageSourceAdvice {
private final CompoundTrigger compoundTrigger;
private final Trigger override;
public CompoundTriggerAdvice(CompoundTrigger compoundTrigger, Trigger overrideTrigger) {
Assert.notNull(compoundTrigger, "'compoundTrigger' cannot be null");
this.compoundTrigger = compoundTrigger;
this.override = overrideTrigger;
}
@Override
public boolean beforeReceive(MessageSource<?> source) {
return true;
}
@Override
public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
if (result == null) {
this.compoundTrigger.setOverride(this.override);
}
else {
this.compoundTrigger.setOverride(null);
}
return result;
}
}

77
spring-integration-core/src/main/java/org/springframework/integration/util/CompoundTrigger.java Normal file
View File

@@ -0,0 +1,77 @@
/*
* Copyright 2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.util;
import java.util.Date;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import reactor.core.support.Assert;
/**
* A {@link Trigger} that delegates the {@link #nextExecutionTime(TriggerContext)}
* to one of two Triggers. If the {@link #setOverride(Trigger) override} trigger is
* {@code null}, the primary trigger is invoked; otherwise the override trigger is
* invoked.
*
* @author Gary Russell
* @since 4.3
*
*/
public class CompoundTrigger implements Trigger {
private volatile Trigger primary;
private volatile Trigger override;
/**
* Construct a compound trigger with the supplied primary trigger.
* @param primary the primary trigger.
*/
public CompoundTrigger(Trigger primary) {
setPrimary(primary);
}
/**
* Set the primary trigger.
* @param primary the trigger.
*/
public final void setPrimary(Trigger primary) {
Assert.notNull(primary, "'primary' cannot be null");
this.primary = primary;
}
/**
* Set the override trigger; set to null to revert to using the
* primary trigger.
* @param override the override trigger, or null.
*/
public void setOverride(Trigger override) {
this.override = override;
}
@Override
public Date nextExecutionTime(TriggerContext triggerContext) {
if (this.override != null) {
return this.override.nextExecutionTime(triggerContext);
}
else {
return this.primary.nextExecutionTime(triggerContext);
}
}
}

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

@@ -18,9 +18,14 @@ package org.springframework.integration.endpoint;
import static org.hamcrest.Matchers.contains;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.atLeast;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import java.util.ArrayList;
import java.util.Collections;
@@ -33,23 +38,27 @@ import 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.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.aop.AbstractMessageSourceAdvice;
import org.springframework.integration.aop.CompoundTriggerAdvice;
import org.springframework.integration.aop.SimpleActiveIdleMessageSourceAdvice;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.scheduling.PollSkipAdvice;
import org.springframework.integration.scheduling.PollSkipStrategy;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.CompoundTrigger;
import org.springframework.integration.util.DynamicPeriodicTrigger;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
* @author Gary Russell
@@ -221,11 +230,11 @@ public class PollerAdviceTests {
return m;
}
});
QueueChannel channel = new QueueChannel();
SimpleActiveIdleMessageSourceAdvice toggling = new SimpleActiveIdleMessageSourceAdvice(trigger);
toggling.setActivePollPeriod(11);
toggling.setIdlePollPeriod(12);
adapter.setAdviceChain(Collections.singletonList(toggling));
adapter.setTrigger(trigger);
configure(adapter);
adapter.afterPropertiesSet();
adapter.start();
@@ -234,7 +243,42 @@ public class PollerAdviceTests {
while (triggerPeriods.size() > 5) {
triggerPeriods.removeLast();
}
assertThat(triggerPeriods, Matchers.contains(10L, 12L, 11L, 12L, 11L));
assertThat(triggerPeriods, contains(10L, 12L, 11L, 12L, 11L));
}
@Test
public void testCompoundTriggerAdvice() throws Exception {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
final CountDownLatch latch = new CountDownLatch(5);
final LinkedList<Object> overridePresent = new LinkedList<Object>();
final CompoundTrigger compoundTrigger = new CompoundTrigger(new PeriodicTrigger(10));
Trigger override = spy(new PeriodicTrigger(5));
final CompoundTriggerAdvice advice = new CompoundTriggerAdvice(compoundTrigger, override);
adapter.setSource(new MessageSource<Object>() {
@Override
public Message<Object> receive() {
overridePresent.add(TestUtils.getPropertyValue(compoundTrigger, "override"));
Message<Object> m = null;
if (latch.getCount() % 2 == 0) {
m = new GenericMessage<Object>("foo");
}
latch.countDown();
return m;
}
});
adapter.setAdviceChain(Collections.singletonList(advice));
adapter.setTrigger(compoundTrigger);
configure(adapter);
adapter.afterPropertiesSet();
adapter.start();
assertTrue(latch.await(10, TimeUnit.SECONDS));
adapter.stop();
while (overridePresent.size() > 5) {
overridePresent.removeLast();
}
assertThat(overridePresent, contains(null, override, null, override, null));
verify(override, atLeast(2)).nextExecutionTime(any(TriggerContext.class));
}
private void configure(SourcePollingChannelAdapter adapter) {
@@ -245,4 +289,29 @@ public class PollerAdviceTests {
adapter.setTaskScheduler(scheduler);
}
@Test
public void testCompoundAdviceXML() throws Exception {
ConfigurableApplicationContext ctx = new ClassPathXmlApplicationContext("compound-trigger-context.xml",
getClass());
SourcePollingChannelAdapter adapter = ctx.getBean(SourcePollingChannelAdapter.class);
Source source = ctx.getBean(Source.class);
adapter.start();
assertTrue(source.latch.await(10, TimeUnit.SECONDS));
assertNotNull(TestUtils.getPropertyValue(adapter, "trigger.override"));
adapter.stop();
ctx.close();
}
public static class Source implements MessageSource<Object> {
private final CountDownLatch latch = new CountDownLatch(5);
@Override
public Message<Object> receive() {
latch.countDown();
return null;
}
}
}

32
spring-integration-core/src/test/java/org/springframework/integration/endpoint/compound-trigger-context.xml Normal file
View File

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

68
src/reference/asciidoc/polling-consumer.adoc
View File

@@ -84,6 +84,16 @@ Message<?> afterReceive(Message<?> result, MessageSource<?> source)
This method is called after the `receive()` method; again, you can reconfigure the source, or take any action perhaps depending on the result (which can be `null` if there was no message created by the source).
You can even return a different message!
.Advice Chain Ordering
[IMPORTANT]
=====
It is important to understand how the advice chain is processed during initialization.
`Advice` objects that do not extend `AbstractMessageSourceAdvice` are applied to the whole poll process and are all invoked first, in order, before any `AbstractMessageSourceAdvice`; then `AbstractMessageSourceAdvice` objects are invoked in order around the `MessageSource` `receive()` method.
If you have, say `Advice` objects `a, b, c, d`, where `b` and `d` are `AbstractMessageSourceAdvice`, they will be applied in the order `a, c, b, d`.
Also, if a `MessageSource` is already a `Proxy`, the `AbstractMessageSourceAdvice` will be invoked after any existing `Advice` objects.
If you wish to change the order, you should wire up the proxy yourself.
=====
===== SimpleActiveIdleMessageSourceAdvice
This advice is a simple implementation of `AbstractMessageSourceAdvice`, when used in conjunction with a `DynamicPeriodicTrigger`, it adjuststhe polling frequency depending on whether or not the previous poll resulted in a message or not.
@@ -95,12 +105,52 @@ This will only work if the advice is called on the poller thread.
It will *not* work if the poller has a `task-executor`.
To use this advice where you wish to use async operations after the result of a poll, do the async handoff later, perhaps by using an `ExecutorChannel`.
.Advice Chain Ordering
[IMPORTANT]
=====
It is important to understand how the advice chain is processed during initialization.
`Advice` objects that do not extend `AbstractMessageSourceAdvice` are applied to the whole poll process and are all invoked first, in order, before any `AbstractMessageSourceAdvice`; then `AbstractMessageSourceAdvice` objects are invoked in order around the `MessageSource` `receive()` method.
If you have, say `Advice` objects `a, b, c, d`, where `b` and `d` are `AbstractMessageSourceAdvice`, they will be applied in the order `a, c, b, d`.
Also, if a `MessageSource` is already a `Proxy`, the `AbstractMessageSourceAdvice` will be invoked after any existing `Advice` objects.
If you wish to change the order, you should wire up the proxy yourself.
=====
===== CompoundTriggerAdvice
This advice allows the selection of one of two triggers based on whether a poll returns a message or not.
Consider a poller that uses a `CronTrigger`; `CronTrigger` s are immutable so cannot be altered once constructed.
Consider a use case where we want to use a cron expression to trigger a poll once each hour but, if no message is
received, poll once per minute and, when a message is retrieved, revert to using the cron expression.
The advice (and poller) use a `CompoundTrigger` for this purpose.
The trigger's `primary` trigger can be a `CronTrigger`.
When the advice detects that no message is received, it adds the secondary trigger to the `CompoundTrigger`.
When the `CompoundTrigger` 's `nextExecutionTime` method is invoked, it will delegate to the secondary trigger, if
present; otherwise the primary trigger.
The poller must also have a reference to the same `CompoundTrigger`.
The following shows the configuration for the hourly cron expression with fall-back to every minute...
[source, xml]
----
<int:inbound-channel-adapter channel="nullChannel" auto-startup="false">
<bean class="org.springframework.integration.endpoint.PollerAdviceTests.Source" />
<int:poller trigger="compoundTrigger">
<int:advice-chain>
<bean class="org.springframework.integration.aop.CompoundTriggerAdvice">
<constructor-arg ref="compoundTrigger"/>
<constructor-arg ref="secondary"/>
</bean>
</int:advice-chain>
</int:poller>
</int:inbound-channel-adapter>
<bean id="compoundTrigger" class="org.springframework.integration.util.CompoundTrigger">
<constructor-arg ref="primary" />
</bean>
<bean id="primary" class="org.springframework.scheduling.support.CronTrigger">
<constructor-arg value="0 0 * * * *" /> <!-- top of every hour -->
</bean>
<bean id="secondary" class="org.springframework.scheduling.support.PeriodicTrigger">
<constructor-arg value="60000" />
</bean>
----
.Important: Async Handoff
IMPORTANT: This advice modifies the trigger based on the `receive()` result.
This will only work if the advice is called on the poller thread.
It will *not* work if the poller has a `task-executor`.
To use this advice where you wish to use async operations after the result of a poll, do the async handoff later, perhaps by using an `ExecutorChannel`.