AMQP-827: Fix @rl reply Message<?> conversion

JIRA: https://jira.spring.io/browse/AMQP-827

Use the correct `payloadConverter` in the `MessagingMessageConverter`
to support `@RabbitListener` `Message<?>` return types.

**cherry-pick to 2.0.x, 1.7.x**

# Conflicts:
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java

Author: garyrussell

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java

@@ -97,6 +97,12 @@ protected final MessagingMessageConverter getMessagingMessageConverter() {
 		return this.messagingMessageConverter;
 	}
 
+	@Override
+	public void setMessageConverter(MessageConverter messageConverter) {
+		super.setMessageConverter(messageConverter);
+		this.messagingMessageConverter.setPayloadConverter(messageConverter);
+	}
+
 	@Override
 	public void onMessage(org.springframework.amqp.core.Message amqpMessage, Channel channel) throws Exception {
 		Message<?> message = toMessagingMessage(amqpMessage);

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java

@@ -18,7 +18,10 @@
 
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.hamcrest.Matchers.startsWith;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -36,6 +39,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
@@ -65,6 +69,7 @@
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessagePostProcessor;
 import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.MessagePropertiesBuilder;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -111,6 +116,7 @@
 import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.test.annotation.DirtiesContext;
@@ -153,7 +159,8 @@ public class EnableRabbitIntegrationTests {
 			"test.converted", "test.converted.list", "test.converted.array", "test.converted.args1",
 			"test.converted.args2", "test.converted.message", "test.notconverted.message",
 			"test.notconverted.channel", "test.notconverted.messagechannel", "test.notconverted.messagingmessage",
-			"test.converted.foomessage", "test.notconverted.messagingmessagenotgeneric", "amqp656dlq");
+			"test.converted.foomessage", "test.notconverted.messagingmessagenotgeneric", "amqp656dlq",
+			"test.messaging.message", "test.amqp.message");
 
 	@Autowired
 	private RabbitTemplate rabbitTemplate;
@@ -577,6 +584,26 @@ public void testPrototypeCache() {
 		assertSame(value, typeCache.get(Foo1.class));
 	}
 
+	@Test
+	public void messagingMessageReturned() {
+		Message message = org.springframework.amqp.core.MessageBuilder.withBody("\"messaging\"".getBytes())
+			.andProperties(MessagePropertiesBuilder.newInstance().setContentType("application/json").build()).build();
+		message = this.rabbitTemplate.sendAndReceive("test.messaging.message", message);
+		assertThat(message, is(notNullValue()));
+		assertThat(new String(message.getBody()), equalTo("{\"field\":\"MESSAGING\"}"));
+		assertThat(message.getMessageProperties().getHeaders().get("foo"), equalTo("bar"));
+	}
+
+	@Test
+	public void amqpMessageReturned() {
+		Message message = org.springframework.amqp.core.MessageBuilder.withBody("amqp".getBytes())
+				.andProperties(MessagePropertiesBuilder.newInstance().setContentType("text/plain").build()).build();
+		message = this.rabbitTemplate.sendAndReceive("test.amqp.message", message);
+		assertThat(message, is(notNullValue()));
+		assertThat(new String(message.getBody()), equalTo("AMQP"));
+		assertThat(message.getMessageProperties().getHeaders().get("foo"), equalTo("bar"));
+	}
+
 	interface TxService {
 
 		@Transactional
@@ -811,6 +838,22 @@ public String handleWithInternalExchangeIgnore(String foo) {
 		public String handleWithDeadLetterDefaultExchange(String foo) {
 			throw new AmqpRejectAndDontRequeueException("dlq");
 		}
+		@RabbitListener(queues = "test.messaging.message", containerFactory = "simpleJsonListenerContainerFactory")
+		public org.springframework.messaging.Message<Bar> messagingMessage(String in) {
+			Bar bar = new Bar();
+			bar.field = in.toUpperCase();
+			return new GenericMessage<>(bar, Collections.singletonMap("foo", "bar"));
+		}
+
+		@RabbitListener(queues = "test.amqp.message")
+		public Message amqpMessage(String in) {
+			return org.springframework.amqp.core.MessageBuilder.withBody(in.toUpperCase().getBytes())
+					.andProperties(MessagePropertiesBuilder.newInstance().setContentType("text/plain")
+							.setHeader("foo", "bar")
+							.build())
+					.build();
+		}
+
 	}
 
 	public static class Foo1 {