From aafeb3a55f79aa724f38b64601d8a0a77aed5f10 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 23 Sep 2016 13:16:11 -0400 Subject: [PATCH] INT-4120: Poller Advice Chain Regression JIRA: https://jira.spring.io/browse/INT-4120 INT-3899 moved creating the pollingTask to start() to allow modification of the advice chain between starts. Since a new pollingTask is created on each start, this was fine for general advices. However, for `AbstractMessageSourceAdvice`s, which only advise the `receive()` operation, the advices are re-applied on every start. When starting, first remove any advices we added on the previous start. This handles the case when the source is a naked object, or already a proxy when supplied to the SPCA. Remove uneeded cast Polishing - PR Comments --- .../endpoint/SourcePollingChannelAdapter.java | 20 +++-- .../endpoint/PollerAdviceTests.java | 74 ++++++++++++++----- 2 files changed, 68 insertions(+), 26 deletions(-) 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 8df0679779..56fe7dbb81 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 @@ -17,6 +17,7 @@ package org.springframework.integration.endpoint; import java.util.Collection; +import java.util.HashSet; import org.aopalliance.aop.Advice; @@ -50,6 +51,10 @@ import org.springframework.util.Assert; public class SourcePollingChannelAdapter extends AbstractPollingEndpoint implements TrackableComponent { + private final MessagingTemplate messagingTemplate = new MessagingTemplate(); + + private final Collection appliedAdvices = new HashSet(); + private volatile MessageSource source; private volatile MessageChannel outputChannel; @@ -58,8 +63,6 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint private volatile boolean shouldTrack; - private final MessagingTemplate messagingTemplate = new MessagingTemplate(); - /** * Specify the source to be polled for Messages. * @@ -129,10 +132,13 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint @Override protected void applyReceiveOnlyAdviceChain(Collection chain) { if (AopUtils.isAopProxy(this.source)) { + for (Advice advice : this.appliedAdvices) { + ((Advised) this.source).removeAdvice(advice); + } for (Advice advice : chain) { - NameMatchMethodPointcutAdvisor sourceAdvice = new NameMatchMethodPointcutAdvisor(advice); - sourceAdvice.addMethodName("receive"); - ((Advised) this.source).addAdvice(advice); + NameMatchMethodPointcutAdvisor sourceAdvisor = new NameMatchMethodPointcutAdvisor(advice); + sourceAdvisor.addMethodName("receive"); + ((Advised) this.source).addAdvisor(sourceAdvisor); } } else { @@ -142,14 +148,16 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint } this.source = (MessageSource) proxyFactory.getProxy(getBeanClassLoader()); } + this.appliedAdvices.clear(); + this.appliedAdvices.addAll(chain); } @Override protected void doStart() { + super.doStart(); if (this.source instanceof Lifecycle) { ((Lifecycle) this.source).start(); } - super.doStart(); } 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 e072756597..64a4f6ffd2 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 @@ -17,6 +17,7 @@ package org.springframework.integration.endpoint; import static org.hamcrest.Matchers.contains; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; @@ -34,6 +35,8 @@ import java.util.LinkedList; import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.aop.Advice; import org.aopalliance.intercept.MethodInterceptor; @@ -41,6 +44,9 @@ import org.aopalliance.intercept.MethodInvocation; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.aop.Advisor; +import org.springframework.aop.framework.Advised; +import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ConfigurableApplicationContext; @@ -58,6 +64,7 @@ import org.springframework.integration.config.ExpressionControlBusFactoryBean; import org.springframework.integration.core.MessageSource; import org.springframework.integration.scheduling.PollSkipAdvice; import org.springframework.integration.scheduling.SimplePollSkipStrategy; +import org.springframework.integration.test.util.OnlyOnceTrigger; import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.util.CompoundTrigger; import org.springframework.integration.util.DynamicPeriodicTrigger; @@ -188,27 +195,18 @@ public class PollerAdviceTests { 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() { - + final AtomicReference latch = new AtomicReference<>(new CountDownLatch(4)); + MessageSource source = new MessageSource() { @Override public Message receive() { callOrder.add("c"); - latch.countDown(); + latch.get().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; - } - }); + }; + adapter.setSource(source); + OnlyOnceTrigger trigger = new OnlyOnceTrigger(); + adapter.setTrigger(trigger); configure(adapter); List adviceChain = new ArrayList(); @@ -217,26 +215,28 @@ public class PollerAdviceTests { @Override public Object invoke(MethodInvocation invocation) throws Throwable { callOrder.add("a"); - latch.countDown(); + latch.get().countDown(); return invocation.proceed(); } } adviceChain.add(new TestGeneralAdvice()); + final AtomicInteger count = new AtomicInteger(); class TestSourceAdvice extends AbstractMessageSourceAdvice { @Override public boolean beforeReceive(MessageSource target) { + count.incrementAndGet(); callOrder.add("b"); - latch.countDown(); + latch.get().countDown(); return true; } @Override public Message afterReceive(Message result, MessageSource target) { callOrder.add("d"); - latch.countDown(); + latch.get().countDown(); return result; } @@ -246,9 +246,43 @@ public class PollerAdviceTests { adapter.setAdviceChain(adviceChain); adapter.afterPropertiesSet(); adapter.start(); - assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertTrue(latch.get().await(10, TimeUnit.SECONDS)); assertThat(callOrder, contains("a", "b", "c", "d")); // advice + advice + source + advice adapter.stop(); + trigger.reset(); + latch.set(new CountDownLatch(4)); + adapter.start(); + assertTrue(latch.get().await(10, TimeUnit.SECONDS)); + adapter.stop(); + assertEquals(2, count.get()); + + // Now test when the source is already a proxy. + + MethodInterceptor dummy = new MethodInterceptor() { + @Override + public Object invoke(MethodInvocation invocation) throws Throwable { + return invocation.proceed(); + } + }; + ProxyFactory pf = new ProxyFactory(source); + pf.addAdvice(dummy); + adapter.setSource((MessageSource) pf.getProxy()); + trigger.reset(); + latch.set(new CountDownLatch(4)); + count.set(0); + callOrder.clear(); + adapter.start(); + assertTrue(latch.get().await(10, TimeUnit.SECONDS)); + assertThat(callOrder, contains("a", "b", "c", "d")); // advice + advice + source + advice + adapter.stop(); + trigger.reset(); + latch.set(new CountDownLatch(4)); + adapter.start(); + assertTrue(latch.get().await(10, TimeUnit.SECONDS)); + adapter.stop(); + assertEquals(2, count.get()); + Advisor[] advisors = ((Advised) adapter.getMessageSource()).getAdvisors(); + assertEquals(2, advisors.length); // make sure we didn't remove the original one } @Test