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

INT-3407: Fix TX-sync-processor potential NPE

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3407
This commit is contained in:
Artem Bilan
2014-05-16 12:44:39 +03:00
parent 11b87034c5
commit 157fc2a084
2 changed files with 115 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java
View File

@@ -1,47 +1,64 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 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
* 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
* 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.transaction;
import java.util.Map.Entry;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.util.Assert;
/**
* This implementation of {@link TransactionSynchronizationFactory}
* allows you to configure SpEL expressions, with their execution being coordinated (synchronized) with a
* transaction - see {@link TransactionSynchronization}. Expressions for before-commit, after-commit, and after-rollback
* are supported, together with a channel for each where the evaluation result (if any) will be sent.
* allows you to configure SpEL expressions, with their execution being coordinated
* (synchronized) with a transaction - see {@link TransactionSynchronization}.
* Expressions for {@code before-commit}, {@code after-commit}, and {@code after-rollback}
* are supported, together with a {@code channel} for each where the evaluation result
* (if any) will be sent.
* <p>
* For each sub-element you can specify 'expression' and/or 'channel' attributes.
* If only the 'channel' attribute is present the received Message will be sent there as part of a particular synchronization scenario.
* If only the 'expression' attribute is present and the result of an expression is a non-Null value, a Message with the
* result as the payload will be generated and sent to a default channel (NullChannel) and will appear in the logs.
* If you want the evaluation result to go to a specific channel add a 'channel' attribute. If the result of an expression is null
* or void, no Message will be generated.
* If only the 'channel' attribute is present the received Message will be sent
* there as part of a particular synchronization scenario.
* <p>
* If only the 'expression' attribute is present and the result of an expression
* is a non-Null value, a Message with the result as the payload will be generated
* and sent to a default channel (NullChannel) and will appear in the logs.
* If you want the evaluation result to go to a specific channel
* add a 'channel' attribute. If the result of an expression is null or void,
* no Message will be generated.
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Artem Bilan
* @since 2.2
*
*/
public class ExpressionEvaluatingTransactionSynchronizationProcessor extends IntegrationObjectSupport implements TransactionSynchronizationProcessor {
public class ExpressionEvaluatingTransactionSynchronizationProcessor extends IntegrationObjectSupport
implements TransactionSynchronizationProcessor, IntegrationEvaluationContextAware {
private volatile StandardEvaluationContext evaluationContext;
private volatile EvaluationContext evaluationContext;
private volatile Expression beforeCommitExpression;
@@ -49,11 +66,16 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
private volatile Expression afterRollbackExpression;
private volatile MessageChannel beforeCommitChannel;
private volatile MessageChannel beforeCommitChannel = new NullChannel();
private volatile MessageChannel afterCommitChannel;
private volatile MessageChannel afterCommitChannel = new NullChannel();
private volatile MessageChannel afterRollbackChannel;
private volatile MessageChannel afterRollbackChannel = new NullChannel();
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
this.evaluationContext = evaluationContext;
}
public void setBeforeCommitChannel(MessageChannel beforeCommitChannel) {
Assert.notNull(beforeCommitChannel, "'beforeCommitChannel' must not be null");
@@ -97,14 +119,16 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
this.doProcess(holder, this.afterRollbackExpression, this.afterRollbackChannel, "afterRollback");
}
private void doProcess(IntegrationResourceHolder holder, Expression expression, MessageChannel messageChannel, String expressionType) {
private void doProcess(IntegrationResourceHolder holder, Expression expression, MessageChannel messageChannel,
String expressionType) {
Message<?> message = holder.getMessage();
if (message != null){
if (expression != null){
if (message != null) {
if (expression != null) {
if (logger.isDebugEnabled()) {
logger.debug("Evaluating " + expressionType + " expression: '" + expression.getExpressionString() + "' on " + message);
logger.debug("Evaluating " + expressionType + " expression: '" + expression.getExpressionString()
+ "' on " + message);
}
StandardEvaluationContext evaluationContextToUse = this.prepareEvaluationContextToUse(holder);
EvaluationContext evaluationContextToUse = this.prepareEvaluationContextToUse(holder);
Object value = expression.getValue(evaluationContextToUse, message);
if (value != null) {
Message<?> spelResultMessage = null;
@@ -124,8 +148,8 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
}
else {
if (logger.isTraceEnabled()) {
logger.trace("Expression evaluation returned null");
}
logger.trace("Expression evaluation returned null");
}
}
}
else {
@@ -136,10 +160,9 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
try {
// rollback will be initiated if any of the previous sync operations fail (e.g., beforeCommit)
// this means that this method will be called without explicit configuration thus no channel
if (messageChannel != null){
this.sendMessage(messageChannel, this.getMessageBuilderFactory().fromMessage(message).build());
}
} catch (Exception e) {
this.sendMessage(messageChannel, this.getMessageBuilderFactory().fromMessage(message).build());
}
catch (Exception e) {
logger.error("Failed to send " + message, e);
}
@@ -147,7 +170,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
}
}
private void sendMessage(MessageChannel channel, Message<?> message){
private void sendMessage(MessageChannel channel, Message<?> message) {
channel.send(message, 0);
}
@@ -157,28 +180,25 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
* @param resource The resource
* @return The context.
*/
private StandardEvaluationContext prepareEvaluationContextToUse(Object resource) {
StandardEvaluationContext evaluationContextToUse;
private EvaluationContext prepareEvaluationContextToUse(Object resource) {
if (resource != null) {
evaluationContextToUse = this.createEvaluationContext();
EvaluationContext evaluationContext = this.createEvaluationContext();
if (resource instanceof IntegrationResourceHolder) {
IntegrationResourceHolder holder = (IntegrationResourceHolder) resource;
for (Entry<String, Object> entry : holder.getAttributes().entrySet()) {
String key = entry.getKey();
evaluationContextToUse.setVariable(key, entry.getValue());
evaluationContext.setVariable(key, entry.getValue());
}
}
return evaluationContext;
}
else {
if (this.evaluationContext == null) {
this.evaluationContext = this.createEvaluationContext();
}
evaluationContextToUse = this.evaluationContext;
return this.evaluationContext;
}
return evaluationContextToUse;
}
protected StandardEvaluationContext createEvaluationContext() {
return ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
}
}

83
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -13,31 +13,37 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.endpoint;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import static org.junit.Assert.*;
import static org.mockito.Mockito.*;
import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
import org.hamcrest.Matchers;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transaction.DefaultTransactionSynchronizationFactory;
import org.springframework.integration.transaction.ExpressionEvaluatingTransactionSynchronizationProcessor;
import org.springframework.integration.transaction.IntegrationResourceHolder;
import org.springframework.integration.transaction.PseudoTransactionManager;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionSynchronization;
@@ -64,7 +70,6 @@ public class PseudoTransactionalMessageSourceTests {
PollableChannel queueChannel = new QueueChannel();
syncProcessor.setBeforeCommitExpression(new SpelExpressionParser().parseExpression("#bix"));
syncProcessor.setBeforeCommitChannel(queueChannel);
syncProcessor.setAfterCommitChannel(queueChannel);
syncProcessor.setAfterCommitExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory =
@@ -78,12 +83,27 @@ public class PseudoTransactionalMessageSourceTests {
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("bix", "qox");
return message;
}
});
MessageChannel afterCommitChannel = TestUtils.getPropertyValue(syncProcessor, "afterCommitChannel",
MessageChannel.class);
assertThat(afterCommitChannel, Matchers.instanceOf(NullChannel.class));
Log logger = TestUtils.getPropertyValue(afterCommitChannel, "logger", Log.class);
logger = Mockito.spy(logger);
Mockito.when(logger.isDebugEnabled()).thenReturn(true);
DirectFieldAccessor dfa = new DirectFieldAccessor(afterCommitChannel);
dfa.setPropertyValue("logger", logger);
TransactionSynchronizationManager.initSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(true);
doPoll(adapter);
@@ -92,9 +112,9 @@ public class PseudoTransactionalMessageSourceTests {
Message<?> beforeCommitMessage = queueChannel.receive(1000);
assertNotNull(beforeCommitMessage);
assertEquals("qox", beforeCommitMessage.getPayload());
Message<?> afterCommitMessage = queueChannel.receive(1000);
assertNotNull(afterCommitMessage);
assertEquals("qux", afterCommitMessage.getPayload());
Mockito.verify(logger).debug(Mockito.anyString());
TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_COMMITTED);
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(false);
@@ -121,7 +141,8 @@ public class PseudoTransactionalMessageSourceTests {
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
return message;
}
});
@@ -164,8 +185,10 @@ public class PseudoTransactionalMessageSourceTests {
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("bix", "qox");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("bix", "qox");
return message;
}
});
@@ -192,13 +215,14 @@ public class PseudoTransactionalMessageSourceTests {
public Object doInTransaction(TransactionStatus status) {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor = new ExpressionEvaluatingTransactionSynchronizationProcessor();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeanFactory(mock(BeanFactory.class));
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory = new DefaultTransactionSynchronizationFactory(
syncProcessor);
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
@@ -236,13 +260,14 @@ public class PseudoTransactionalMessageSourceTests {
public Object doInTransaction(TransactionStatus status) {
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor = new ExpressionEvaluatingTransactionSynchronizationProcessor();
ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
new ExpressionEvaluatingTransactionSynchronizationProcessor();
syncProcessor.setBeanFactory(mock(BeanFactory.class));
syncProcessor.setAfterRollbackChannel(queueChannel);
syncProcessor.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#baz"));
DefaultTransactionSynchronizationFactory syncFactory = new DefaultTransactionSynchronizationFactory(
syncProcessor);
DefaultTransactionSynchronizationFactory syncFactory =
new DefaultTransactionSynchronizationFactory(syncProcessor);
adapter.setTransactionSynchronizationFactory(syncFactory);
@@ -324,9 +349,12 @@ public class PseudoTransactionalMessageSourceTests {
TransactionSynchronizationManager.setActualTransactionActive(false);
assertEquals(1, txSyncCounter.get());
/*TODO: Failed with 'java.lang.IllegalStateException: Already value
TODO: [org.springframework.integration.transaction.IntegrationResourceHolder@46b8c8e6]
TODO: for key [org.springframework.integration.endpoint.PseudoTransactionalMessageSourceTests$8@78a1d1f4] bound to thread [main]'*/
/*
TODO: Failed with 'java.lang.IllegalStateException: Already value
[org.springframework.integration.transaction.IntegrationResourceHolder@46b8c8e6]
for key [org.springframework.integration.endpoint.PseudoTransactionalMessageSourceTests$8@78a1d1f4]
bound to thread [main]'
*/
//TODO: Need new JIRA issue to fix it
TransactionSynchronizationManager.initSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(true);
@@ -349,8 +377,11 @@ public class PseudoTransactionalMessageSourceTests {
}
public class Bar {
public String getValue() {
return "bar";
}
}
}