Merge pull request #588 from olegz/INT-2713

build.gradle

@@ -150,6 +150,7 @@ project('spring-integration-amqp') {
 		}
 		testCompile project(":spring-integration-stream")
 		testCompile project(":spring-integration-test")
+		testCompile project(":spring-integration-http") // need to test INT-2713
 	}
 	bundlor {
 		bundleSymbolicName = 'org.springframework.integration.amqp'

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java

@@ -35,21 +35,21 @@ import org.springframework.util.StringUtils;
  * <p/>
  * By default this implementation will only copy AMQP properties (e.g. contentType) to and from
  * Spring Integration MessageHeaders. Any user-defined headers within the AMQP
- * MessageProperties will NOT be copied to or from an AMQP Message unless 
- * explicitly identified via 'requestHeaderNames' and/or 'replyHeaderNames' 
- * (see {@link AbstractHeaderMapper#setRequestHeaderNames(String[])} 
- * and {@link AbstractHeaderMapper#setReplyHeaderNames(String[])}} 
+ * MessageProperties will NOT be copied to or from an AMQP Message unless
+ * explicitly identified via 'requestHeaderNames' and/or 'replyHeaderNames'
+ * (see {@link AbstractHeaderMapper#setRequestHeaderNames(String[])}
+ * and {@link AbstractHeaderMapper#setReplyHeaderNames(String[])}}
  * as well as 'mapped-request-headers' and 'mapped-reply-headers' attributes of the AMQP adapters).
  * If you need to copy all user-defined headers simply use wild-card character '*'.
  * <p/>
  * Constants for the AMQP header keys are defined in {@link AmqpHeaders}.
- * 
+ *
  * @author Mark Fisher
  * @author Oleg Zhurakousky
  * @since 2.1
  */
 public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessageProperties> implements AmqpHeaderMapper {
-	
+
 	private static final List<String> STANDARD_HEADER_NAMES = new ArrayList<String>();
 
 	static {
@@ -212,7 +212,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		if (contentLength != null) {
 			amqpMessageProperties.setContentLength(contentLength);
 		}
-		String contentType = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_TYPE, String.class);
+		String contentType = this.extractContentTypeAsString(headers);
+
 		if (StringUtils.hasText(contentType)) {
 			amqpMessageProperties.setContentType(contentType);
 		}
@@ -309,4 +310,36 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		return AmqpHeaders.PREFIX;
 	}
 
+	/**
+	 * Will extract Content-Type from MessageHeaders and convert it to String if possible
+	 * Required since Content-Type can be represented as org.springframework.http.MediaType
+	 * see INT-2713 for more details
+	 *
+	 * @param headers
+	 * @return
+	 */
+	private String extractContentTypeAsString(Map<String, Object> headers){
+		String contentTypeStringValue = null;
+
+		Object contentType = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_TYPE, Object.class);
+
+		if (contentType != null){
+			String contentTypeClassName = contentType.getClass().getName();
+
+			if (contentTypeClassName.equals("org.springframework.http.MediaType")){ // see INT-2713
+				contentTypeStringValue = contentType.toString();
+			}
+			else if (contentType instanceof String) {
+				contentTypeStringValue = (String) contentType;
+			}
+			else {
+				if (logger.isWarnEnabled()) {
+					logger.warn("skipping header '" + AmqpHeaders.CONTENT_TYPE +
+							"' since it is not of expected type [" + contentTypeClassName + "]");
+				}
+			}
+		}
+		return contentTypeStringValue;
+	}
+
 }

spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java

@@ -16,6 +16,9 @@
 
 package org.springframework.integration.amqp.support;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
 import java.util.Date;
 import java.util.HashMap;
 import java.util.Map;
@@ -28,15 +31,14 @@ import org.springframework.amqp.core.MessageDeliveryMode;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.support.converter.JsonMessageConverter;
+import org.springframework.http.MediaType;
 import org.springframework.integration.MessageHeaders;
 import org.springframework.integration.amqp.AmqpHeaders;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
 /**
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Oleg Zhurakousky
  * @since 2.1
  */
 public class DefaultAmqpHeaderMapperTests {
@@ -96,6 +98,21 @@ public class DefaultAmqpHeaderMapperTests {
 		assertEquals("test.replyTo2", amqpProperties.getHeaders().get(RabbitTemplate.STACKED_REPLY_TO_HEADER));
 	}
 
+	@Test
+	public void fromHeadersWithContentTypeAsMediaType() {
+		DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
+		Map<String, Object> headerMap = new HashMap<String, Object>();
+
+		MediaType contentType = MediaType.parseMediaType("text/html");
+		headerMap.put(AmqpHeaders.CONTENT_TYPE, contentType);
+
+		MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
+		MessageProperties amqpProperties = new MessageProperties();
+		headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
+
+		assertEquals("text/html", amqpProperties.getContentType());
+	}
+
 	@Test
 	public void toHeaders() {
 		DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();

spring-integration-core/src/main/java/org/springframework/integration/mapping/AbstractHeaderMapper.java

@@ -26,7 +26,6 @@ import java.util.Map;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.integration.MessageHeaders;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -135,7 +134,7 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
 	public Map<String, Object> toHeadersFromReply(T source) {
 		return this.toHeaders(source, this.replyHeaderNames);
 	}
-	
+
 	private void fromHeaders(MessageHeaders headers, T target, List<String> headerPatterns){
 		try {
 			Map<String, Object> subset = new HashMap<String, Object>();
@@ -248,12 +247,16 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
 		}
 		if (!type.isAssignableFrom(value.getClass())) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "]");
+				logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "], it is [" +
+                        value.getClass() + "]");
 			}
+			return null;
+		}
+		else {
+			return (V) value;
 		}
-		return (V) value;
 	}
-	
+
 	private boolean containsElementIgnoreCase(List<String> headerNames, String name) {
 		for (String headerName : headerNames) {
 			if (headerName.equalsIgnoreCase(name)){
@@ -287,7 +290,7 @@ public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMappe
 	protected List<String> getStandardRequestHeaderNames(){
 		return Collections.emptyList();
 	}
-	
+
 	/**
 	 * Returns the list of standard REPLY headers. Implementation provided by a subclass
 	 */