From cb1b1c81f1b8bb1c09b39d19b14c2a463c8dbbdd Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 22 Feb 2019 10:20:51 -0500 Subject: [PATCH] GH-2759: Fix CorrelationData.future * 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` --- build.gradle | 2 +- .../AbstractAmqpOutboundEndpoint.java | 28 ++++++- .../AmqpOutboundEndpointTests-context.xml | 1 + .../outbound/AmqpOutboundEndpointTests.java | 77 ++++++++++--------- 4 files changed, 70 insertions(+), 38 deletions(-) diff --git a/build.gradle b/build.gradle index 76811213ba..10f39ea8a5 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java index 0cfa8b0a02..5ba1806b3c 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java @@ -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 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); + } } } diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml index 7fef86f1bc..6e2d6282d1 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests-context.xml @@ -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" /> diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java index 889c2cf2a1..66833b74e3 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpointTests.java @@ -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; }