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`
This commit is contained in:
@@ -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);
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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));
|
||||
|
||||
|
||||
Reference in New Issue
Block a user