From 2aadf5ee3dae903b6b5783c018d99d78348a9a2c Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Mon, 7 Nov 2016 10:02:39 -0500
Subject: [PATCH] Fix `receiveOnlyAdviceChain` condition

https://build.spring.io/browse/INT-MASTER-415/

The `Stream` `Collectors.toList()` returns empty list if nothing pass the `.filter()`, therefore condition as `if (receiveOnlyAdviceChain != null)` is not enough
and since `SourcePollingChannelAdapter.applyReceiveOnlyAdviceChain()` doesn't have conditions as well, the target `MessageSource` is proxyed for nothing.
When `TransactionSynchronizationManager.getResource(this)` is called for the `MessageSource` it can't find it because the proxy doesn't match an original object.

* Make condition as `if (!CollectionUtils.isEmpty(receiveOnlyAdviceChain))` in the `AbstractPollingEndpoint` and `SourcePollingChannelAdapter`
* Increase group removal wait timeout in the `gemfire.DelayerHandlerRescheduleIntegrationTests`
---
 .../endpoint/AbstractPollingEndpoint.java     |  2 +-
 .../endpoint/SourcePollingChannelAdapter.java | 29 ++++++++++---------
 ...ayerHandlerRescheduleIntegrationTests.java |  2 +-
 3 files changed, 18 insertions(+), 15 deletions(-)

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 6ca8e29e5e..683b1fb635 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
@@ -193,7 +193,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
 			}
 			pollingTask = (Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader);
 		}
-		if (receiveOnlyAdviceChain != null) {
+		if (!CollectionUtils.isEmpty(receiveOnlyAdviceChain)) {
 			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 65b93a8d58..86dc8d8c95 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
@@ -38,6 +38,7 @@ 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
@@ -131,22 +132,24 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
 
 	@Override
 	protected void applyReceiveOnlyAdviceChain(Collection<Advice> chain) {
-		if (AopUtils.isAopProxy(this.source)) {
-			Advised source = (Advised) this.source;
-			this.appliedAdvices.forEach(source::removeAdvice);
-			for (Advice advice : chain) {
-				source.addAdvisor(adviceToReceiveAdvisor(advice));
+		if (!CollectionUtils.isEmpty(chain)) {
+			if (AopUtils.isAopProxy(this.source)) {
+				Advised source = (Advised) this.source;
+				this.appliedAdvices.forEach(source::removeAdvice);
+				for (Advice advice : chain) {
+					source.addAdvisor(adviceToReceiveAdvisor(advice));
+				}
 			}
-		}
-		else {
-			ProxyFactory proxyFactory = new ProxyFactory(this.source);
-			for (Advice advice : chain) {
-				proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice));
+			else {
+				ProxyFactory proxyFactory = new ProxyFactory(this.source);
+				for (Advice advice : chain) {
+					proxyFactory.addAdvisor(adviceToReceiveAdvisor(advice));
+				}
+				this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
 			}
-			this.source = (MessageSource<?>) proxyFactory.getProxy(getBeanClassLoader());
+			this.appliedAdvices.clear();
+			this.appliedAdvices.addAll(chain);
 		}
-		this.appliedAdvices.clear();
-		this.appliedAdvices.addAll(chain);
 	}
 
 	private NameMatchMethodPointcutAdvisor adviceToReceiveAdvisor(Advice advice) {
diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java
index 8fcd0fb443..e32dce6158 100644
--- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java
+++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/DelayerHandlerRescheduleIntegrationTests.java
@@ -140,7 +140,7 @@ public class DelayerHandlerRescheduleIntegrationTests {
 		assertEquals(1, messageStore.getMessageGroupCount());
 		int n = 0;
 		while (n++ < 200 && messageStore.messageGroupSize(delayerMessageGroupId) > 0) {
-			Thread.sleep(50);
+			Thread.sleep(100);
 		}
 		assertEquals(0, messageStore.messageGroupSize(delayerMessageGroupId));