Skip to content

GH-2759: Fix CorrelationData.future #2761

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Feb 22, 2019
Merged

GH-2759: Fix CorrelationData.future #2761

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7262144
GH-2759: Fix CorrelationData.future
garyrussell Feb 21, 2019
ffbd329
Polishing - remove redundant override.
garyrussell Feb 22, 2019
c286fd4
Add debug log with null correlation data
garyrussell Feb 22, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 26 additions & 2 deletions ...main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.SettableListenableFuture;

/**
* @author Gary Russell
Expand Down Expand Up @@ -479,8 +480,14 @@ protected CorrelationData generateCorrelationData(Message<?> requestMessage) {
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);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like previously we have had a CorrelationDataWrapper instance independently of the userData, but now you restrict it only if that one is not null.

Also I see that you are protected against null in the CorrelationDataWrapper any way.

So, what is the motivation to not do that any more?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because we get an NPE in handleConfirm() if the expression resolves to null (bug - try to create a message with a null payload).

I should have mentioned it in the commit comment and, I suppose, this part should be back-ported.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we emit WARN then in case of correlationDataGenerator resolves to null?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I decided to go with DEBUG (there is a corresponding DEBUG in handleConfirm).

Someone might want to use this as a feature ("I don't care about a confirm for this message").

}
else {
this.logger.debug("'confirmCorrelationExpression' resolved to 'null'; "
+ "no publisher confirm will be sent to the ack or nack channel");
}
}
return correlationData;
}
Expand Down Expand Up @@ -604,6 +611,23 @@ public Message<?> getMessage() {
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 change: 1 addition & 0 deletions .../java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"/>
Expand Down
13 changes: 12 additions & 1 deletion ...rc/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,16 @@

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;
Expand Down Expand Up @@ -143,11 +147,18 @@ public void adapterWithPublisherConfirms() throws Exception {
@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
Expand Down