Add support for MIME-based message conversion

The MessageConverter interface in spring-messaging is now explicitly
designed to support conversion of the payload of a Message<?> to and
from serialized form based on MIME type message header.
By default, the MessageHeaders.CONTENT_HEADER header is used but a
custom ContentTypeResolver can be configured to customize that.

Currently available are Jackson 2, String, and byte array converters.
A CompositeMessageConverter can be used to configure several
message converters in various places such as a messaging template.

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/core/AbstractDestinationResolvingMessagingTemplate.java

@@ -15,16 +15,22 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
 
 /**
+ * Base class for a messaging template that can resolve String-based destinations.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
-public abstract class AbstractDestinationResolvingMessagingTemplate<D> extends AbstractMessagingTemplate<D>
-		implements DestinationResolvingMessageSendingOperations<D>,
+public abstract class AbstractDestinationResolvingMessagingTemplate<D> extends
+		AbstractMessagingTemplate<D> implements
+		DestinationResolvingMessageSendingOperations<D>,
 		DestinationResolvingMessageReceivingOperations<D>,
 		DestinationResolvingMessageRequestReplyOperations<D> {
 
@@ -48,14 +54,29 @@ protected final D resolveDestination(String destinationName) {
 	}
 
 	@Override
-	public <T> void convertAndSend(String destinationName, T message) {
-		this.convertAndSend(destinationName, message, null);
+	public <T> void convertAndSend(String destinationName, T payload) {
+		Map<String, Object> headers = null;
+		this.convertAndSend(destinationName, payload, headers);
+	}
+
+	@Override
+	public <T> void convertAndSend(String destinationName, T payload, Map<String, Object> headers) {
+		MessagePostProcessor postProcessor = null;
+		this.convertAndSend(destinationName, payload, headers, postProcessor);
 	}
 
 	@Override
-	public <T> void convertAndSend(String destinationName, T message, MessagePostProcessor postProcessor) {
+	public <T> void convertAndSend(String destinationName, T payload, MessagePostProcessor postProcessor) {
+		Map<String, Object> headers = null;
+		this.convertAndSend(destinationName, payload, headers, postProcessor);
+	}
+
+	@Override
+	public <T> void convertAndSend(String destinationName, T payload, Map<String, Object> headers,
+			MessagePostProcessor postProcessor) {
+
 		D destination = resolveDestination(destinationName);
-		super.convertAndSend(destination, message, postProcessor);
+		super.convertAndSend(destination, payload, headers, postProcessor);
 	}
 
 	@Override
@@ -65,9 +86,9 @@ public <P> Message<P> receive(String destinationName) {
 	}
 
 	@Override
-	public Object receiveAndConvert(String destinationName) {
+	public <T> T receiveAndConvert(String destinationName, Class<T> targetClass) {
 		D destination = resolveDestination(destinationName);
-		return super.receiveAndConvert(destination);
+		return super.receiveAndConvert(destination, targetClass);
 	}
 
 	@Override
@@ -77,15 +98,33 @@ public Message<?> sendAndReceive(String destinationName, Message<?> requestMessa
 	}
 
 	@Override
-	public Object convertSendAndReceive(String destinationName, Object request) {
+	public <T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass) {
+		D destination = resolveDestination(destinationName);
+		return super.convertSendAndReceive(destination, request, targetClass);
+	}
+
+	@Override
+	public <T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
+			Class<T> targetClass) {
+
+		D destination = resolveDestination(destinationName);
+		return super.convertSendAndReceive(destination, request, headers, targetClass);
+	}
+
+	@Override
+	public <T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass,
+			MessagePostProcessor postProcessor) {
+
 		D destination = resolveDestination(destinationName);
-		return super.convertSendAndReceive(destination, request);
+		return super.convertSendAndReceive(destination, request, targetClass, postProcessor);
 	}
 
 	@Override
-	public Object convertSendAndReceive(String destinationName, Object request, MessagePostProcessor postProcessor) {
+	public <T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
+			Class<T> targetClass, MessagePostProcessor postProcessor) {
+
 		D destination = resolveDestination(destinationName);
-		return super.convertSendAndReceive(destination, request, postProcessor);
+		return super.convertSendAndReceive(destination, request, headers, targetClass, postProcessor);
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/core/AbstractMessageSendingTemplate.java

@@ -15,17 +15,27 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.support.converter.ByteArrayMessageConverter;
+import org.springframework.messaging.support.converter.CompositeMessageConverter;
 import org.springframework.messaging.support.converter.MessageConverter;
-import org.springframework.messaging.support.converter.SimplePayloadMessageConverter;
+import org.springframework.messaging.support.converter.StringMessageConverter;
 import org.springframework.util.Assert;
 
 
 /**
+ * Base class for templates that support sending messages.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public abstract class AbstractMessageSendingTemplate<D> implements MessageSendingOperations<D> {
@@ -34,9 +44,16 @@ public abstract class AbstractMessageSendingTemplate<D> implements MessageSendin
 
 	private volatile D defaultDestination;
 
-	private volatile MessageConverter converter = new SimplePayloadMessageConverter();
+	private volatile MessageConverter converter;
 
 
+	public AbstractMessageSendingTemplate() {
+		Collection<MessageConverter> converters = new ArrayList<MessageConverter>();
+		converters.add(new StringMessageConverter());
+		converters.add(new ByteArrayMessageConverter());
+		this.converter = new CompositeMessageConverter(converters);
+	}
+
 	public void setDefaultDestination(D defaultDestination) {
 		this.defaultDestination = defaultDestination;
 	}
@@ -58,19 +75,13 @@ public void setMessageConverter(MessageConverter messageConverter) {
 	/**
 	 * @return the configured {@link MessageConverter}
 	 */
-	public MessageConverter getConverter() {
+	public MessageConverter getMessageConverter() {
 		return this.converter;
 	}
 
-	/**
-	 * @param converter the converter to set
-	 */
-	public void setConverter(MessageConverter converter) {
-		this.converter = converter;
-	}
 
 	@Override
-	public <P> void send(Message<P> message) {
+	public void send(Message<?> message) {
 		this.send(getRequiredDefaultDestination(), message);
 	}
 
@@ -82,34 +93,48 @@ protected final D getRequiredDefaultDestination() {
 	}
 
 	@Override
-	public <P> void send(D destination, Message<P> message) {
+	public void send(D destination, Message<?> message) {
 		this.doSend(destination, message);
 	}
 
 	protected abstract void doSend(D destination, Message<?> message);
 
 
 	@Override
-	public <T> void convertAndSend(T message) {
+	public void convertAndSend(Object message) throws MessagingException {
 		this.convertAndSend(getRequiredDefaultDestination(), message);
 	}
 
 	@Override
-	public <T> void convertAndSend(D destination, T object) {
-		this.convertAndSend(destination, object, null);
+	public void convertAndSend(D destination, Object payload) throws MessagingException {
+		this.convertAndSend(destination, payload, (Map<String, Object>) null);
 	}
 
 	@Override
-	public <T> void convertAndSend(T object, MessagePostProcessor postProcessor) {
-		this.convertAndSend(getRequiredDefaultDestination(), object, postProcessor);
+	public void convertAndSend(D destination, Object payload, Map<String, Object> headers) throws MessagingException {
+		MessagePostProcessor postProcessor = null;
+		this.convertAndSend(destination, payload, headers, postProcessor);
 	}
 
 	@Override
-	public <T> void convertAndSend(D destination, T object, MessagePostProcessor postProcessor)
+	public void convertAndSend(Object payload, MessagePostProcessor postProcessor) throws MessagingException {
+		this.convertAndSend(getRequiredDefaultDestination(), payload, postProcessor);
+	}
+
+	@Override
+	public void convertAndSend(D destination, Object payload, MessagePostProcessor postProcessor)
 			throws MessagingException {
 
-		@SuppressWarnings("unchecked")
-		Message<?> message = this.converter.toMessage(object);
+		Map<String, Object> headers = null;
+		this.convertAndSend(destination, payload, headers, postProcessor);
+	}
+
+	@Override
+	public void convertAndSend(D destination, Object payload, Map<String, Object> headers,
+			MessagePostProcessor postProcessor) throws MessagingException {
+
+		MessageHeaders messageHeaders = (headers != null) ? new MessageHeaders(headers) : null;
+		Message<?> message = this.converter.toMessage(payload, messageHeaders);
 		if (postProcessor != null) {
 			message = postProcessor.postProcessMessage(message);
 		}

spring-messaging/src/main/java/org/springframework/messaging/core/AbstractMessagingTemplate.java

@@ -15,11 +15,17 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 
 
 /**
+ * Base class for a messaging template that send and receive messages.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public abstract class AbstractMessagingTemplate<D> extends AbstractMessageSendingTemplate<D>
@@ -40,18 +46,22 @@ public <P> Message<P> receive(D destination) {
 
 
 	@Override
-	public Object receiveAndConvert() {
-		return this.receiveAndConvert(getRequiredDefaultDestination());
+	public <T> T receiveAndConvert(Class<T> targetClass) {
+		return this.receiveAndConvert(getRequiredDefaultDestination(), targetClass);
 	}
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public Object receiveAndConvert(D destination) {
+	public <T> T receiveAndConvert(D destination, Class<T> targetClass) {
 		Message<?> message = this.doReceive(destination);
-		return (message != null) ? getConverter().fromMessage(message, null) : null;
+		if (message != null) {
+			return (T) getMessageConverter().fromMessage(message, targetClass);
+		}
+		else {
+			return null;
+		}
 	}
 
-
 	@Override
 	public Message<?> sendAndReceive(Message<?> requestMessage) {
 		return this.sendAndReceive(getRequiredDefaultDestination(), requestMessage);
@@ -66,29 +76,48 @@ public Message<?> sendAndReceive(D destination, Message<?> requestMessage) {
 
 
 	@Override
-	public Object convertSendAndReceive(Object request) {
-		return this.convertSendAndReceive(getRequiredDefaultDestination(), request);
+	public <T> T convertSendAndReceive(Object request, Class<T> targetClass) {
+		return this.convertSendAndReceive(getRequiredDefaultDestination(), request, targetClass);
+	}
+
+	@Override
+	public <T> T convertSendAndReceive(D destination, Object request, Class<T> targetClass) {
+		Map<String, Object> headers = null;
+		return this.convertSendAndReceive(destination, request, headers, targetClass);
 	}
 
 	@Override
-	public Object convertSendAndReceive(D destination, Object request) {
+	public <T> T convertSendAndReceive(D destination, Object request, Map<String, Object> headers,
+			Class<T> targetClass) {
+
 		return this.convertSendAndReceive(destination, request, null);
 	}
 
 	@Override
-	public Object convertSendAndReceive(Object request, MessagePostProcessor postProcessor) {
-		return this.convertSendAndReceive(getRequiredDefaultDestination(), request, postProcessor);
+	public <T> T convertSendAndReceive(Object request, Class<T> targetClass, MessagePostProcessor postProcessor) {
+		return this.convertSendAndReceive(getRequiredDefaultDestination(), request, targetClass, postProcessor);
+	}
+
+	@Override
+	public <T> T convertSendAndReceive(D destination, Object request, Class<T> targetClass,
+			MessagePostProcessor postProcessor) {
+
+		Map<String, Object> headers = null;
+		return this.convertSendAndReceive(destination, request, headers, targetClass, postProcessor);
 	}
 
 	@SuppressWarnings("unchecked")
 	@Override
-	public Object convertSendAndReceive(D destination, Object request, MessagePostProcessor postProcessor) {
-		Message<?> requestMessage = getConverter().toMessage(request);
+	public <T> T convertSendAndReceive(D destination, Object request, Map<String, Object> headers,
+			Class<T> targetClass, MessagePostProcessor postProcessor) {
+
+		MessageHeaders messageHeaders = (headers != null) ? new MessageHeaders(headers) : null;
+		Message<?> requestMessage = getMessageConverter().toMessage(request, messageHeaders);
 		if (postProcessor != null) {
 			requestMessage = postProcessor.postProcessMessage(requestMessage);
 		}
 		Message<?> replyMessage = this.sendAndReceive(destination, requestMessage);
-		return getConverter().fromMessage(replyMessage, null);
+		return (T) getMessageConverter().fromMessage(replyMessage, targetClass);
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/core/DestinationResolvingMessageReceivingOperations.java

@@ -20,13 +20,16 @@
 
 
 /**
+ * A {@link MessageReceivingOperations} that can resolve a String-based destinations.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public interface DestinationResolvingMessageReceivingOperations<D> extends MessageReceivingOperations<D> {
 
 	<P> Message<P> receive(String destinationName) throws MessagingException;
 
-	Object receiveAndConvert(String destinationName) throws MessagingException;
+	<T> T receiveAndConvert(String destinationName, Class<T> targetClass) throws MessagingException;
 
 }

spring-messaging/src/main/java/org/springframework/messaging/core/DestinationResolvingMessageRequestReplyOperations.java

@@ -15,19 +15,33 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
 
 
 /**
+ * A {@link MessageRequestReplyOperations} that can resolve a String-based destinations.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public interface DestinationResolvingMessageRequestReplyOperations<D> extends MessageRequestReplyOperations<D> {
 
-	Message<?> sendAndReceive(String destinationName, Message<?> requestMessage);
+	Message<?> sendAndReceive(String destinationName, Message<?> requestMessage) throws MessagingException;
+
+	<T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass)
+			throws MessagingException;
+
+	<T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
+			Class<T> targetClass) throws MessagingException;
 
-	Object convertSendAndReceive(String destinationName, Object request);
+	<T> T convertSendAndReceive(String destinationName, Object request,
+			Class<T> targetClass, MessagePostProcessor requestPostProcessor) throws MessagingException;
 
-	Object convertSendAndReceive(String destinationName, Object request, MessagePostProcessor requestPostProcessor);
+	<T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
+			Class<T> targetClass, MessagePostProcessor requestPostProcessor) throws MessagingException;
 
 }