INT-4371: More polishing - consumer queue header

JIRA: https://jira.spring.io/browse/INT-4371

Author: garyrussell

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/inbound/AmqpMessageSource.java

@@ -169,6 +169,7 @@ protected AbstractIntegrationMessageBuilder<Object> doReceive() {
 					.createCallback(new AmqpAckInfo(connection, channel, this.transacted, resp));
 			MessageProperties messageProperties = this.propertiesConverter.toMessageProperties(resp.getProps(),
 					resp.getEnvelope(), StandardCharsets.UTF_8.name());
+			messageProperties.setConsumerQueue(this.queue);
 			Map<String, Object> headers = this.headerMapper.toHeadersFromRequest(messageProperties);
 			org.springframework.amqp.core.Message amqpMessage = new org.springframework.amqp.core.Message(resp.getBody(), messageProperties);
 			Object payload = this.messageConverter.fromMessage(amqpMessage);

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/AmqpMessageSourceTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.amqp.inbound;
 
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertThat;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -32,6 +33,7 @@
 import org.junit.Test;
 
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.integration.amqp.support.AmqpMessageHeaderErrorMessageStrategy;
 import org.springframework.integration.support.AcknowledgmentCallback.Status;
 import org.springframework.integration.support.StaticMessageHeaderAccessor;
@@ -72,6 +74,7 @@ public void testAck() throws Exception {
 		Message<?> received = source.receive();
 		assertThat(received.getHeaders().get(AmqpMessageHeaderErrorMessageStrategy.AMQP_RAW_MESSAGE),
 				instanceOf(org.springframework.amqp.core.Message.class));
+		assertThat(received.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE), equalTo("foo"));
 		// make sure channel is not cached
 		org.springframework.amqp.rabbit.connection.Connection conn = ccf.createConnection();
 		Channel notCached = conn.createChannel(false); // should not have been "closed"