Fix conversion of Message<?> payload for replies

If a custom MessageConverter is set, it is not used for replies defined
via the Message abstraction. This commit harmonizes the behaviour of the
`MessagingMessageConverter` so that the conversion of the payload can
be converted for both incoming and outgoing messages.

Issue: SPR-12912

spring-jms/src/main/java/org/springframework/jms/listener/adapter/AbstractAdaptableMessageListener.java

 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -396,6 +396,15 @@ public abstract class AbstractAdaptableMessageListener
 		protected Object extractPayload(Message message) throws JMSException {
 			return extractMessage(message);
 		}
+
+		@Override
+		protected Message createMessageForPayload(Object payload, Session session) throws JMSException {
+			MessageConverter converter = getMessageConverter();
+			if (converter != null) {
+				return converter.toMessage(payload, session);
+			}
+			throw new IllegalStateException("No message converter, cannot handle '" + payload + "'");
+		}
 	}
 
 

spring-jms/src/main/java/org/springframework/jms/support/converter/MessagingMessageConverter.java

 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -93,7 +93,7 @@ public class MessagingMessageConverter implements MessageConverter, Initializing
 					Message.class.getName() + "] is handled by this converter");
 		}
 		Message<?> input = (Message<?>) object;
-		javax.jms.Message reply = this.payloadConverter.toMessage(input.getPayload(), session);
+		javax.jms.Message reply = createMessageForPayload(input.getPayload(), session);
 		this.headerMapper.fromHeaders(input.getHeaders(), reply);
 		return reply;
 	}
@@ -119,4 +119,12 @@ public class MessagingMessageConverter implements MessageConverter, Initializing
 		return this.payloadConverter.fromMessage(message);
 	}
 
+	/**
+	 * Create a JMS message for the specified payload.
+	 * @see MessageConverter#toMessage(Object, Session)
+	 */
+	protected javax.jms.Message createMessageForPayload(Object payload, Session session) throws JMSException {
+		return this.payloadConverter.toMessage(payload, session);
+	}
+
 }

spring-jms/src/test/java/org/springframework/jms/listener/adapter/MessagingMessageListenerAdapterTests.java

 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -17,6 +17,8 @@
 package org.springframework.jms.listener.adapter;
 
 import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
 import javax.jms.Destination;
 import javax.jms.JMSException;
 import javax.jms.Session;
@@ -28,6 +30,7 @@ import org.junit.Test;
 import org.springframework.beans.factory.support.StaticListableBeanFactory;
 import org.springframework.jms.StubTextMessage;
 import org.springframework.jms.support.JmsHeaders;
+import org.springframework.jms.support.converter.MessageConverter;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConversionException;
 import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
@@ -111,6 +114,38 @@ public class MessagingMessageListenerAdapterTests {
 		}
 	}
 
+	@Test
+	public void incomingMessageUsesMessageConverter() throws JMSException {
+		javax.jms.Message jmsMessage = mock(javax.jms.Message.class);
+		Session session = mock(Session.class);
+		MessageConverter messageConverter = mock(MessageConverter.class);
+		given(messageConverter.fromMessage(jmsMessage)).willReturn("FooBar");
+		MessagingMessageListenerAdapter listener = getSimpleInstance("simple", Message.class);
+		listener.setMessageConverter(messageConverter);
+		listener.onMessage(jmsMessage, session);
+		verify(messageConverter, times(1)).fromMessage(jmsMessage);
+		assertEquals(1, sample.simples.size());
+		assertEquals("FooBar", sample.simples.get(0).getPayload());
+	}
+
+	@Test
+	public void replyUsesMessageConverterForPayload() throws JMSException {
+		Session session = mock(Session.class);
+		MessageConverter messageConverter = mock(MessageConverter.class);
+		given(messageConverter.toMessage("Response", session)).willReturn(new StubTextMessage("Response"));
+
+		Message<String> result = MessageBuilder.withPayload("Response")
+				.build();
+
+		MessagingMessageListenerAdapter listener = getSimpleInstance("echo", Message.class);
+		listener.setMessageConverter(messageConverter);
+		javax.jms.Message replyMessage = listener.buildMessage(session, result);
+
+		verify(messageConverter, times(1)).toMessage("Response", session);
+		assertNotNull("reply should never be null", replyMessage);
+		assertEquals("Response", ((TextMessage) replyMessage).getText());
+	}
+
 	protected MessagingMessageListenerAdapter getSimpleInstance(String methodName, Class... parameterTypes) {
 		Method m = ReflectionUtils.findMethod(SampleBean.class, methodName, parameterTypes);
 		return createInstance(m);
@@ -131,6 +166,12 @@ public class MessagingMessageListenerAdapterTests {
 	@SuppressWarnings("unused")
 	private static class SampleBean {
 
+		public final List<Message<String>> simples = new ArrayList<>();
+
+		public void simple(Message<String> input) {
+			simples.add(input);
+		}
+
 		public Message<String> echo(Message<String> input) {
 			return MessageBuilder.withPayload(input.getPayload())
 					.setHeader(JmsHeaders.TYPE, "reply")