From 9bc2b6f62b35efbeb2d4fbcea6544a85f7e5a821 Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@pivotal.io>
Date: Thu, 29 Sep 2016 21:36:11 -0400
Subject: [PATCH] Fix `DefaultAmqpHeaderMapper` for latest SA

The `correlationId` is `String` now, not `byte[]`
---
 .../amqp/support/DefaultAmqpHeaderMapper.java        | 12 +++++++-----
 .../amqp/support/DefaultAmqpHeaderMapperTests.java   |  3 ++-
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
index 806c4e3af8..03380410a4 100644
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java
@@ -122,8 +122,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 			if (StringUtils.hasText(contentType)) {
 				headers.put(AmqpHeaders.CONTENT_TYPE, contentType);
 			}
-			byte[] correlationId = amqpMessageProperties.getCorrelationId();
-			if (correlationId != null && correlationId.length > 0) {
+			String correlationId = amqpMessageProperties.getCorrelationId();
+			if (StringUtils.hasText(contentType)) {
 				headers.put(AmqpHeaders.CORRELATION_ID, correlationId);
 			}
 			MessageDeliveryMode receivedDeliveryMode = amqpMessageProperties.getReceivedDeliveryMode();
@@ -234,10 +234,12 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		if (StringUtils.hasText(contentType)) {
 			amqpMessageProperties.setContentType(contentType);
 		}
-		Object correlationId = headers.get(AmqpHeaders.CORRELATION_ID);
-		if (correlationId instanceof byte[]) {
-			amqpMessageProperties.setCorrelationId((byte[]) correlationId);
+
+		String correlationId = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class);
+		if (StringUtils.hasText(correlationId)) {
+			amqpMessageProperties.setCorrelationId(correlationId);
 		}
+
 		Integer delay = getHeaderIfAvailable(headers, AmqpHeaders.DELAY, Integer.class);
 		if (delay != null) {
 			amqpMessageProperties.setDelay(delay);
diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
index 8f69dc2e62..96649a7eb7 100644
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapperTests.java
@@ -45,6 +45,7 @@ import org.springframework.util.MimeTypeUtils;
  * @author Gary Russell
  * @author Oleg Zhurakousky
  * @author Stephane Nicoll
+ * @author Artem Bilan
  * @since 2.1
  */
 public class DefaultAmqpHeaderMapperTests {
@@ -158,7 +159,7 @@ public class DefaultAmqpHeaderMapperTests {
 		amqpProperties.setContentEncoding("test.contentEncoding");
 		amqpProperties.setContentLength(99L);
 		amqpProperties.setContentType("test.contentType");
-		byte[] testCorrelationId = new byte[] { 1, 2, 3 };
+		String testCorrelationId = "foo";
 		amqpProperties.setCorrelationId(testCorrelationId);
 		amqpProperties.setReceivedDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
 		amqpProperties.setDeliveryTag(1234L);