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

Don't wrap SpEL result message to another message

Browse Source 
The `ExpressionEvaluatingTransactionSynchronizationProcessor` wraps
an expression evaluation result to the `Message` unconditionally

* Don't wrap one message to another if SpEL result is a `Message` per se
* Don't wrap the received message to another if the `expression` isn't
configured.

**Cherry-pick to 4.3.x**
This commit is contained in:
Artem Bilan
2017-11-21 12:12:38 -05:00
committed by Gary Russell
parent 28f1769899
commit c3ca61d4a0
2 changed files with 31 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -51,6 +51,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Artem Bilan
*
* @since 2.2
*
*/
@@ -114,19 +115,20 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
}
public void processBeforeCommit(IntegrationResourceHolder holder) {
this.doProcess(holder, this.beforeCommitExpression, this.beforeCommitChannel, "beforeCommit");
doProcess(holder, this.beforeCommitExpression, this.beforeCommitChannel, "beforeCommit");
}
public void processAfterCommit(IntegrationResourceHolder holder) {
this.doProcess(holder, this.afterCommitExpression, this.afterCommitChannel, "afterCommit");
doProcess(holder, this.afterCommitExpression, this.afterCommitChannel, "afterCommit");
}
public void processAfterRollback(IntegrationResourceHolder holder) {
this.doProcess(holder, this.afterRollbackExpression, this.afterRollbackChannel, "afterRollback");
doProcess(holder, this.afterRollbackExpression, this.afterRollbackChannel, "afterRollback");
}
private void doProcess(IntegrationResourceHolder holder, Expression expression, MessageChannel messageChannel,
String expressionType) {
Message<?> message = holder.getMessage();
if (message != null) {
if (expression != null) {
@@ -134,19 +136,27 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
logger.debug("Evaluating " + expressionType + " expression: '" + expression.getExpressionString()
+ "' on " + message);
}
EvaluationContext evaluationContextToUse = this.prepareEvaluationContextToUse(holder);
EvaluationContext evaluationContextToUse = prepareEvaluationContextToUse(holder);
Object value = expression.getValue(evaluationContextToUse, message);
if (value != null) {
Message<?> spelResultMessage = null;
if (logger.isDebugEnabled()) {
logger.debug("Sending expression result message to " + messageChannel + " " +
"as part of '" + expressionType + "' transaction synchronization");
}
Message<?> spelResultMessage = null;
try {
spelResultMessage = this.getMessageBuilderFactory().withPayload(value)
.copyHeaders(message.getHeaders())
.build();
this.sendMessage(messageChannel, spelResultMessage);
if (value instanceof Message<?>) {
spelResultMessage = (Message<?>) value;
}
else {
spelResultMessage =
getMessageBuilderFactory()
.withPayload(value)
.copyHeaders(message.getHeaders())
.build();
}
sendMessage(messageChannel, spelResultMessage);
}
catch (Exception e) {
logger.error("Failed to send " + expressionType + " evaluation result " + spelResultMessage, e);
@@ -166,7 +176,7 @@ 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
this.sendMessage(messageChannel, this.getMessageBuilderFactory().fromMessage(message).build());
sendMessage(messageChannel, message);
}
catch (Exception e) {
logger.error("Failed to send " + message, e);
@@ -188,7 +198,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
*/
private EvaluationContext prepareEvaluationContextToUse(Object resource) {
if (resource != null) {
EvaluationContext evaluationContext = this.createEvaluationContext();
EvaluationContext evaluationContext = createEvaluationContext();
if (resource instanceof IntegrationResourceHolder) {
IntegrationResourceHolder holder = (IntegrationResourceHolder) resource;
for (Entry<String, Object> entry : holder.getAttributes().entrySet()) {
@@ -204,7 +214,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
}
protected StandardEvaluationContext createEvaluationContext() {
return ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
return ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.endpoint;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
@@ -62,6 +63,7 @@ import org.springframework.transaction.support.TransactionTemplate;
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Artem Bilan
*
* @since 2.2
*
*/
@@ -189,13 +191,16 @@ public class PseudoTransactionalMessageSourceTests {
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
final Message<?> testMessage = new GenericMessage<>("foo");
adapter.setSource(new MessageSource<String>() {
@Override
public Message<String> receive() {
GenericMessage<String> message = new GenericMessage<String>("foo");
((IntegrationResourceHolder) TransactionSynchronizationManager.getResource(this))
.addAttribute("baz", "qux");
.addAttribute("baz", testMessage);
return message;
}
});
@@ -206,7 +211,7 @@ public class PseudoTransactionalMessageSourceTests {
TransactionSynchronizationUtils.triggerAfterCompletion(TransactionSynchronization.STATUS_ROLLED_BACK);
Message<?> rollbackMessage = queueChannel.receive(1000);
assertNotNull(rollbackMessage);
assertEquals("qux", rollbackMessage.getPayload());
assertSame(testMessage, rollbackMessage);
TransactionSynchronizationManager.clearSynchronization();
TransactionSynchronizationManager.setActualTransactionActive(false);
}