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.

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 @@ public abstract class AbstractDestinationResolvingMessagingTemplate<D> extends A
 	}
 
 	@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 message, MessagePostProcessor postProcessor) {
+	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 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 abstract class AbstractDestinationResolvingMessagingTemplate<D> extends A
 	}
 
 	@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 abstract class AbstractDestinationResolvingMessagingTemplate<D> extends A
 	}
 
 	@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);
+		return super.convertSendAndReceive(destination, request, targetClass);
 	}
 
 	@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) {
+
 		D destination = resolveDestination(destinationName);
-		return super.convertSendAndReceive(destination, request, postProcessor);
+		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, targetClass, postProcessor);
+	}
+
+	@Override
+	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, 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 abstract class AbstractMessageSendingTemplate<D> implements MessageSendin
 	/**
 	 * @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,7 +93,7 @@ public abstract class AbstractMessageSendingTemplate<D> implements MessageSendin
 	}
 
 	@Override
-	public <P> void send(D destination, Message<P> message) {
+	public void send(D destination, Message<?> message) {
 		this.doSend(destination, message);
 	}
 
@@ -90,26 +101,40 @@ public abstract class AbstractMessageSendingTemplate<D> implements MessageSendin
 
 
 	@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 abstract class AbstractMessagingTemplate<D> extends AbstractMessageSendin
 
 
 	@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 abstract class AbstractMessagingTemplate<D> extends AbstractMessageSendin
 
 
 	@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 Object convertSendAndReceive(D destination, Object request) {
+	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 <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 @@ import org.springframework.messaging.MessagingException;
 
 
 /**
+ * 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;
 
-	Object convertSendAndReceive(String destinationName, Object request);
+	<T> T convertSendAndReceive(String destinationName, Object request, Class<T> targetClass)
+			throws MessagingException;
 
-	Object convertSendAndReceive(String destinationName, Object request, MessagePostProcessor requestPostProcessor);
+	<T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
+			Class<T> targetClass) throws MessagingException;
+
+	<T> T convertSendAndReceive(String destinationName, Object request,
+			Class<T> targetClass, MessagePostProcessor requestPostProcessor) throws MessagingException;
+
+	<T> T convertSendAndReceive(String destinationName, Object request, Map<String, Object> headers,
+			Class<T> targetClass, MessagePostProcessor requestPostProcessor) throws MessagingException;
 
 }

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

@@ -15,12 +15,17 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 
 
 /**
+ * A {@link MessageSendingOperations} that can resolve a String-based destinations.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public interface DestinationResolvingMessageSendingOperations<D> extends MessageSendingOperations<D> {
@@ -29,7 +34,12 @@ public interface DestinationResolvingMessageSendingOperations<D> extends Message
 
 	<T> void convertAndSend(String destinationName, T payload) throws MessagingException;
 
-	<T> void convertAndSend(String destinationName, T payload, MessagePostProcessor postProcessor)
-			throws MessagingException;
+	<T> void convertAndSend(String destinationName, T payload, Map<String, Object> headers) throws MessagingException;
+
+	<T> void convertAndSend(String destinationName, T payload,
+			MessagePostProcessor postProcessor) throws MessagingException;
+
+	<T> void convertAndSend(String destinationName, T payload, Map<String, Object> headers,
+			MessagePostProcessor postProcessor) throws MessagingException;
 
 }

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

@@ -33,6 +33,9 @@ import org.springframework.util.Assert;
 
 
 /**
+ * A messaging template for sending to and/or receiving messages from a
+ * {@link MessageChannel}.
+ *
  * @author Mark Fisher
  * @since 4.0
  */

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

@@ -20,7 +20,12 @@ import org.springframework.messaging.MessagingException;
 
 
 /**
+ * A set of operations receiving messages from a destination.
+ *
+ * @param <D> the type of destination from which messages can be received
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public interface MessageReceivingOperations<D> {
@@ -29,8 +34,10 @@ public interface MessageReceivingOperations<D> {
 
 	<P> Message<P> receive(D destination) throws MessagingException;
 
-	Object receiveAndConvert() throws MessagingException;
+	<T> T receiveAndConvert(Class<T> targetClass) throws MessagingException;
 
-	Object receiveAndConvert(D destination) throws MessagingException;
+	<T> T receiveAndConvert(D destination, Class<T> targetClass) throws MessagingException;
 
 }
+
+

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

@@ -15,25 +15,41 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
 
 
 /**
+ * A set of operations for exchanging messages to and from a destination.
+ *
+ * @param <D> the type of destination to send and receive messages from
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public interface MessageRequestReplyOperations<D> {
 
-	Message<?> sendAndReceive(Message<?> requestMessage);
+	Message<?> sendAndReceive(Message<?> requestMessage) throws MessagingException;
 
-	Message<?> sendAndReceive(D destination, Message<?> requestMessage);
+	Message<?> sendAndReceive(D destination, Message<?> requestMessage) throws MessagingException;
 
-	Object convertSendAndReceive(Object request);
+	<T> T convertSendAndReceive(Object request, Class<T> targetClass) throws MessagingException;
 
-	Object convertSendAndReceive(D destination, Object request);
+	<T> T convertSendAndReceive(D destination, Object request, Class<T> targetClass) throws MessagingException;
 
-	Object convertSendAndReceive(Object request, MessagePostProcessor requestPostProcessor);
+	<T> T convertSendAndReceive(D destination, Object request, Map<String, Object> headers, Class<T> targetClass)
+			throws MessagingException;
 
-	Object convertSendAndReceive(D destination, Object request, MessagePostProcessor requestPostProcessor);
+	<T> T convertSendAndReceive(Object request, Class<T> targetClass, MessagePostProcessor requestPostProcessor)
+			throws MessagingException;
+
+	<T> T convertSendAndReceive(D destination, Object request, Class<T> targetClass,
+			MessagePostProcessor requestPostProcessor) throws MessagingException;
+
+	<T> T convertSendAndReceive(D destination, Object request, Map<String, Object> headers,
+			Class<T> targetClass, MessagePostProcessor requestPostProcessor) throws MessagingException;
 
 }

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

@@ -15,26 +15,39 @@
  */
 package org.springframework.messaging.core;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 
 
 /**
+ * A set of operations sending messages to a destination.
+ *
+ * @param <D> the type of destination to which messages can be sent
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
 public interface MessageSendingOperations<D> {
 
-	<P> void send(Message<P> message) throws MessagingException;
+	void send(Message<?> message) throws MessagingException;
 
-	<P> void send(D destination, Message<P> message) throws MessagingException;
+	void send(D destination, Message<?> message) throws MessagingException;
 
-	<T> void convertAndSend(T payload) throws MessagingException;
+	void convertAndSend(Object payload) throws MessagingException;
 
-	<T> void convertAndSend(D destination, T payload) throws MessagingException;
+	void convertAndSend(D destination, Object payload) throws MessagingException;
 
-	<T> void convertAndSend(T payload, MessagePostProcessor postProcessor) throws MessagingException;
+	void convertAndSend(D destination, Object payload, Map<String, Object> headers) throws MessagingException;
 
-	<T> void convertAndSend(D destination, T payload, MessagePostProcessor postProcessor) throws MessagingException;
+	void convertAndSend(Object payload, MessagePostProcessor postProcessor) throws MessagingException;
+
+	void convertAndSend(D destination, Object payload,
+			MessagePostProcessor postProcessor) throws MessagingException;
+
+	void convertAndSend(D destination, Object payload, Map<String, Object> headers,
+			MessagePostProcessor postProcessor) throws MessagingException;
 
 }

spring-messaging/src/main/java/org/springframework/messaging/handler/annotation/support/MessageBodyMethodArgumentResolver.java

@@ -35,10 +35,10 @@ import org.springframework.util.Assert;
  */
 public class MessageBodyMethodArgumentResolver implements HandlerMethodArgumentResolver {
 
-	private final MessageConverter<?> converter;
+	private final MessageConverter converter;
 
 
-	public MessageBodyMethodArgumentResolver(MessageConverter<?> converter) {
+	public MessageBodyMethodArgumentResolver(MessageConverter converter) {
 		Assert.notNull(converter, "converter is required");
 		this.converter = converter;
 	}

spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageHeaderAccessor.java

@@ -20,9 +20,7 @@ import java.security.Principal;
 import java.util.List;
 import java.util.Map;
 
-import org.springframework.http.MediaType;
 import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.NativeMessageHeaderAccessor;
 import org.springframework.util.Assert;
 
@@ -114,14 +112,6 @@ public class SimpMessageHeaderAccessor extends NativeMessageHeaderAccessor {
 		return (String) getHeader(DESTINATION_HEADER);
 	}
 
-	public MediaType getContentType() {
-		return (MediaType) getHeader(MessageHeaders.CONTENT_TYPE);
-	}
-
-	public void setContentType(MediaType contentType) {
-		setHeader(MessageHeaders.CONTENT_TYPE, contentType);
-	}
-
 	public String getSubscriptionId() {
 		return (String) getHeader(SUBSCRIPTION_ID_HEADER);
 	}

spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessageSendingOperations.java

@@ -16,6 +16,8 @@
 
 package org.springframework.messaging.simp;
 
+import java.util.Map;
+
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.core.MessagePostProcessor;
 import org.springframework.messaging.core.MessageSendingOperations;
@@ -35,19 +37,25 @@ public interface SimpMessageSendingOperations extends MessageSendingOperations<S
 	 *
 	 * @param user the user that should receive the message.
 	 * @param destination the destination to send the message to.
-	 * @param message the message to send
+	 * @param payload the payload to send
 	 */
-	<T> void convertAndSendToUser(String user, String destination, T message) throws MessagingException;
+	void convertAndSendToUser(String user, String destination, Object payload) throws MessagingException;
+
+	void convertAndSendToUser(String user, String destination, Object payload, Map<String, Object> headers)
+			throws MessagingException;
 
 	/**
 	 * Send a message to a specific user.
 	 *
 	 * @param user the user that should receive the message.
 	 * @param destination the destination to send the message to.
-	 * @param message the message to send
+	 * @param payload the payload to send
 	 * @param postProcessor a postProcessor to post-process or modify the created message
 	 */
-	<T> void convertAndSendToUser(String user, String destination, T message, MessagePostProcessor postProcessor)
-			throws MessagingException;
+	void convertAndSendToUser(String user, String destination, Object payload,
+			MessagePostProcessor postProcessor) throws MessagingException;
+
+	void convertAndSendToUser(String user, String destination, Object payload, Map<String, Object> headers,
+			MessagePostProcessor postProcessor) throws MessagingException;
 
 }

spring-messaging/src/main/java/org/springframework/messaging/simp/SimpMessagingTemplate.java

@@ -15,6 +15,8 @@
  */
 package org.springframework.messaging.simp;
 
+import java.util.Map;
+
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
@@ -90,7 +92,7 @@ public class SimpMessagingTemplate extends AbstractMessageSendingTemplate<String
 
 
 	@Override
-	public <P> void send(Message<P> message) {
+	public void send(Message<?> message) {
 		SimpMessageHeaderAccessor headers = SimpMessageHeaderAccessor.wrap(message);
 		String destination = headers.getDestination();
 		destination = (destination != null) ? destination : getRequiredDefaultDestination();
@@ -120,16 +122,33 @@ public class SimpMessagingTemplate extends AbstractMessageSendingTemplate<String
 
 
 	@Override
-	public <T> void convertAndSendToUser(String user, String destination, T message) throws MessagingException {
-		convertAndSendToUser(user, destination, message, null);
+	public void convertAndSendToUser(String user, String destination, Object payload) throws MessagingException {
+		MessagePostProcessor postProcessor = null;
+		this.convertAndSendToUser(user, destination, payload, postProcessor);
 	}
 
 	@Override
-	public <T> void convertAndSendToUser(String user, String destination, T message,
+	public void convertAndSendToUser(String user, String destination, Object payload,
+			Map<String, Object> headers) throws MessagingException {
+
+		MessagePostProcessor postProcessor = null;
+		this.convertAndSendToUser(user, destination, payload, headers, postProcessor);
+	}
+
+	@Override
+	public void convertAndSendToUser(String user, String destination, Object payload,
+			MessagePostProcessor postProcessor) throws MessagingException {
+
+		Map<String, Object> headers = null;
+		this.convertAndSendToUser(user, destination, payload, headers, postProcessor);
+	}
+
+	@Override
+	public void convertAndSendToUser(String user, String destination, Object payload, Map<String, Object> headers,
 			MessagePostProcessor postProcessor) throws MessagingException {
 
 		Assert.notNull(user, "user is required");
-		convertAndSend(this.userDestinationPrefix + user + destination, message, postProcessor);
+		super.convertAndSend(this.userDestinationPrefix + user + destination, payload, headers, postProcessor);
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/simp/config/WebSocketMessageBrokerConfigurationSupport.java

@@ -16,6 +16,9 @@
 
 package org.springframework.messaging.simp.config;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.springframework.context.annotation.Bean;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.SubscribableChannel;
@@ -28,11 +31,18 @@ import org.springframework.messaging.simp.handler.MutableUserQueueSuffixResolver
 import org.springframework.messaging.simp.handler.SimpleUserQueueSuffixResolver;
 import org.springframework.messaging.simp.handler.UserDestinationMessageHandler;
 import org.springframework.messaging.support.channel.ExecutorSubscribableChannel;
+import org.springframework.messaging.support.converter.ByteArrayMessageConverter;
+import org.springframework.messaging.support.converter.CompositeMessageConverter;
+import org.springframework.messaging.support.converter.DefaultContentTypeResolver;
 import org.springframework.messaging.support.converter.MappingJackson2MessageConverter;
 import org.springframework.messaging.support.converter.MessageConverter;
+import org.springframework.messaging.support.converter.StringMessageConverter;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeTypeUtils;
 import org.springframework.web.servlet.HandlerMapping;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurationSupport;
 import org.springframework.web.servlet.handler.AbstractHandlerMapping;
 import org.springframework.web.socket.WebSocketHandler;
 import org.springframework.web.socket.server.config.SockJsServiceRegistration;
@@ -50,6 +60,10 @@ import org.springframework.web.socket.server.config.SockJsServiceRegistration;
  */
 public abstract class WebSocketMessageBrokerConfigurationSupport {
 
+	private static final boolean jackson2Present =
+			ClassUtils.isPresent("com.fasterxml.jackson.databind.ObjectMapper", WebMvcConfigurationSupport.class.getClassLoader()) &&
+					ClassUtils.isPresent("com.fasterxml.jackson.core.JsonGenerator", WebMvcConfigurationSupport.class.getClassLoader());
+
 	private MessageBrokerConfigurer messageBrokerConfigurer;
 
 
@@ -129,7 +143,7 @@ public abstract class WebSocketMessageBrokerConfigurationSupport {
 		AnnotationMethodMessageHandler handler =
 				new AnnotationMethodMessageHandler(brokerMessagingTemplate(), webSocketResponseChannel());
 		handler.setDestinationPrefixes(getMessageBrokerConfigurer().getAnnotationMethodDestinationPrefixes());
-		handler.setMessageConverter(brokerMessageConverter());
+		handler.setMessageConverter(simpMessageConverter());
 		webSocketRequestChannel().subscribe(handler);
 		return handler;
 	}
@@ -184,7 +198,7 @@ public abstract class WebSocketMessageBrokerConfigurationSupport {
 	@Bean
 	public SimpMessageSendingOperations brokerMessagingTemplate() {
 		SimpMessagingTemplate template = new SimpMessagingTemplate(brokerChannel());
-		template.setMessageConverter(brokerMessageConverter());
+		template.setMessageConverter(simpMessageConverter());
 		return template;
 	}
 
@@ -194,8 +208,16 @@ public abstract class WebSocketMessageBrokerConfigurationSupport {
 	}
 
 	@Bean
-	public MessageConverter<?> brokerMessageConverter() {
-		return new MappingJackson2MessageConverter();
+	public CompositeMessageConverter simpMessageConverter() {
+		DefaultContentTypeResolver contentTypeResolver = new DefaultContentTypeResolver();
+		List<MessageConverter> converters = new ArrayList<MessageConverter>();
+		converters.add(new StringMessageConverter());
+		converters.add(new ByteArrayMessageConverter());
+		if (jackson2Present) {
+			converters.add(new MappingJackson2MessageConverter());
+			contentTypeResolver.setDefaultMimeType(MimeTypeUtils.APPLICATION_JSON);
+		}
+		return new CompositeMessageConverter(converters, contentTypeResolver);
 	}
 
 

spring-messaging/src/main/java/org/springframework/messaging/simp/handler/AnnotationMethodMessageHandler.java

@@ -83,7 +83,7 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
 
 	private Collection<String> destinationPrefixes = new ArrayList<String>();
 
-	private MessageConverter<?> messageConverter;
+	private MessageConverter messageConverter;
 
 	private ApplicationContext applicationContext;
 
@@ -147,7 +147,7 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
 		return this.destinationPrefixes;
 	}
 
-	public void setMessageConverter(MessageConverter<?> converter) {
+	public void setMessageConverter(MessageConverter converter) {
 		this.messageConverter = converter;
 		if (converter != null) {
 			((AbstractMessageSendingTemplate<?>) this.webSocketResponseTemplate).setMessageConverter(converter);
@@ -176,7 +176,7 @@ public class AnnotationMethodMessageHandler implements MessageHandler, Applicati
 		this.customReturnValueHandlers = customReturnValueHandlers;
 	}
 
-	public MessageConverter<?> getMessageConverter() {
+	public MessageConverter getMessageConverter() {
 		return this.messageConverter;
 	}
 

spring-messaging/src/main/java/org/springframework/messaging/simp/stomp/StompHeaderAccessor.java

@@ -23,12 +23,13 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 
-import org.springframework.http.MediaType;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.messaging.simp.SimpMessageType;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
 import org.springframework.util.StringUtils;
 
 
@@ -117,7 +118,7 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
 
 		values = extHeaders.get(StompHeaderAccessor.STOMP_CONTENT_TYPE_HEADER);
 		if (!CollectionUtils.isEmpty(values)) {
-			super.setContentType(MediaType.parseMediaType(values.get(0)));
+			super.setContentType(MimeTypeUtils.parseMimeType(values.get(0)));
 		}
 
 		if (StompCommand.SUBSCRIBE.equals(command) || StompCommand.UNSUBSCRIBE.equals(command)) {
@@ -183,7 +184,7 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
 			result.put(STOMP_DESTINATION_HEADER, Arrays.asList(destination));
 		}
 
-		MediaType contentType = getContentType();
+		MimeType contentType = super.getContentType();
 		if (contentType != null) {
 			result.put(STOMP_CONTENT_TYPE_HEADER, Arrays.asList(contentType.toString()));
 		}
@@ -281,16 +282,9 @@ public class StompHeaderAccessor extends SimpMessageHeaderAccessor {
 		return new long[] { Long.valueOf(rawValues[0]), Long.valueOf(rawValues[1])};
 	}
 
-	public void setContentType(MediaType mediaType) {
-		if (mediaType != null) {
-			super.setContentType(mediaType);
-			setNativeHeader(STOMP_CONTENT_TYPE_HEADER, mediaType.toString());
-		}
-	}
-
-	public MediaType getContentType() {
-		String value = getFirstNativeHeader(STOMP_CONTENT_TYPE_HEADER);
-		return (value != null) ? MediaType.parseMediaType(value) : null;
+	public void setContentType(MimeType contentType) {
+		super.setContentType(contentType);
+		setNativeHeader(STOMP_CONTENT_TYPE_HEADER, contentType.toString());
 	}
 
 	public Integer getContentLength() {

spring-messaging/src/main/java/org/springframework/messaging/support/MessageHeaderAccessor.java

@@ -29,6 +29,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.PatternMatchUtils;
 import org.springframework.util.StringUtils;
@@ -235,6 +236,15 @@ public class MessageHeaderAccessor {
 		setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName);
 	}
 
+    public MimeType getContentType() {
+        return (MimeType) getHeader(MessageHeaders.CONTENT_TYPE);
+    }
+
+	public void setContentType(MimeType contentType) {
+		setHeader(MessageHeaders.CONTENT_TYPE, contentType);
+	}
+
+
 	@Override
 	public String toString() {
 		return getClass().getSimpleName() + " [originalHeaders=" + this.originalHeaders

spring-messaging/src/main/java/org/springframework/messaging/support/converter/AbstractMessageConverter.java

@@ -0,0 +1,206 @@
+/*
+ * Copyright 2002-2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.support.converter;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+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.support.MessageBuilder;
+import org.springframework.util.Assert;
+import org.springframework.util.MimeType;
+
+
+/**
+ * Abstract base class for {@link MessageConverter} implementations including support for
+ * common properties and a partial implementation of the conversion methods mainly to
+ * check if the converter supports the conversion based on the payload class and MIME
+ * type.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public abstract class AbstractMessageConverter implements MessageConverter {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	private final List<MimeType> supportedMimeTypes;
+
+	private Class<?> serializedPayloadClass = byte[].class;
+
+	private ContentTypeResolver contentTypeResolver;
+
+
+	/**
+	 * Construct an {@code AbstractMessageConverter} with one supported MIME type.
+	 * @param supportedMimeType the supported MIME type
+	 */
+	protected AbstractMessageConverter(MimeType supportedMimeType) {
+		this.supportedMimeTypes = Collections.<MimeType>singletonList(supportedMimeType);
+	}
+
+	/**
+	 * Construct an {@code AbstractMessageConverter} with multiple supported MIME type.
+	 * @param supportedMimeTypes the supported MIME types
+	 */
+	protected AbstractMessageConverter(Collection<MimeType> supportedMimeTypes) {
+		Assert.notNull(supportedMimeTypes, "'supportedMimeTypes' is required");
+		this.supportedMimeTypes = new ArrayList<MimeType>(supportedMimeTypes);
+	}
+
+
+	/**
+	 * Return the configured supported MIME types.
+	 */
+	public List<MimeType> getSupportedMimeTypes() {
+		return Collections.unmodifiableList(this.supportedMimeTypes);
+	}
+
+	/**
+	 * Configure the {@link ContentTypeResolver} to use.
+	 * <p>
+	 * The default value is {@code null}. However when {@link CompositeMessageConverter}
+	 * is used it configures all of its delegates with a default resolver.
+	 */
+	public void setContentTypeResolver(ContentTypeResolver resolver) {
+		this.contentTypeResolver = resolver;
+	}
+
+	/**
+	 * Return the default {@link ContentTypeResolver}.
+	 */
+	public ContentTypeResolver getContentTypeResolver() {
+		return this.contentTypeResolver;
+	}
+
+	/**
+	 * Configure the preferred serialization class to use (byte[] or String) when
+	 * converting an Object payload to a {@link Message}.
+	 * <p>
+	 * The default value is byte[].
+	 *
+	 * @param clazz either byte[] or String
+	 */
+	public void setSerializedPayloadClass(Class<?> clazz) {
+		Assert.isTrue(byte[].class.equals(clazz) || String.class.equals(clazz),
+				"Payload class must be byte[] or String: " + clazz);
+		this.serializedPayloadClass = clazz;
+	}
+
+	/**
+	 * Return the configured preferred serialization payload class.
+	 */
+	public Class<?> getSerializedPayloadClass() {
+		return this.serializedPayloadClass;
+	}
+
+	/**
+	 * Returns the default content type for the payload. Called when
+	 * {@link #toMessage(Object, MessageHeaders)} is invoked without message headers or
+	 * without a content type header.
+	 * <p>
+	 * By default, this returns the first element of the {@link #getSupportedMimeTypes()
+	 * supportedMimeTypes}, if any. Can be overridden in sub-classes.
+	 *
+	 * @param payload the payload being converted to message
+	 * @return the content type, or {@code null} if not known
+	 */
+	protected MimeType getDefaultContentType(Object payload) {
+		List<MimeType> mimeTypes = getSupportedMimeTypes();
+		return (!mimeTypes.isEmpty() ? mimeTypes.get(0) : null);
+	}
+
+	/**
+	 * Whether the given class is supported by this converter.
+	 *
+	 * @param clazz the class to test for support
+	 * @return {@code true} if supported; {@code false} otherwise
+	 */
+	protected abstract boolean supports(Class<?> clazz);
+
+
+	@Override
+	public final Object fromMessage(Message<?> message, Class<?> targetClass) {
+		if (!canConvertFrom(message, targetClass)) {
+			return null;
+		}
+		return convertFromInternal(message, targetClass);
+	}
+
+	protected boolean canConvertFrom(Message<?> message, Class<?> targetClass) {
+		return (supports(targetClass) && supportsMimeType(message.getHeaders()));
+	}
+
+	/**
+	 * Convert the message payload from serialized form to an Object.
+	 */
+	public abstract Object convertFromInternal(Message<?> message, Class<?> targetClass);
+
+	@Override
+	public final Message<?> toMessage(Object payload, MessageHeaders headers) {
+		if (!canConvertTo(payload, headers)) {
+			return null;
+		}
+		payload = convertToInternal(payload, headers);
+		MessageBuilder<?> builder = MessageBuilder.withPayload(payload);
+		if (headers != null) {
+			builder.copyHeaders(headers);
+		}
+		MimeType mimeType = getDefaultContentType(payload);
+		if (mimeType != null) {
+			builder.setHeaderIfAbsent(MessageHeaders.CONTENT_TYPE, mimeType);
+		}
+		return builder.build();
+	}
+
+	protected boolean canConvertTo(Object payload, MessageHeaders headers) {
+		Class<?> clazz = (payload != null) ? payload.getClass() : null;
+		return (supports(clazz) && supportsMimeType(headers));
+	}
+
+	/**
+	 * Convert the payload object to serialized form.
+	 */
+	public abstract Object convertToInternal(Object payload, MessageHeaders headers);
+
+	protected boolean supportsMimeType(MessageHeaders headers) {
+		MimeType mimeType = getMimeType(headers);
+		if (mimeType == null) {
+			return true;
+		}
+		if (getSupportedMimeTypes().isEmpty()) {
+			return true;
+		}
+		for (MimeType supported : getSupportedMimeTypes()) {
+			if (supported.getType().equals(mimeType.getType()) &&
+					supported.getSubtype().equals(mimeType.getSubtype())) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	protected MimeType getMimeType(MessageHeaders headers) {
+		return (this.contentTypeResolver != null) ? this.contentTypeResolver.resolve(headers) : null;
+	}
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/converter/ByteArrayMessageConverter.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2002-2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.support.converter;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeTypeUtils;
+
+
+/**
+ * A {@link MessageConverter} that supports MIME type "application/octet-stream" with the
+ * payload converted to and from a byte[].
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class ByteArrayMessageConverter extends AbstractMessageConverter {
+
+
+	public ByteArrayMessageConverter() {
+		super(MimeTypeUtils.APPLICATION_OCTET_STREAM);
+	}
+
+
+	@Override
+	protected boolean supports(Class<?> clazz) {
+		return byte[].class.equals(clazz);
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		return message.getPayload();
+	}
+
+	@Override
+	public Object convertToInternal(Object payload, MessageHeaders headers) {
+		return payload;
+	}
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/converter/CompositeMessageConverter.java

@@ -0,0 +1,110 @@
+/*
+ * Copyright 2002-2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.support.converter;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.Assert;
+
+
+/**
+ * A {@link MessageConverter} that delegates to a list of other converters to invoke until
+ * one of them returns a non-null value.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class CompositeMessageConverter implements MessageConverter {
+
+	private final List<MessageConverter> converters;
+
+	private ContentTypeResolver contentTypeResolver;
+
+
+	/**
+	 * Create a new instance with the given {@link MessageConverter}s in turn configuring
+	 * each with a {@link DefaultContentTypeResolver}.
+	 */
+	public CompositeMessageConverter(Collection<MessageConverter> converters) {
+		this(new ArrayList<MessageConverter>(converters), new DefaultContentTypeResolver());
+	}
+
+	/**
+	 * Create an instance with the given {@link MessageConverter}s and configure all with
+	 * the given {@link ContentTypeResolver}.
+	 */
+	public CompositeMessageConverter(Collection<MessageConverter> converters, ContentTypeResolver resolver) {
+		Assert.notEmpty(converters, "converters is required");
+		Assert.notNull(resolver, "contentTypeResolver is required");
+		this.converters = new ArrayList<MessageConverter>(converters);
+		this.contentTypeResolver = resolver;
+		applyContentTypeResolver(converters, resolver);
+	}
+
+
+	private static void applyContentTypeResolver(Collection<MessageConverter> converters,
+			ContentTypeResolver resolver) {
+
+		for (MessageConverter converter : converters) {
+			if (converter instanceof AbstractMessageConverter) {
+				((AbstractMessageConverter) converter).setContentTypeResolver(resolver);
+			}
+		}
+	}
+
+
+	public void setContentTypeResolver(ContentTypeResolver resolver) {
+		this.contentTypeResolver = resolver;
+		applyContentTypeResolver(getConverters(), resolver);
+	}
+
+	public ContentTypeResolver getContentTypeResolver() {
+		return this.contentTypeResolver;
+	}
+
+	public Collection<MessageConverter> getConverters() {
+		return this.converters;
+	}
+
+
+	@Override
+	public Object fromMessage(Message<?> message, Class<?> targetClass) {
+		for (MessageConverter converter : this.converters) {
+			Object result = converter.fromMessage(message, targetClass);
+			if (result != null) {
+				return result;
+			}
+		}
+		return null;
+	}
+
+	@Override
+	public Message<?> toMessage(Object payload, MessageHeaders headers) {
+		for (MessageConverter converter : this.converters) {
+			Message<?> result = converter.toMessage(payload, headers);
+			if (result != null) {
+				return result;
+			}
+		}
+		return null;
+	}
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/converter/ContentTypeResolver.java

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *      http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -13,28 +13,21 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.springframework.messaging.support.converter;
 
-import java.lang.reflect.Type;
-
-import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeType;
 
 
 /**
- * @author Mark Fisher
+ * Resolve the content type for a message given a set of {@link MessageHeaders}.
+ *
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
-public class SimplePayloadMessageConverter implements MessageConverter<Object> {
+public interface ContentTypeResolver {
 
-	@Override
-	public Message<Object> toMessage(Object object) {
-		return MessageBuilder.withPayload(object).build();
-	}
-
-	@Override
-	public Object fromMessage(Message<?> message, Type targetClass) {
-		return message.getPayload();
-	}
+	MimeType resolve(MessageHeaders headers);
 
 }

spring-messaging/src/main/java/org/springframework/messaging/support/converter/DefaultContentTypeResolver.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2002-2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.support.converter;
+
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeType;
+
+
+/**
+ * A default {@link ContentTypeResolver} that checks the
+ * {@link MessageHeaders#CONTENT_TYPE} header or falls back to a default, if a default is
+ * configured.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class DefaultContentTypeResolver implements ContentTypeResolver {
+
+	private MimeType defaultMimeType;
+
+
+	/**
+	 * Set the default MIME type to use, if the message headers don't have one.
+	 * By default this property is set to {@code null}.
+	 */
+	public void setDefaultMimeType(MimeType defaultMimeType) {
+		this.defaultMimeType = defaultMimeType;
+	}
+
+	/**
+	 * Return the default MIME type to use.
+	 */
+	public MimeType getDefaultMimeType() {
+		return this.defaultMimeType;
+	}
+
+	@Override
+	public MimeType resolve(MessageHeaders headers) {
+		MimeType mimeType = null;
+		if (headers != null) {
+			mimeType = headers.get(MessageHeaders.CONTENT_TYPE, MimeType.class);
+		}
+		return (mimeType != null) ? mimeType : this.defaultMimeType;
+	}
+
+}

spring-messaging/src/main/java/org/springframework/messaging/support/converter/MappingJackson2MessageConverter.java

@@ -20,66 +20,88 @@ import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.StringWriter;
 import java.io.Writer;
-import java.lang.reflect.Type;
-import java.util.Map;
+import java.nio.charset.Charset;
 
 import org.springframework.messaging.Message;
-import org.springframework.messaging.support.MessageBuilder;
-import org.springframework.util.Assert;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeType;
 
+import com.fasterxml.jackson.core.JsonEncoding;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.util.DefaultPrettyPrinter;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
 
 
 /**
+ * A Jackson 2 based {@link MessageConverter} implementation.
+ *
  * @author Rossen Stoyanchev
- * @sicne 4.0
+ * @since 4.0
  */
-public class MappingJackson2MessageConverter implements MessageConverter<Object> {
+public class MappingJackson2MessageConverter extends AbstractMessageConverter {
 
 	private ObjectMapper objectMapper = new ObjectMapper();
 
-	private Type defaultObjectType = Map.class;
-
-	private Class<?> defaultMessagePayloadClass = byte[].class;
+	private Boolean prettyPrint;
 
 
-	/**
-	 * Set the default target Object class to convert to in
-	 * {@link #fromMessage(Message, Class)}.
-	 */
-	public void setDefaultObjectClass(Type defaultObjectType) {
-		Assert.notNull(defaultObjectType, "defaultObjectType is required");
-		this.defaultObjectType = defaultObjectType;
+	public MappingJackson2MessageConverter() {
+		super(new MimeType("application", "json", Charset.forName("UTF-8")));
 	}
 
+
 	/**
-	 * Set the type of Message payload to convert to in {@link #toMessage(Object)}.
-	 * @param payloadClass either byte[] or String
+	 * Whether to use the {@link DefaultPrettyPrinter} when writing JSON.
+	 * This is a shortcut for setting up an {@code ObjectMapper} as follows:
+	 * <pre class="code">
+	 * ObjectMapper mapper = new ObjectMapper();
+	 * mapper.configure(SerializationFeature.INDENT_OUTPUT, true);
+	 * converter.setObjectMapper(mapper);
+	 * </pre>
 	 */
-	public void setDefaultTargetPayloadClass(Class<?> payloadClass) {
-		Assert.isTrue(byte[].class.equals(payloadClass) || String.class.equals(payloadClass),
-				"Payload class must be byte[] or String: " + payloadClass);
-		this.defaultMessagePayloadClass = payloadClass;
+	public void setPrettyPrint(boolean prettyPrint) {
+		this.prettyPrint = prettyPrint;
+		configurePrettyPrint();
+	}
+
+	private void configurePrettyPrint() {
+		if (this.prettyPrint != null) {
+			this.objectMapper.configure(SerializationFeature.INDENT_OUTPUT, this.prettyPrint);
+		}
 	}
 
 	@Override
-	public Object fromMessage(Message<?> message, Type objectType) {
+	protected boolean canConvertFrom(Message<?> message, Class<?> targetClass) {
+		if (targetClass == null) {
+			return false;
+		}
+		JavaType type = this.objectMapper.constructType(targetClass);
+		return (this.objectMapper.canDeserialize(type) && supportsMimeType(message.getHeaders()));
+	}
 
-		JavaType javaType = (objectType != null) ?
-				this.objectMapper.constructType(objectType) :
-					this.objectMapper.constructType(this.defaultObjectType);
+	@Override
+	protected boolean canConvertTo(Object payload, MessageHeaders headers) {
+		return (this.objectMapper.canSerialize(payload.getClass()) && supportsMimeType(headers));
+	}
 
+	@Override
+	protected boolean supports(Class<?> clazz) {
+		// should not be called, since we override canConvertFrom/canConvertTo instead
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		JavaType javaType = this.objectMapper.constructType(targetClass);
 		Object payload = message.getPayload();
 		try {
 			if (payload instanceof byte[]) {
 				return this.objectMapper.readValue((byte[]) payload, javaType);
 			}
-			else if (payload instanceof String) {
-				return this.objectMapper.readValue((String) payload, javaType);
-			}
 			else {
-				throw new IllegalArgumentException("Unexpected message payload type: " + payload);
+				return this.objectMapper.readValue((String) payload, javaType);
 			}
 		}
 		catch (IOException ex) {
@@ -87,30 +109,55 @@ public class MappingJackson2MessageConverter implements MessageConverter<Object>
 		}
 	}
 
-	@SuppressWarnings("unchecked")
 	@Override
-	public <P> Message<P> toMessage(Object object) {
-		P payload;
+	public Object convertToInternal(Object payload, MessageHeaders headers) {
 		try {
-			if (byte[].class.equals(this.defaultMessagePayloadClass)) {
+			if (byte[].class.equals(getSerializedPayloadClass())) {
 				ByteArrayOutputStream out = new ByteArrayOutputStream();
-				this.objectMapper.writeValue(out, object);
-				payload = (P) out.toByteArray();
-			}
-			else if (String.class.equals(this.defaultMessagePayloadClass)) {
-				Writer writer = new StringWriter();
-				this.objectMapper.writeValue(writer, object);
-				payload = (P) writer.toString();
+				JsonEncoding encoding = getJsonEncoding(getMimeType(headers));
+
+				// The following has been deprecated as late as Jackson 2.2 (April 2013);
+				// preserved for the time being, for Jackson 2.0/2.1 compatibility.
+				@SuppressWarnings("deprecation")
+				JsonGenerator generator = this.objectMapper.getJsonFactory().createJsonGenerator(out, encoding);
+
+				// A workaround for JsonGenerators not applying serialization features
+				// https://github.com/FasterXML/jackson-databind/issues/12
+				if (this.objectMapper.isEnabled(SerializationFeature.INDENT_OUTPUT)) {
+					generator.useDefaultPrettyPrinter();
+				}
+
+				this.objectMapper.writeValue(generator, payload);
+				payload = out.toByteArray();
 			}
 			else {
-				// Should never happen..
-				throw new IllegalStateException("Unexpected payload class: " + defaultMessagePayloadClass);
+				Writer writer = new StringWriter();
+				this.objectMapper.writeValue(writer, payload);
+				payload = writer.toString();
 			}
 		}
 		catch (IOException ex) {
 			throw new MessageConversionException("Could not write JSON: " + ex.getMessage(), ex);
 		}
-		return MessageBuilder.withPayload(payload).build();
+		return payload;
+	}
+
+	/**
+	 * Determine the JSON encoding to use for the given content type.
+	 *
+	 * @param contentType the MIME type from the MessageHeaders, if any
+	 * @return the JSON encoding to use (never {@code null})
+	 */
+	protected JsonEncoding getJsonEncoding(MimeType contentType) {
+		if ((contentType != null) && (contentType.getCharSet() != null)) {
+			Charset charset = contentType.getCharSet();
+			for (JsonEncoding encoding : JsonEncoding.values()) {
+				if (charset.name().equals(encoding.getJavaName())) {
+					return encoding;
+				}
+			}
+		}
+		return JsonEncoding.UTF8;
 	}
 
 }

spring-messaging/src/main/java/org/springframework/messaging/support/converter/MessageConverter.java

@@ -16,19 +16,53 @@
 
 package org.springframework.messaging.support.converter;
 
-import java.lang.reflect.Type;
-
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 
 
 /**
+ * A converter to turn the payload of a {@link Message} from serialized form to a typed
+ * Object and vice versa. The {@link MessageHeaders#CONTENT_TYPE} message header may be
+ * used to specify the media type of the message content.
+ *
  * @author Mark Fisher
+ * @author Rossen Stoyanchev
  * @since 4.0
  */
-public interface MessageConverter<T> {
+public interface MessageConverter {
 
-	<P> Message<P> toMessage(T object);
+	/**
+	 * Convert the payload of a {@link Message} from serialized form to a typed Object of
+	 * the specified target class. The {@link MessageHeaders#CONTENT_TYPE} header should
+	 * indicate the MIME type to convert from.
+	 * <p>
+	 * If the converter does not support the specified media type or cannot perform the
+	 * conversion, it should return {@code null}.
+	 *
+	 * @param message the input message
+	 * @param targetClass the target class for the conversion
+	 *
+	 * @return the result of the conversion or {@code null} if the converter cannot
+	 *         perform the conversion
+	 */
+	Object fromMessage(Message<?> message, Class<?> targetClass);
 
-	T fromMessage(Message<?> message, Type targetClass);
+	/**
+	 * Create a {@link Message} whose payload is the result of converting the given
+	 * payload Object to serialized form. The optional {@link MessageHeaders} parameter
+	 * may contain a {@link MessageHeaders#CONTENT_TYPE} header to specify the target
+	 * media type for the conversion and it may contain additional headers to be added to
+	 * the message.
+	 * <p>
+	 * If the converter does not support the specified media type or cannot perform the
+	 * conversion, it should return {@code null}.
+	 *
+	 * @param payload the Object to convert
+	 * @param header optional headers for the message, may be {@code null}
+	 *
+	 * @return the new message or {@code null} if the converter does not support the
+	 *         Object type or the target media type
+	 */
+	Message<?> toMessage(Object payload, MessageHeaders header);
 
 }

spring-messaging/src/main/java/org/springframework/messaging/support/converter/StringMessageConverter.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2002-2013 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.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.messaging.support.converter;
+
+import java.nio.charset.Charset;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.util.MimeType;
+
+
+/**
+ * A {@link MessageConverter} that supports MIME type "text/plain" with the
+ * payload converted to and from a String.
+ *
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class StringMessageConverter extends AbstractMessageConverter {
+
+	private final Charset defaultCharset;
+
+
+	public StringMessageConverter() {
+		this(Charset.forName("UTF-8"));
+	}
+
+	public StringMessageConverter(Charset defaultCharset) {
+		super(new MimeType("text", "plain", defaultCharset));
+		this.defaultCharset = defaultCharset;
+	}
+
+
+	@Override
+	protected boolean supports(Class<?> clazz) {
+		return String.class.equals(clazz);
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		Charset charset = getContentTypeCharset(getMimeType(message.getHeaders()));
+		Object payload = message.getPayload();
+		return (payload instanceof String) ? payload : new String((byte[]) payload, charset);
+	}
+
+	@Override
+	public Object convertToInternal(Object payload, MessageHeaders headers) {
+		if (byte[].class.equals(getSerializedPayloadClass())) {
+			Charset charset = getContentTypeCharset(getMimeType(headers));
+			payload = ((String) payload).getBytes(charset);
+		}
+		return payload;
+	}
+
+	private Charset getContentTypeCharset(MimeType mimeType) {
+		if (mimeType != null && mimeType.getCharSet() != null) {
+			return mimeType.getCharSet();
+		}
+		else {
+			return this.defaultCharset;
+		}
+	}
+
+}