Minor updates for custom content types

- renamed MessageConverterUtils.getJavaType to MessageConverterUtils.getJavaTypeForJavaObjectContentType and restricted usage only to 'application/x-java-object' types

spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java

@@ -4,10 +4,13 @@ import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.isA;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.notNullValue;
 import static org.junit.Assert.assertTrue;
 
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
+import org.junit.Assert;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -52,8 +55,10 @@ public class CustomMessageConverterTests {
 		assertThat(customMessageConverters, hasItem(isA(FooConverter.class)));
 		assertThat(customMessageConverters, hasItem(isA(BarConverter.class)));
 		testSource.output().send(MessageBuilder.withPayload(new Foo("hi")).build());
+		@SuppressWarnings("unchecked")
 		Message<String> received = (Message<String>) ((TestSupportBinder) binderFactory.getBinder(null))
-				.messageCollector().forChannel(testSource.output()).poll();
+				.messageCollector().forChannel(testSource.output()).poll(1, TimeUnit.SECONDS);
+		Assert.assertThat(received, notNullValue());
 		assertThat(received.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(),
 				equalTo("application/x-java-object;type=org.springframework.cloud.stream.config.CustomMessageConverterTests$Bar"));
 	}

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

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.converter;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
 
@@ -25,13 +26,14 @@ import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.MimeType;
+import org.springframework.util.ObjectUtils;
 
 
 /**
  * A factory for creating an instance of {@link CompositeMessageConverter} for a given target MIME type
- *
  * @author David Turanski
  * @author Ilayaperumal Gopinathan
+ * @author Marius Bogoevici
  */
 public class CompositeMessageConverterFactory {
 
@@ -42,12 +44,11 @@ public class CompositeMessageConverterFactory {
 	 */
 	public CompositeMessageConverterFactory(Collection<AbstractFromMessageConverter> converters) {
 		Assert.notNull(converters, "'converters' cannot be null");
-		this.converters = new ArrayList<AbstractFromMessageConverter>(converters);
+		this.converters = new ArrayList<>(converters);
 	}
 
 	/**
 	 * Creation method.
-	 *
 	 * @param targetMimeType the target MIME type
 	 * @return a converter for the target MIME type
 	 */
@@ -69,23 +70,18 @@ public class CompositeMessageConverterFactory {
 		List<Class<?>> supportedDataTypes = new ArrayList<>();
 		// Make sure to check if the target type is of explicit java object type.
 		if (MessageConverterUtils.X_JAVA_OBJECT.includes(targetMimeType)) {
-			supportedDataTypes.add(MessageConverterUtils.getJavaTypeForContentType(targetMimeType));
+			supportedDataTypes.add(MessageConverterUtils.getJavaTypeForJavaObjectContentType(targetMimeType));
 		}
 		else {
 			for (AbstractFromMessageConverter converter : converters) {
 				if (converter.supportsTargetMimeType(targetMimeType)) {
 					Class<?>[] targetTypes = converter.supportedTargetTypes();
-					if (targetTypes != null) {
-						Class<?>[] dataTypes = converter.supportedTargetTypes();
-						for (Class<?> dataType : dataTypes) {
-							if (!supportedDataTypes.contains(dataType)) {
-								supportedDataTypes.add(dataType);
-							}
-						}
+					if (!ObjectUtils.isEmpty(targetTypes)) {
+						supportedDataTypes.addAll(Arrays.asList(targetTypes));
 					}
 				}
 			}
 		}
-		return supportedDataTypes.toArray(new Class<?>[0]);
+		return supportedDataTypes.toArray(new Class<?>[supportedDataTypes.size()]);
 	}
 }

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

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.converter;
 
 import org.springframework.tuple.Tuple;
+import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 import org.springframework.util.MimeType;
 import org.springframework.util.StringUtils;
@@ -51,20 +52,18 @@ public class MessageConverterUtils {
 	 *
 	 * @return the class for the content type
 	 */
-	public static Class<?> getJavaTypeForContentType(MimeType contentType) {
-		Class<?> javaType = Object.class;
-		if (X_JAVA_OBJECT.includes(contentType)) {
-			if (contentType.getParameter("type") != null) {
-				try {
-					javaType = ClassUtils.forName(contentType.getParameter("type"),
-							Thread.currentThread().getContextClassLoader());
-				}
-				catch (Exception e) {
-					throw new ConversionException(e.getMessage(), e);
-				}
+	public static Class<?> getJavaTypeForJavaObjectContentType(MimeType contentType) {
+		Assert.isTrue(X_JAVA_OBJECT.includes(contentType), "Content type must be " + X_JAVA_OBJECT.toString() + ", or " +
+				"included in it");
+		if (contentType.getParameter("type") != null) {
+			try {
+				return ClassUtils.forName(contentType.getParameter("type"), null);
+			}
+			catch (Exception e) {
+				throw new ConversionException(e.getMessage(), e);
 			}
 		}
-		return javaType;
+		return Object.class;
 	}
 
 	/**
@@ -93,13 +92,13 @@ public class MessageConverterUtils {
 	public static MimeType resolveContentType(String type) throws ClassNotFoundException, LinkageError {
 		if (!type.contains("/")) {
 			Class<?> javaType = resolveJavaType(type);
-			return MessageConverterUtils.javaObjectMimeType(javaType);
+			return javaObjectMimeType(javaType);
 		}
 		return MimeType.valueOf(type);
 	}
 
 	public static Class<?> resolveJavaType(String type) throws ClassNotFoundException, LinkageError {
-		return ClassUtils.forName(type, Thread.currentThread().getContextClassLoader());
+		return ClassUtils.forName(type, null);
 	}
 
 }