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

# Conflicts:
#	spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java

* Updated Spring AMQP dependency to `2.1.5.BUILD-SNAPSHOT`
* Moved `AmqpOutboundEndpointTests` assertions to `AssertJ` to avoid
conflicts with `master`
This commit is contained in:
Gary Russell
2019-02-22 10:20:51 -05:00
committed by Artem Bilan
parent 40fd8d3254
commit cb1b1c81f1
4 changed files with 70 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -134,7 +134,7 @@ subprojects { subproject ->
romeToolsVersion = '1.9.0'
servletApiVersion = '4.0.0'
smackVersion = '4.3.1'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.1.4.RELEASE'
springAmqpVersion = project.hasProperty('springAmqpVersion') ? project.springAmqpVersion : '2.1.5.BUILD-SNAPSHOT'
springDataJpaVersion = '2.1.5.RELEASE'
springDataMongoVersion = '2.1.5.RELEASE'
springDataRedisVersion = '2.1.5.RELEASE'

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"/>

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2019 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.
@@ -16,11 +16,9 @@
package org.springframework.integration.amqp.outbound;
import static org.hamcrest.Matchers.instanceOf;
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.assertj.core.api.Assertions.assertThat;
import java.util.concurrent.TimeUnit;
import org.junit.Rule;
import org.junit.Test;
@@ -28,6 +26,8 @@ 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;
@@ -108,24 +108,24 @@ public class AmqpOutboundEndpointTests {
.build();
this.pcRequestChannel.send(message);
Message<?> ack = this.ackChannel.receive(10000);
assertNotNull(ack);
assertEquals("foo", ack.getPayload());
assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
assertThat(ack).isNotNull();
assertThat(ack.getPayload()).isEqualTo("foo");
assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
org.springframework.amqp.core.Message received = this.amqpTemplateConfirms.receive(this.queue.getName());
assertEquals("\"hello\"", new String(received.getBody(), "UTF-8"));
assertEquals("application/json", received.getMessageProperties().getContentType());
assertEquals("java.lang.String", received.getMessageProperties().getHeaders()
.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")));
assertThat(new String(received.getBody(), "UTF-8")).isEqualTo("\"hello\"");
assertThat(received.getMessageProperties().getContentType()).isEqualTo("application/json");
assertThat(received.getMessageProperties().getHeaders()
.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))).isEqualTo("java.lang.String");
// test whole message is correlation
message = MessageBuilder.withPayload("hello")
.build();
this.pcMessageCorrelationRequestChannel.send(message);
ack = ackChannel.receive(10000);
assertNotNull(ack);
assertSame(message.getPayload(), ack.getPayload());
assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
assertThat(ack).isNotNull();
assertThat(ack.getPayload()).isSameAs(message.getPayload());
assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
while (this.amqpTemplateConfirms.receive(this.queue.getName()) != null) {
// drain
@@ -139,19 +139,26 @@ public class AmqpOutboundEndpointTests {
.build();
this.pcRequestChannelForAdapter.send(message);
Message<?> ack = this.ackChannel.receive(10000);
assertNotNull(ack);
assertEquals("foo", ack.getPayload());
assertEquals(Boolean.TRUE, ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM));
assertThat(ack).isNotNull();
assertThat(ack.getPayload()).isEqualTo("foo");
assertThat(ack.getHeaders().get(AmqpHeaders.PUBLISH_CONFIRM)).isEqualTo(Boolean.TRUE);
}
@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);
assertNotNull(returned);
assertEquals(message.getPayload(), returned.getPayload());
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
@@ -159,11 +166,11 @@ public class AmqpOutboundEndpointTests {
Message<?> message = MessageBuilder.withPayload("hello").build();
this.returnRequestChannel.send(message);
Message<?> returned = returnChannel.receive(10000);
assertNotNull(returned);
assertThat(returned, instanceOf(ErrorMessage.class));
assertThat(returned.getPayload(), instanceOf(ReturnedAmqpMessageException.class));
assertThat(returned).isNotNull();
assertThat(returned).isInstanceOf(ErrorMessage.class);
assertThat(returned.getPayload()).isInstanceOf(ReturnedAmqpMessageException.class);
ReturnedAmqpMessageException payload = (ReturnedAmqpMessageException) returned.getPayload();
assertEquals(message.getPayload(), payload.getFailedMessage().getPayload());
assertThat(payload.getFailedMessage().getPayload()).isEqualTo(message.getPayload());
}
@Test
@@ -178,18 +185,18 @@ public class AmqpOutboundEndpointTests {
.build();
this.ctRequestChannel.send(message);
org.springframework.amqp.core.Message m = receive(template);
assertNotNull(m);
assertEquals("\"hello\"", new String(m.getBody(), "UTF-8"));
assertEquals("application/json", m.getMessageProperties().getContentType());
assertEquals("java.lang.String",
m.getMessageProperties().getHeaders().get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")));
assertThat(m).isNotNull();
assertThat(new String(m.getBody(), "UTF-8")).isEqualTo("\"hello\"");
assertThat(m.getMessageProperties().getContentType()).isEqualTo("application/json");
assertThat(m.getMessageProperties().getHeaders().get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, "")))
.isEqualTo("java.lang.String");
message = MessageBuilder.withPayload("hello")
.build();
this.ctRequestChannel.send(message);
m = receive(template);
assertNotNull(m);
assertEquals("hello", new String(m.getBody(), "UTF-8"));
assertEquals("text/plain", m.getMessageProperties().getContentType());
assertThat(m).isNotNull();
assertThat(new String(m.getBody(), "UTF-8")).isEqualTo("hello");
assertThat(m.getMessageProperties().getContentType()).isEqualTo("text/plain");
while (template.receive() != null) {
// drain
}
@@ -202,7 +209,7 @@ public class AmqpOutboundEndpointTests {
Thread.sleep(100);
message = template.receive();
}
assertNotNull(message);
assertThat(message).isNotNull();
return message;
}