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

INT-4163: UnProxy MessageSource for TX Resource

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4163

When `MessageSource` is proxy, the `TransactionSynchronizationManager.getResource(this)` logic in the `MessageSource` doesn't work,
because TX resource is bound to the `Proxy` in the `SourcePollingChannelAdapter`

* Introduce `SourcePollingChannelAdapter.originalSource` property and store there a target `MessageSource` object extracted from the AOP Proxy
* Use `originalSource` as a resource to bind to the TX
* Modify `MongoDbInboundChannelAdapterIntegrationTests` to ensure that `AbstractMessageSourceAdvice` proxying the `MessageSource` doesn't effect `TransactionSynchronizationManager.getResource(this)` logic
* Refactor for some MongoDb test to rely on the `@RunWith(SpringJUnit4ClassRunner.class)` for context loading for better test class performance

**Cherry-pick to 4.3.x, 4.2.x**

Fallback to provided source if `target` from Proxy is `null`

Fix [UnusedImport] issue
This commit is contained in:
Artem Bilan
2016-11-07 12:41:03 -05:00
committed by Gary Russell
parent acf7424dbc
commit e2d18054e6
6 changed files with 296 additions and 210 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -25,6 +25,7 @@ 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.beans.factory.BeanCreationException;
import org.springframework.context.Lifecycle;
import org.springframework.integration.aop.AbstractMessageSourceAdvice;
import org.springframework.integration.context.ExpressionCapable;
@@ -55,6 +56,8 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
private final Collection<Advice> appliedAdvices = new HashSet<Advice>();
private volatile MessageSource<?> originalSource;
private volatile MessageSource<?> source;
private volatile MessageChannel outputChannel;
@@ -70,6 +73,10 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
*/
public void setSource(MessageSource<?> source) {
this.source = source;
Object target = extractProxyTarget(source);
this.originalSource = target != null ? (MessageSource<?>) target : source;
if (source instanceof ExpressionCapable) {
setPrimaryExpression(((ExpressionCapable) source).getExpression());
}
@@ -219,7 +226,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
@Override
protected Object getResourceToBind() {
return this.source;
return this.originalSource;
}
@Override
@@ -227,4 +234,21 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
return IntegrationResourceHolder.MESSAGE_SOURCE;
}
private static Object extractProxyTarget(Object target) {
if (!(target instanceof Advised)) {
return target;
}
Advised advised = (Advised) target;
if (advised.getTargetSource() == null) {
return null;
}
try {
return extractProxyTarget(advised.getTargetSource().getTarget());
}
catch (Exception e) {
throw new BeanCreationException("Could not extract target", e);
}
}
}