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

GH-2759: Fix CorrelationData.future

Browse Source 
* GH-2759: Fix CorrelationData.future

Fixes https://github.com/spring-projects/spring-integration/issues/2759

The outbound endpoints wrap user correlation data in a wrapper.
If the user data is a `CorrelationData`, we must delegate methods
involving the `Future<?>` and `returnedMessage` to the user data.

**cherry-pick to 5.1 and switch AMQP to snapshots**

* Polishing - remove redundant override.

* Add debug log with null correlation data
This commit is contained in:
Gary Russell
2019-02-22 10:20:51 -05:00
committed by Artem Bilan
parent 221393e02f
commit d6ac866871
3 changed files with 39 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
View File

@@ -44,6 +44,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.SettableListenableFuture;
/**
* @author Gary Russell
@@ -479,8 +480,14 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
if (messageId == null) {
messageId = NO_ID;
}
correlationData = new CorrelationDataWrapper(messageId.toString(),
this.correlationDataGenerator.processMessage(requestMessage), requestMessage);
Object userData = this.correlationDataGenerator.processMessage(requestMessage);
if (userData != null) {
correlationData = new CorrelationDataWrapper(messageId.toString(), userData, requestMessage);
}
else {
this.logger.debug("'confirmCorrelationExpression' resolved to 'null'; "
+ "no publisher confirm will be sent to the ack or nack channel");
}
}
return correlationData;
}
@@ -604,6 +611,23 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
return this.message;
}
@Override
public SettableListenableFuture<Confirm> getFuture() {
if (this.userData instanceof CorrelationData) {
return ((CorrelationData) this.userData).getFuture();
}
else {
return super.getFuture();
}
}
@Override
public void setReturnedMessage(org.springframework.amqp.core.Message returnedMessage) {
if (this.userData instanceof CorrelationData) {
((CorrelationData) this.userData).setReturnedMessage(returnedMessage);
}
super.setReturnedMessage(returnedMessage);
}
}
}

1
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
View File

@@ -70,6 +70,7 @@
routing-key="#{queue.name + queue.name}"
mapped-request-headers="foo*"
amqp-template="amqpTemplateReturns"
confirm-correlation-expression="headers['corrData']"
return-channel="returnChannel" />
<int:channel id="returnRequestChannel"/>

13
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
View File

@@ -18,12 +18,16 @@ package org.springframework.integration.amqp.outbound;
import static org.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.TimeUnit;
import org.junit.Rule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.junit.BrokerRunning;
import org.springframework.amqp.support.AmqpHeaders;
@@ -143,11 +147,18 @@ public class AmqpOutboundEndpointTests {
@Test
public void adapterWithReturns() throws Exception {
this.withReturns.setErrorMessageStrategy(null);
Message<?> message = MessageBuilder.withPayload("hello").build();
CorrelationData corrData = new CorrelationData("adapterWithReturns");
Message<?> message = MessageBuilder.withPayload("hello")
.setHeader("corrData", corrData)
.build();
this.returnRequestChannel.send(message);
Message<?> returned = returnChannel.receive(10000);
assertThat(returned).isNotNull();
assertThat(returned.getPayload()).isEqualTo(message.getPayload());
Confirm confirm = corrData.getFuture().get(10, TimeUnit.SECONDS);
assertThat(confirm).isNotNull();
assertThat(confirm.isAck()).isTrue();
assertThat(corrData.getReturnedMessage()).isNotNull();
}
@Test