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 f7a328b8f9..4a9f111b2c 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,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. + *

* 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; @@ -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 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 db8991a1ef..21f447859a 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,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 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 +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 receive() { GenericMessage message = new GenericMessage("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 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; } }); @@ -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"; } + } + }