From c49d8f4af70b665611654fdde605424fd2ca28c2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 16 May 2014 12:44:39 +0300 Subject: [PATCH] INT-3407: Fix TX-sync-processor potential NPE JIRA: https://jira.spring.io/browse/INT-3407 Conflicts: spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java Resolved. Fixed imports and reference to MessageBuilderFactory. --- ...ngTransactionSynchronizationProcessor.java | 99 ++++++++++++------- ...PseudoTransactionalMessageSourceTests.java | 83 ++++++++++++---- 2 files changed, 125 insertions(+), 57 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java index 4a1101f747..f503bf9166 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java @@ -1,48 +1,65 @@ /* - * 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.Message; import org.springframework.integration.MessageChannel; +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.integration.support.MessageBuilder; 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. + *

* 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. + *

+ * 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; @@ -50,11 +67,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"); @@ -86,26 +108,31 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int this.afterRollbackExpression = afterRollbackExpression; } + @Override public void processBeforeCommit(IntegrationResourceHolder holder) { this.doProcess(holder, this.beforeCommitExpression, this.beforeCommitChannel, "beforeCommit"); } + @Override public void processAfterCommit(IntegrationResourceHolder holder) { this.doProcess(holder, this.afterCommitExpression, this.afterCommitChannel, "afterCommit"); } + @Override public void processAfterRollback(IntegrationResourceHolder holder) { 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; @@ -125,8 +152,8 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int } else { if (logger.isTraceEnabled()) { - logger.trace("Expression evaluation returned null"); - } + logger.trace("Expression evaluation returned null"); + } } } else { @@ -137,10 +164,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, MessageBuilder.fromMessage(message).build()); - } - } catch (Exception e) { + this.sendMessage(messageChannel, MessageBuilder.fromMessage(message).build()); + } + catch (Exception e) { logger.error("Failed to send " + message, e); } @@ -148,7 +174,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int } } - private void sendMessage(MessageChannel channel, Message message){ + private void sendMessage(MessageChannel channel, Message message) { channel.send(message, 0); } @@ -158,28 +184,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 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()); } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java index d3aed36e8e..d73353ce05 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java @@ -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,26 +13,35 @@ * 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.assertThat; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; 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.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.NullChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.message.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; @@ -64,7 +73,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 = @@ -76,14 +84,30 @@ public class PseudoTransactionalMessageSourceTests { adapter.setOutputChannel(outputChannel); adapter.setSource(new MessageSource() { + @Override public Message receive() { GenericMessage message = new GenericMessage("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 +116,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); @@ -119,9 +143,11 @@ public class PseudoTransactionalMessageSourceTests { adapter.setOutputChannel(outputChannel); adapter.setSource(new MessageSource() { + @Override public Message receive() { GenericMessage message = new GenericMessage("foo"); - ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)).addAttribute("baz", "qux"); + ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)) + .addAttribute("baz", "qux"); return message; } }); @@ -143,6 +169,7 @@ public class PseudoTransactionalMessageSourceTests { TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager()); transactionTemplate.execute(new TransactionCallback() { + @Override public Object doInTransaction(TransactionStatus status) { SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor = @@ -162,10 +189,13 @@ public class PseudoTransactionalMessageSourceTests { adapter.setOutputChannel(outputChannel); adapter.setSource(new MessageSource() { + @Override public Message receive() { GenericMessage message = new GenericMessage("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; } }); @@ -189,16 +219,18 @@ public class PseudoTransactionalMessageSourceTests { try { transactionTemplate.execute(new TransactionCallback() { + @Override 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); @@ -206,6 +238,7 @@ public class PseudoTransactionalMessageSourceTests { adapter.setOutputChannel(outputChannel); adapter.setSource(new MessageSource() { + @Override public Message receive() { GenericMessage message = new GenericMessage("foo"); ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)) @@ -233,16 +266,18 @@ public class PseudoTransactionalMessageSourceTests { TransactionTemplate transactionTemplate = new TransactionTemplate(new PseudoTransactionManager()); transactionTemplate.execute(new TransactionCallback() { + @Override 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); @@ -250,6 +285,7 @@ public class PseudoTransactionalMessageSourceTests { adapter.setOutputChannel(outputChannel); adapter.setSource(new MessageSource() { + @Override public Message receive() { GenericMessage message = new GenericMessage("foo"); ((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this)) @@ -274,6 +310,7 @@ public class PseudoTransactionalMessageSourceTests { adapter.setSource(new MessageSource() { + @Override public Message receive() { return null; } @@ -298,6 +335,7 @@ public class PseudoTransactionalMessageSourceTests { TransactionSynchronizationFactory syncFactory = new TransactionSynchronizationFactory() { + @Override public TransactionSynchronization create(Object key) { return new TransactionSynchronizationAdapter() { @Override @@ -311,6 +349,7 @@ public class PseudoTransactionalMessageSourceTests { adapter.setTransactionSynchronizationFactory(syncFactory); adapter.setSource(new MessageSource() { + @Override public Message receive() { return null; } @@ -324,9 +363,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 +391,11 @@ public class PseudoTransactionalMessageSourceTests { } public class Bar { + public String getValue() { return "bar"; } + } + }