Add message converters

- These are the message converters being used in Spring XD; porting them in S-C-S

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/AbstractFromMessageConverter.java

@@ -0,0 +1,204 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.AbstractMessageConverter;
+import org.springframework.messaging.converter.ContentTypeResolver;
+import org.springframework.util.Assert;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeType;
+
+
+/**
+ * Base class for converters applied via Spring Integration 4.x data type channels.
+ *
+ * Extend this class to implement {@link org.springframework.messaging.converter.MessageConverter MessageConverters}
+ * used with custom Message conversion. Only {@link #fromMessage} is supported.
+ *
+ * @author David Turanski
+ */
+public abstract class AbstractFromMessageConverter extends AbstractMessageConverter {
+
+	protected Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	protected final List<MimeType> targetMimeTypes;
+
+	/**
+	 * Creates a converter that ignores content-type message headers
+	 *
+	 * @param targetMimeType the required target type
+	 */
+	protected AbstractFromMessageConverter(MimeType targetMimeType) {
+		this(new ArrayList<MimeType>(), targetMimeType, new StrictContentTypeResolver(targetMimeType));
+	}
+
+	protected AbstractFromMessageConverter(Collection<MimeType> targetMimeTypes) {
+		this(new ArrayList<MimeType>(), targetMimeTypes, new StringConvertingContentTypeResolver());
+	}
+
+	/**
+	 * Creates a converter that handles one or more content-type message headers
+	 *
+	 * @param supportedSourceMimeTypes list of {@link MimeType} that may present in content-type header
+	 * @param targetMimeType the required target type
+	 */
+	protected AbstractFromMessageConverter(Collection<MimeType> supportedSourceMimeTypes, MimeType targetMimeType,
+			ContentTypeResolver contentTypeResolver) {
+		super(supportedSourceMimeTypes);
+		Assert.notNull(targetMimeType, "'targetMimeType' cannot be null");
+		setContentTypeResolver(contentTypeResolver);
+		this.targetMimeTypes = Collections.singletonList(targetMimeType);
+	}
+
+	/**
+	 * Creates a converter that handles one or more content-type message headers and one or more target MIME types
+	 *
+	 * @param supportedSourceMimeTypes a list of supported content types
+	 * @param targetMimeTypes a list of supported target types
+	 * @param contentTypeResolver the {@link ContentTypeResolver} to use
+	 */
+	protected AbstractFromMessageConverter(Collection<MimeType> supportedSourceMimeTypes,
+			Collection<MimeType> targetMimeTypes,
+			ContentTypeResolver contentTypeResolver) {
+		super(supportedSourceMimeTypes);
+		Assert.notNull(targetMimeTypes, "'targetMimeTypes' cannot be null");
+		setContentTypeResolver(contentTypeResolver);
+		this.targetMimeTypes = new ArrayList<MimeType>(targetMimeTypes);
+	}
+
+	/**
+	 * Creates a converter that requires a specific content-type message header
+	 *
+	 * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header
+	 * @param targetMimeType the required target type
+	 */
+	protected AbstractFromMessageConverter(MimeType supportedSourceMimeType, MimeType targetMimeType) {
+		this(Collections.singletonList(supportedSourceMimeType), targetMimeType, new StrictContentTypeResolver(
+				supportedSourceMimeType));
+	}
+
+	/**
+	 * Creates a converter that requires a specific content-type message header and supports multiple target MIME types.
+	 *
+	 * @param supportedSourceMimeType {@link MimeType} that must be present in content-type header
+	 * @param targetMimeTypes a list of supported target types
+	 */
+	protected AbstractFromMessageConverter(MimeType supportedSourceMimeType, Collection<MimeType> targetMimeTypes) {
+		this(Collections.singletonList(supportedSourceMimeType), targetMimeTypes, new StrictContentTypeResolver(
+				supportedSourceMimeType));
+	}
+
+	/**
+	 * Subclasses implement this to specify supported target types
+	 *
+	 * @return an array of supported classes or null if any target type is supported
+	 */
+	protected abstract Class<?>[] supportedTargetTypes();
+
+	/**
+	 * Subclasses implement this to specify supported payload types
+	 *
+	 * @return an array of supported classes or null if any target type is supported
+	 */
+	protected abstract Class<?>[] supportedPayloadTypes();
+
+	protected boolean supportsPayloadType(Class<?> clazz) {
+		return supportsType(clazz, supportedPayloadTypes());
+	}
+
+	@Override
+	protected boolean supports(Class<?> clazz) {
+		return supportsType(clazz, supportedTargetTypes());
+	}
+
+	private boolean supportsType(Class<?> clazz, Class<?>[] supportedTypes) {
+		if (supportedTypes != null) {
+			for (Class<?> targetType : supportedTypes) {
+				if (ClassUtils.isAssignable(clazz, targetType)) {
+					return true;
+				}
+			}
+			return false;
+		}
+		return true;
+	}
+
+	@Override
+	protected boolean canConvertFrom(Message<?> message, Class<?> targetClass) {
+		return super.canConvertFrom(message, targetClass) && supportsPayloadType(message.getPayload().getClass());
+	}
+
+	public boolean supportsTargetMimeType(MimeType mimeType) {
+		for (MimeType targetMimeType : targetMimeTypes) {
+			if (mimeType.getType().equals(targetMimeType.getType()) && mimeType.getSubtype().equals(
+					targetMimeType.getSubtype())) {
+				return true;
+			}
+		}
+		return false;
+	}
+
+	@Override
+	// TODO: This will likely be fixed in core Spring
+	public void setContentTypeResolver(ContentTypeResolver resolver) {
+		if (getContentTypeResolver() == null) {
+			super.setContentTypeResolver(resolver);
+		}
+	}
+
+	/**
+	 * Not supported by default
+	 */
+	@Override
+	protected boolean canConvertTo(Object payload, MessageHeaders headers) {
+		return false;
+	}
+
+	/**
+	 * Not supported by default
+	 */
+	@Override
+	public Object convertToInternal(Object payload, MessageHeaders headers) {
+		throw new UnsupportedOperationException("'convertTo' not supported");
+	}
+
+	/**
+	 * Convenience method to construct a converted message
+	 *
+	 * @param payload the converted payload
+	 * @param headers the existing message headers
+	 * @param contentType the value of the content-type header
+	 * @return the converted message
+	 */
+	protected final Message<?> buildConvertedMessage(Object payload, MessageHeaders headers, MimeType contentType) {
+		return MessageBuilder.withPayload(payload).copyHeaders(headers)
+				.copyHeaders(
+						Collections.singletonMap(MessageHeaders.CONTENT_TYPE,
+								contentType)).build();
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ByteArrayToStringMessageConverter.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.io.UnsupportedEncodingException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.ContentTypeResolver;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert from byte[] to String applying the Charset provided in
+ * the content-type header if any.
+ *
+ * @author David Turanski
+ */
+public class ByteArrayToStringMessageConverter extends AbstractFromMessageConverter {
+
+	private final static ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
+
+	private final static List<MimeType> targetMimeTypes = new ArrayList<MimeType>();
+	static {
+		targetMimeTypes.add(MessageConverterUtils.X_SPRING_STRING);
+		targetMimeTypes.add(MessageConverterUtils.X_JAVA_OBJECT);
+		targetMimeTypes.add(MimeTypeUtils.TEXT_PLAIN);
+	}
+
+	public ByteArrayToStringMessageConverter() {
+		super(Arrays.asList(new MimeType[] { MimeTypeUtils.APPLICATION_OCTET_STREAM, MimeTypeUtils.TEXT_PLAIN }),
+				targetMimeTypes, contentTypeResolver);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { String.class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] { byte[].class };
+	}
+
+	/**
+	 * Don't need to manipulate message headers. Just return payload
+	 */
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		MimeType mimeType = contentTypeResolver.resolve(message.getHeaders());
+
+		String converted = null;
+
+		if (mimeType == null || mimeType.getParameter("Charset") == null) {
+			converted = new String((byte[]) message.getPayload());
+		}
+		else {
+			String encoding = mimeType.getParameter("Charset");
+			if (encoding != null) {
+				try {
+					converted = new String((byte[]) message.getPayload(), encoding);
+				}
+				catch (UnsupportedEncodingException e) {
+					logger.error(e.getMessage(), e);
+				}
+			}
+		}
+		return converted;
+
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CompositeMessageConverterFactory.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.util.Assert;
+import org.springframework.util.CollectionUtils;
+import org.springframework.util.MimeType;
+
+
+/**
+ * A factory for creating an instance of {@link CompositeMessageConverter} for a given target MIME type
+ *
+ * @author David Turanski
+ */
+public class CompositeMessageConverterFactory {
+
+	private final List<AbstractFromMessageConverter> converters;
+
+	/**
+	 * @param converters a list of {@link AbstractFromMessageConverter}
+	 */
+	public CompositeMessageConverterFactory(Collection<AbstractFromMessageConverter> converters) {
+		Assert.notNull(converters, "'converters' cannot be null");
+		this.converters = new ArrayList<AbstractFromMessageConverter>(converters);
+	}
+
+	/**
+	 * Creation method.
+	 *
+	 * @param targetMimeType the target MIME type
+	 * @return a converter for the target MIME type
+	 */
+	public CompositeMessageConverter newInstance(MimeType targetMimeType) {
+		List<MessageConverter> targetMimeTypeConverters = new ArrayList<MessageConverter>();
+		for (AbstractFromMessageConverter converter : converters) {
+			if (converter.supportsTargetMimeType(targetMimeType)) {
+				targetMimeTypeConverters.add(converter);
+			}
+		}
+		if (CollectionUtils.isEmpty(targetMimeTypeConverters)) {
+			throw new ConversionException("No message converter is registered for "
+					+ targetMimeType.toString());
+		}
+		return new CompositeMessageConverter(targetMimeTypeConverters);
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ConversionException.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+/**
+ * Exception thrown when an error is encountered during message conversion.
+ * 
+ * @author David Turanski
+ */
+@SuppressWarnings("serial")
+public class ConversionException extends RuntimeException {
+
+	public ConversionException(String message) {
+		super(message);
+	}
+
+	public ConversionException(String message, Throwable t) {
+		super(message, t);
+	}
+
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/CustomMimeTypeConverter.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 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.cloud.stream.converter;
+
+import org.springframework.core.convert.converter.Converter;
+import org.springframework.http.MediaType;
+import org.springframework.util.MimeType;
+
+/**
+ * A custom converter for {@link MediaType} that accepts a plain java class name as a shorthand for
+ * {@code application/x-java-object;type=the.qualified.ClassName}.
+ * 
+ * 
+ * @author Eric Bottard
+ * @author David Turanski
+ */
+public class CustomMimeTypeConverter implements Converter<String, MimeType> {
+
+	@Override
+	public MimeType convert(String source) {
+		if (!source.contains("/")) {
+			return MimeType.valueOf("application/x-java-object;type=" + source);
+		}
+		return MimeType.valueOf(source);
+	}
+
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JavaToSerializedMessageConverter.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
+import org.springframework.messaging.Message;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert from a POJO to byte[] with Java.io serialization if any.
+ *
+ * @author David Turanski
+ */
+public class JavaToSerializedMessageConverter extends AbstractFromMessageConverter {
+
+	public JavaToSerializedMessageConverter() {
+		super(MessageConverterUtils.X_JAVA_OBJECT, MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { byte[].class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] { Serializable.class };
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		ByteArrayOutputStream bos = new ByteArrayOutputStream();
+		try {
+			new ObjectOutputStream(bos).writeObject(message.getPayload());
+		}
+		catch (IOException e) {
+			logger.error(e.getMessage(), e);
+			return null;
+		}
+
+		return buildConvertedMessage(bos.toByteArray(), message.getHeaders(),
+				MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT);
+	}
+
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToPojoMessageConverter.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import org.springframework.messaging.Message;
+import org.springframework.util.MimeTypeUtils;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+
+/**
+ *
+ * @author David Turanski
+ */
+public class JsonToPojoMessageConverter extends AbstractFromMessageConverter {
+
+	private final ObjectMapper mapper = new ObjectMapper();
+
+	public JsonToPojoMessageConverter() {
+		super(MimeTypeUtils.APPLICATION_JSON, MessageConverterUtils.X_JAVA_OBJECT);
+		mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] {String.class, byte[].class};
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return null; // any type
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		Object result = null;
+		try {
+			Object payload = message.getPayload();
+
+			if (payload instanceof byte[]) {
+				result = mapper.readValue((byte[]) payload, targetClass);
+			}
+			else if (payload instanceof String) {
+				result = mapper.readValue((String) payload, targetClass);
+			}
+		}
+		catch (Exception e) {
+			logger.error(e.getMessage(), e);
+			return null;
+		}
+		return buildConvertedMessage(result, message.getHeaders(),
+				MessageConverterUtils.javaObjectMimeType(targetClass));
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/JsonToTupleMessageConverter.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.messaging.Message;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.cloud.stream.tuple.Tuple;
+import org.springframework.cloud.stream.tuple.TupleBuilder;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert from a JSON (byte[] or String) to a {@link Tuple}.
+ *
+ * @author David Turanski
+ */
+public class JsonToTupleMessageConverter extends AbstractFromMessageConverter {
+
+	private final static List<MimeType> targetMimeTypes = new ArrayList<MimeType>();
+
+	static {
+		targetMimeTypes.add(MessageConverterUtils.X_SPRING_TUPLE);
+		targetMimeTypes.add(MessageConverterUtils.X_JAVA_OBJECT);
+	}
+
+	public JsonToTupleMessageConverter() {
+		super(MimeTypeUtils.APPLICATION_JSON, targetMimeTypes);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { Tuple.class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] { byte[].class, String.class };
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		String source = null;
+		if (message.getPayload() instanceof byte[]) {
+			source = new String((byte[]) message.getPayload());
+		}
+		else {
+			source = (String) message.getPayload();
+		}
+		Tuple t = TupleBuilder.fromString(source);
+		return buildConvertedMessage(t, message.getHeaders(), MessageConverterUtils.javaObjectMimeType(t.getClass()));
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/MessageConverterUtils.java

@@ -0,0 +1,135 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import static org.springframework.util.MimeType.valueOf;
+import static org.springframework.util.MimeTypeUtils.APPLICATION_JSON;
+import static org.springframework.util.MimeTypeUtils.APPLICATION_OCTET_STREAM;
+
+import org.springframework.cloud.stream.tuple.Tuple;
+import org.springframework.util.ClassUtils;
+import org.springframework.util.MimeType;
+import org.springframework.cloud.stream.tuple.DefaultTuple;
+import org.springframework.util.StringUtils;
+
+
+/**
+ * Message conversion utility methods.
+ *
+ * @author David Turanski
+ * @author Gary Russell
+ */
+public class MessageConverterUtils {
+
+	/**
+	 * An MimeType specifying a {@link Tuple}.
+	 */
+	public static final MimeType X_SPRING_TUPLE = MimeType.valueOf("application/x-spring-tuple");
+
+	/**
+	 * An MimeType for specifying a String.
+	 */
+	public static final MimeType X_SPRING_STRING = MimeType.valueOf("application/x-spring-string");
+
+	/**
+	 * A general MimeType for Java Types.
+	 */
+	public static final MimeType X_JAVA_OBJECT = MimeType.valueOf("application/x-java-object");
+
+	/**
+	 * A general MimeType for a Java serialized byte array.
+	 */
+	public static final MimeType X_JAVA_SERIALIZED_OBJECT = MimeType.valueOf("application/x-java-serialized-object");
+
+	/**
+	 * Map the contentType to a target class.
+	 *
+	 * @param contentType the content type
+	 * @param classLoader the class loader used to resolve the class
+	 * @return the class for the content type
+	 */
+	public static Class<?> getJavaTypeForContentType(MimeType contentType, ClassLoader classLoader) {
+		if (X_JAVA_OBJECT.includes(contentType)) {
+			if (contentType.getParameter("type") != null) {
+				try {
+					return ClassUtils.forName(contentType.getParameter("type"), classLoader);
+				}
+				catch (Exception e) {
+					throw new ConversionException(e.getMessage(), e);
+				}
+			}
+			else {
+				return Object.class;
+			}
+		}
+		else if (APPLICATION_JSON.equals(contentType)) {
+			return String.class;
+		}
+		else if (valueOf("text/*").includes(contentType)) {
+			return String.class;
+		}
+		else if (X_SPRING_TUPLE.includes(contentType)) {
+			return DefaultTuple.class;
+		}
+		else if (APPLICATION_OCTET_STREAM.includes(contentType)) {
+			return byte[].class;
+		}
+		else if (X_JAVA_SERIALIZED_OBJECT.includes(contentType)) {
+			return byte[].class;
+		}
+		else if (X_SPRING_STRING.includes(contentType)) {
+			return String.class;
+		}
+		return null;
+	}
+
+	/**
+	 * Build the conventional {@link MimeType} for a java object
+	 *
+	 * @param clazz the java type
+	 * @return the MIME type
+	 */
+	public static MimeType javaObjectMimeType(Class<?> clazz) {
+		return MimeType.valueOf("application/x-java-object;type=" + clazz.getName());
+	}
+
+	public static MimeType getMimeType(String contentTypeString) {
+		MimeType mimeType = null;
+		if (StringUtils.hasText(contentTypeString)) {
+			try {
+				mimeType = resolveContentType(contentTypeString);
+			}
+			catch (ClassNotFoundException cfe) {
+				throw new IllegalArgumentException("Could not find the class required for " + contentTypeString, cfe);
+			}
+		}
+		return mimeType;
+	}
+
+	public static MimeType resolveContentType(String type) throws ClassNotFoundException, LinkageError {
+		if (!type.contains("/")) {
+			Class<?> javaType = resolveJavaType(type);
+			return MessageConverterUtils.javaObjectMimeType(javaType);
+		}
+		return MimeType.valueOf(type);
+	}
+
+	public static Class<?> resolveJavaType(String type) throws ClassNotFoundException, LinkageError {
+		return ClassUtils.forName(type, Thread.currentThread().getContextClassLoader());
+	}
+
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToJsonMessageConverter.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.messaging.Message;
+import org.springframework.util.MimeTypeUtils;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert a Java object to a JSON String
+ *
+ * @author David Turanski
+ * @author David Liu
+ */
+public class PojoToJsonMessageConverter extends AbstractFromMessageConverter {
+
+	private final ObjectMapper mapper = new ObjectMapper();
+
+	@Value("${typeconversion.json.prettyPrint:false}")
+	private volatile boolean prettyPrint;
+
+	public PojoToJsonMessageConverter() {
+		super(MimeTypeUtils.APPLICATION_JSON);
+		mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] {String.class};
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return null;
+	}
+
+
+	public void setPrettyPrint(boolean prettyPrint) {
+		this.prettyPrint = prettyPrint;
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		Object result;
+		try {
+			if (prettyPrint) {
+				result = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(message.getPayload());
+			}
+			else {
+				result = mapper.writeValueAsString(message.getPayload());
+			}
+		}
+		catch (JsonProcessingException e) {
+			logger.error(e.getMessage(), e);
+			return null;
+		}
+		return buildConvertedMessage(result, message.getHeaders(), MimeTypeUtils.APPLICATION_JSON);
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/PojoToStringMessageConverter.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import org.springframework.messaging.Message;
+import org.springframework.util.MimeTypeUtils;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert a Java object to a String using toString()
+ *
+ * @author David Turanski
+ */
+public class PojoToStringMessageConverter extends AbstractFromMessageConverter {
+
+	public PojoToStringMessageConverter() {
+		super(MimeTypeUtils.TEXT_PLAIN);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { String.class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return null;
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		return buildConvertedMessage(message.getPayload().toString(), message.getHeaders(), MimeTypeUtils.TEXT_PLAIN);
+	}
+
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/SerializedToJavaMessageConverter.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.io.ByteArrayInputStream;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+
+import org.springframework.messaging.Message;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to deserialize {@link Serializable} Java objects.
+ *
+ * @author David Turanski
+ */
+public class SerializedToJavaMessageConverter extends AbstractFromMessageConverter {
+
+	public SerializedToJavaMessageConverter() {
+		super(MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT, MessageConverterUtils.X_JAVA_OBJECT);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { Serializable.class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] { byte[].class };
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		ByteArrayInputStream bis = new ByteArrayInputStream((byte[]) (message.getPayload()));
+		Object result = null;
+		try {
+			result = new ObjectInputStream(bis).readObject();
+		}
+		catch (Exception e) {
+			logger.error(e.getMessage(), e);
+			return null;
+		}
+
+		return buildConvertedMessage(result, message.getHeaders(),
+				MessageConverterUtils.javaObjectMimeType(targetClass));
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StrictContentTypeResolver.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import org.springframework.util.MimeType;
+
+
+/**
+ * A {@link StringConvertingContentTypeResolver} that requires a the content-type to be present.
+ * 
+ * @author David Turanski
+ */
+// TODO: This will likely be pushed to core Spring
+public class StrictContentTypeResolver extends StringConvertingContentTypeResolver {
+
+	/**
+	 * @param defaultMimeType the required {@link MimeType}
+	 */
+	public StrictContentTypeResolver(MimeType defaultMimeType) {
+		super();
+		setDefaultMimeType(defaultMimeType);
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringConvertingContentTypeResolver.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.converter.DefaultContentTypeResolver;
+import org.springframework.util.MimeType;
+
+/**
+ * A {@link DefaultContentTypeResolver} that can parse String values.
+ * 
+ * @author David Turanski
+ */
+public class StringConvertingContentTypeResolver extends DefaultContentTypeResolver {
+
+	private ConcurrentMap<String,MimeType> mimeTypeCache = new ConcurrentHashMap<>();
+
+	@Override
+	public MimeType resolve(MessageHeaders headers) {
+		return resolve((Map<String, Object>) headers);
+	}
+
+	public MimeType resolve(Map<String,Object> headers) {
+		Object value = headers.get(MessageHeaders.CONTENT_TYPE);
+		if (value instanceof MimeType) {
+			return (MimeType) value;
+		}
+		else if (value instanceof String) {
+			MimeType mimeType = mimeTypeCache.get(value);
+			if (mimeType == null) {
+				String valueAsString = (String) value;
+				mimeType = MimeType.valueOf(valueAsString);
+				mimeTypeCache.put(valueAsString,mimeType);
+			}
+			return mimeType;
+		}
+		return getDefaultMimeType();
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/StringToByteArrayMessageConverter.java

@@ -0,0 +1,83 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.io.UnsupportedEncodingException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.ContentTypeResolver;
+import org.springframework.util.MimeType;
+import org.springframework.util.MimeTypeUtils;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert a String to a byte[], applying the provided Charset in
+ * the content-type header if any.
+ *
+ * @author David Turanski
+ */
+public class StringToByteArrayMessageConverter extends AbstractFromMessageConverter {
+
+	private final static ContentTypeResolver contentTypeResolver = new StringConvertingContentTypeResolver();
+
+	private final static List<MimeType> targetMimeTypes = new ArrayList<MimeType>();
+	static {
+		targetMimeTypes.add(MimeTypeUtils.APPLICATION_OCTET_STREAM);
+	}
+
+	public StringToByteArrayMessageConverter() {
+		super(targetMimeTypes);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { byte[].class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] { String.class };
+	}
+
+	/**
+	 * Don't need to manipulate message headers. Just return the payload
+	 */
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		MimeType mimeType = contentTypeResolver.resolve(message.getHeaders());
+		byte[] converted = null;
+		if (mimeType == null || mimeType.getParameter("Charset") == null) {
+			converted = ((String) message.getPayload()).getBytes();
+		}
+		else {
+			String encoding = mimeType.getParameter("Charset");
+			if (encoding != null) {
+				try {
+					converted = ((String) message.getPayload()).getBytes(encoding);
+				}
+				catch (UnsupportedEncodingException e) {
+					logger.error(e.getMessage(), e);
+				}
+			}
+		}
+		return converted;
+
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/TupleToJsonMessageConverter.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright 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.
+ * 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.cloud.stream.converter;
+
+import java.io.IOException;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.messaging.Message;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.cloud.stream.tuple.Tuple;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+
+/**
+ * A {@link org.springframework.messaging.converter.MessageConverter}
+ * to convert a {@link Tuple} to a JSON String
+ *
+ * @author David Turanski
+ */
+public class TupleToJsonMessageConverter extends AbstractFromMessageConverter {
+
+	@Value("${typeconversion.json.prettyPrint:false}")
+	private volatile boolean prettyPrint;
+
+	public void setPrettyPrint(boolean prettyPrint) {
+		this.prettyPrint = prettyPrint;
+	}
+
+	public TupleToJsonMessageConverter() {
+		super(MimeTypeUtils.APPLICATION_JSON);
+	}
+
+	@Override
+	protected Class<?>[] supportedTargetTypes() {
+		return new Class<?>[] { String.class };
+	}
+
+	@Override
+	protected Class<?>[] supportedPayloadTypes() {
+		return new Class<?>[] { Tuple.class };
+	}
+
+	@Override
+	public Object convertFromInternal(Message<?> message, Class<?> targetClass) {
+		Tuple t = (Tuple) message.getPayload();
+		String json;
+		if (prettyPrint) {
+			ObjectMapper mapper = new ObjectMapper();
+			mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
+			try {
+				Object tmp = mapper.readValue(t.toString(), Object.class);
+				json = mapper.writerWithDefaultPrettyPrinter().writeValueAsString(tmp);
+			}
+			catch (IOException e) {
+				logger.error(e.getMessage(), e);
+				return null;
+			}
+		}
+		else {
+			json = t.toString();
+		}
+		return buildConvertedMessage(json, message.getHeaders(), MimeTypeUtils.APPLICATION_JSON);
+	}
+
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/package-info.java

@@ -0,0 +1,22 @@
+/*
+ * Copyright 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.
+ * 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 for message converters.
+ */
+
+package org.springframework.cloud.stream.converter;