Populate a JsonHeaders.RESOLVABLE_TYPE on reply (#3163)

* Populate a JsonHeaders.RESOLVABLE_TYPE on reply Fixes https://github.com/spring-projects/spring-integration-samples/issues/277 In the `AbstractAmqpOutboundEndpoint` the `JsonHeaders.RESOLVABLE_TYPE` from request message is copied to reply message making inconsistency downstream. The `JsonToObjectTransformer` consults first a `JsonHeaders.RESOLVABLE_TYPE` and deserialize payload to wrong type * Fix `DefaultAmqpHeaderMapper` to populate a `JsonHeaders.RESOLVABLE_TYPE` alongside with other `JsonHeaders` populated from the reply AMQP message. This way a `JsonHeaders.RESOLVABLE_TYPE` from request message won't have effect * To get access to classes, supply `AbstractHeaderMapper` with a bean factory `ClassLoader` * Introduce a couple utility methods into `JsonHeaders` for building a `ResolvableType` **Cherry-pick to 5.2.x** * * Fix code formatting for arguments wrapping
2020-01-31 14:17:19 -05:00
parent 94c8cf7e21
commit 2f4394d8b5
7 changed files with 225 additions and 138 deletions
--- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
+++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AbstractAmqpOutboundEndpoint.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2020 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.
@@ -41,6 +41,7 @@ import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
+import org.springframework.integration.mapping.AbstractHeaderMapper;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
 import org.springframework.integration.support.DefaultErrorMessageStrategy;
 import org.springframework.integration.support.ErrorMessageStrategy;
@@ -416,6 +417,10 @@ public abstract class AbstractAmqpOutboundEndpoint extends AbstractReplyProducin
 		configureDelayGenerator(beanFactory);

 		endpointInit();
+
+		if (this.headerMapper instanceof AbstractHeaderMapper) {
+			((AbstractHeaderMapper<?>) this.headerMapper).setBeanClassLoader(getBeanClassLoader());
+		}
 	}

 	private void configureExchangeNameGenerator(BeanFactory beanFactory) {
--- 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
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -109,40 +109,41 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		Map<String, Object> headers = new HashMap<>();
 		try {
 			JavaUtils.INSTANCE
-				.acceptIfNotNull(AmqpHeaders.APP_ID, amqpMessageProperties.getAppId(), headers::put)
-				.acceptIfNotNull(AmqpHeaders.CLUSTER_ID, amqpMessageProperties.getClusterId(), headers::put)
-				.acceptIfNotNull(AmqpHeaders.CONTENT_ENCODING, amqpMessageProperties.getContentEncoding(),
-						headers::put);
+					.acceptIfNotNull(AmqpHeaders.APP_ID, amqpMessageProperties.getAppId(), headers::put)
+					.acceptIfNotNull(AmqpHeaders.CLUSTER_ID, amqpMessageProperties.getClusterId(), headers::put)
+					.acceptIfNotNull(AmqpHeaders.CONTENT_ENCODING, amqpMessageProperties.getContentEncoding(),
+							headers::put);
 			long contentLength = amqpMessageProperties.getContentLength();
 			JavaUtils.INSTANCE
-				.acceptIfCondition(contentLength > 0, AmqpHeaders.CONTENT_LENGTH, contentLength, headers::put)
-				.acceptIfHasText(AmqpHeaders.CONTENT_TYPE, amqpMessageProperties.getContentType(), headers::put)
-				.acceptIfHasText(AmqpHeaders.CORRELATION_ID, amqpMessageProperties.getCorrelationId(), headers::put)
-				.acceptIfNotNull(AmqpHeaders.RECEIVED_DELIVERY_MODE, amqpMessageProperties.getReceivedDeliveryMode(),
-						headers::put);
+					.acceptIfCondition(contentLength > 0, AmqpHeaders.CONTENT_LENGTH, contentLength, headers::put)
+					.acceptIfHasText(AmqpHeaders.CONTENT_TYPE, amqpMessageProperties.getContentType(), headers::put)
+					.acceptIfHasText(AmqpHeaders.CORRELATION_ID, amqpMessageProperties.getCorrelationId(), headers::put)
+					.acceptIfNotNull(AmqpHeaders.RECEIVED_DELIVERY_MODE,
+							amqpMessageProperties.getReceivedDeliveryMode(), headers::put);
 			long deliveryTag = amqpMessageProperties.getDeliveryTag();
 			JavaUtils.INSTANCE
-				.acceptIfCondition(deliveryTag > 0, AmqpHeaders.DELIVERY_TAG, deliveryTag, headers::put)
-				.acceptIfHasText(AmqpHeaders.EXPIRATION, amqpMessageProperties.getExpiration(), headers::put);
+					.acceptIfCondition(deliveryTag > 0, AmqpHeaders.DELIVERY_TAG, deliveryTag, headers::put)
+					.acceptIfHasText(AmqpHeaders.EXPIRATION, amqpMessageProperties.getExpiration(), headers::put);
 			Integer messageCount = amqpMessageProperties.getMessageCount();
 			JavaUtils.INSTANCE
-				.acceptIfCondition(messageCount != null && messageCount > 0, AmqpHeaders.MESSAGE_COUNT, messageCount,
-				headers::put)
-				.acceptIfHasText(AmqpHeaders.MESSAGE_ID, amqpMessageProperties.getMessageId(), headers::put);
+					.acceptIfCondition(messageCount != null && messageCount > 0, AmqpHeaders.MESSAGE_COUNT,
+							messageCount, headers::put)
+					.acceptIfHasText(AmqpHeaders.MESSAGE_ID, amqpMessageProperties.getMessageId(), headers::put);
 			Integer priority = amqpMessageProperties.getPriority();
 			JavaUtils.INSTANCE
-				.acceptIfCondition(priority != null && priority > 0, IntegrationMessageHeaderAccessor.PRIORITY,
-					priority, headers::put)
-				.acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelay(), headers::put)
-				.acceptIfNotNull(AmqpHeaders.RECEIVED_EXCHANGE, amqpMessageProperties.getReceivedExchange(),
-						headers::put)
-				.acceptIfHasText(AmqpHeaders.RECEIVED_ROUTING_KEY, amqpMessageProperties.getReceivedRoutingKey(),
-						headers::put)
-				.acceptIfNotNull(AmqpHeaders.REDELIVERED, amqpMessageProperties.isRedelivered(), headers::put)
-				.acceptIfNotNull(AmqpHeaders.REPLY_TO, amqpMessageProperties.getReplyTo(), headers::put)
-				.acceptIfNotNull(AmqpHeaders.TIMESTAMP, amqpMessageProperties.getTimestamp(), headers::put)
-				.acceptIfHasText(AmqpHeaders.TYPE, amqpMessageProperties.getType(), headers::put)
-				.acceptIfHasText(AmqpHeaders.RECEIVED_USER_ID, amqpMessageProperties.getReceivedUserId(), headers::put);
+					.acceptIfCondition(priority != null && priority > 0, IntegrationMessageHeaderAccessor.PRIORITY,
+							priority, headers::put)
+					.acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelay(), headers::put)
+					.acceptIfNotNull(AmqpHeaders.RECEIVED_EXCHANGE, amqpMessageProperties.getReceivedExchange(),
+							headers::put)
+					.acceptIfHasText(AmqpHeaders.RECEIVED_ROUTING_KEY, amqpMessageProperties.getReceivedRoutingKey(),
+							headers::put)
+					.acceptIfNotNull(AmqpHeaders.REDELIVERED, amqpMessageProperties.isRedelivered(), headers::put)
+					.acceptIfNotNull(AmqpHeaders.REPLY_TO, amqpMessageProperties.getReplyTo(), headers::put)
+					.acceptIfNotNull(AmqpHeaders.TIMESTAMP, amqpMessageProperties.getTimestamp(), headers::put)
+					.acceptIfHasText(AmqpHeaders.TYPE, amqpMessageProperties.getType(), headers::put)
+					.acceptIfHasText(AmqpHeaders.RECEIVED_USER_ID,
+							amqpMessageProperties.getReceivedUserId(), headers::put);

 			for (String jsonHeader : JsonHeaders.HEADERS) {
 				Object value = amqpMessageProperties.getHeaders().get(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""));
@@ -151,6 +152,7 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 				}
 			}

+			createJsonResolvableTypHeaderInAny(headers);
 		}
 		catch (Exception e) {
 			if (logger.isWarnEnabled()) {
@@ -160,6 +162,15 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 		return headers;
 	}

+	private void createJsonResolvableTypHeaderInAny(Map<String, Object> headers) {
+		Object typeIdHeader = headers.get(JsonHeaders.TYPE_ID);
+		if (typeIdHeader != null) {
+			headers.put(JsonHeaders.RESOLVABLE_TYPE,
+					JsonHeaders.buildResolvableType(getClassLoader(), typeIdHeader,
+							headers.get(JsonHeaders.CONTENT_TYPE_ID), headers.get(JsonHeaders.KEY_TYPE_ID)));
+		}
+	}
+
 	/**
 	 * Extract user-defined headers from an AMQP MessageProperties instance.
 	 */
@@ -186,28 +197,28 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 			MessageProperties amqpMessageProperties) {

 		JavaUtils.INSTANCE
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class),
-					amqpMessageProperties::setAppId)
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class),
-					amqpMessageProperties::setClusterId)
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class),
-					amqpMessageProperties::setContentEncoding)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class),
-					amqpMessageProperties::setContentLength)
-			.acceptIfHasText(this.extractContentTypeAsString(headers),
-					amqpMessageProperties::setContentType)
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CORRELATION_ID, String.class),
-					amqpMessageProperties::setCorrelationId)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELAY, Integer.class),
-					amqpMessageProperties::setDelay)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.class),
-					amqpMessageProperties::setDeliveryMode)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class),
-					amqpMessageProperties::setDeliveryTag)
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class),
-					amqpMessageProperties::setExpiration)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class),
-					amqpMessageProperties::setMessageCount);
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class),
+						amqpMessageProperties::setAppId)
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class),
+						amqpMessageProperties::setClusterId)
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class),
+						amqpMessageProperties::setContentEncoding)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class),
+						amqpMessageProperties::setContentLength)
+				.acceptIfHasText(this.extractContentTypeAsString(headers),
+						amqpMessageProperties::setContentType)
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.CORRELATION_ID, String.class),
+						amqpMessageProperties::setCorrelationId)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELAY, Integer.class),
+						amqpMessageProperties::setDelay)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.class),
+						amqpMessageProperties::setDeliveryMode)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class),
+						amqpMessageProperties::setDeliveryTag)
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class),
+						amqpMessageProperties::setExpiration)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class),
+						amqpMessageProperties::setMessageCount);
 		String messageId = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_ID, String.class);
 		if (StringUtils.hasText(messageId)) {
 			amqpMessageProperties.setMessageId(messageId);
@@ -219,16 +230,17 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 			}
 		}
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(getHeaderIfAvailable(headers, IntegrationMessageHeaderAccessor.PRIORITY, Integer.class),
-					amqpMessageProperties::setPriority)
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class),
-					amqpMessageProperties::setReceivedExchange)
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class),
-					amqpMessageProperties::setReceivedRoutingKey)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class),
-					amqpMessageProperties::setRedelivered)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, String.class),
-					amqpMessageProperties::setReplyTo);
+				.acceptIfNotNull(getHeaderIfAvailable(headers, IntegrationMessageHeaderAccessor.PRIORITY,
+						Integer.class),
+						amqpMessageProperties::setPriority)
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class),
+						amqpMessageProperties::setReceivedExchange)
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class),
+						amqpMessageProperties::setReceivedRoutingKey)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class),
+						amqpMessageProperties::setRedelivered)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, String.class),
+						amqpMessageProperties::setReplyTo);
 		Date timestamp = getHeaderIfAvailable(headers, AmqpHeaders.TIMESTAMP, Date.class);
 		if (timestamp != null) {
 			amqpMessageProperties.setTimestamp(timestamp);
@@ -240,18 +252,19 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
 			}
 		}
 		JavaUtils.INSTANCE
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class),
-					amqpMessageProperties::setType)
-			.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class),
-					amqpMessageProperties::setUserId);
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class),
+						amqpMessageProperties::setType)
+				.acceptIfNotNull(getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class),
+						amqpMessageProperties::setUserId);

 		mapJsonHeaders(headers, amqpMessageProperties);

 		JavaUtils.INSTANCE
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class),
-					replyCorrelation -> amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation))
-			.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_TO_STACK, String.class),
-					replyToStack -> amqpMessageProperties.setHeader("spring_reply_to", replyToStack));
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class),
+						replyCorrelation -> amqpMessageProperties
+								.setHeader("spring_reply_correlation", replyCorrelation))
+				.acceptIfHasText(getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_TO_STACK, String.class),
+						replyToStack -> amqpMessageProperties.setHeader("spring_reply_to", replyToStack));
 	}

 	private void mapJsonHeaders(Map<String, Object> headers, MessageProperties amqpMessageProperties) {
--- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java
+++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/OutboundEndpointTests.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2019 the original author or authors.
+ * Copyright 2002-2020 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.
@@ -27,9 +27,10 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;

+import java.util.Date;
 import java.util.concurrent.atomic.AtomicReference;

-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;

 import org.springframework.amqp.core.Message;
@@ -38,10 +39,13 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory;
 import org.springframework.amqp.rabbit.connection.CorrelationData;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
 import org.springframework.beans.factory.BeanFactory;
+import org.springframework.core.ResolvableType;
 import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
 import org.springframework.integration.channel.NullChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.mapping.support.JsonHeaders;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.support.GenericMessage;
@@ -50,6 +54,7 @@ import org.springframework.scheduling.TaskScheduler;
 /**
 * @author Gary Russell
 * @author Artem Bilan
+ *
 * @since 3.0
 */
 public class OutboundEndpointTests {
@@ -94,9 +99,9 @@ public class OutboundEndpointTests {
 				new SimpleMessageListenerContainer(connectionFactory), "replyTo"));
 		amqpTemplate.setTaskScheduler(mock(TaskScheduler.class));
 		AsyncAmqpOutboundGateway gateway = new AsyncAmqpOutboundGateway(amqpTemplate);
-		willAnswer(
-				invocation -> amqpTemplate.new RabbitMessageFuture("foo", invocation.getArgument(2)))
-					.given(amqpTemplate).sendAndReceive(anyString(), anyString(), any(Message.class));
+		willAnswer(invocation -> amqpTemplate.new RabbitMessageFuture("foo", invocation.getArgument(2)))
+				.given(amqpTemplate)
+				.sendAndReceive(anyString(), anyString(), any(Message.class));
 		gateway.setExchangeName("foo");
 		gateway.setRoutingKey("bar");
 		gateway.setDelayExpressionString("42");
@@ -158,6 +163,33 @@ public class OutboundEndpointTests {
 		assertThat(amqpMessage.get().getMessageProperties().getHeaders().get(MessageHeaders.REPLY_CHANNEL)).isNull();
 	}

+	@Test
+	public void testReplyHeadersWin() {
+		ConnectionFactory connectionFactory = mock(ConnectionFactory.class);
+		TestRabbitTemplate amqpTemplate = spy(new TestRabbitTemplate(connectionFactory));
+		amqpTemplate.setUseTemporaryReplyQueues(true);
+		AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(amqpTemplate);
+		endpoint.setExpectReply(true);
+		willAnswer(invocation ->
+				org.springframework.amqp.core.MessageBuilder.withBody(new byte[0])
+						.setHeader(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, String.class.getName())
+						.build()
+		).given(amqpTemplate)
+				.doSendAndReceiveWithTemporary(isNull(), isNull(), any(Message.class), isNull());
+		QueueChannel replyChannel = new QueueChannel();
+		org.springframework.messaging.Message<?> message = MessageBuilder.withPayload("foo")
+				.setHeader(JsonHeaders.RESOLVABLE_TYPE, ResolvableType.forClass(Date.class))
+				.setReplyChannel(replyChannel)
+				.build();
+		endpoint.handleMessage(message);
+		org.springframework.messaging.Message<?> receive = replyChannel.receive(10_000);
+		assertThat(receive).isNotNull();
+		assertThat(receive.getHeaders())
+				.containsEntry(AbstractJavaTypeMapper.DEFAULT_CLASSID_FIELD_NAME, String.class.getName())
+				.containsEntry(JsonHeaders.TYPE_ID, String.class.getName())
+				.containsEntry(JsonHeaders.RESOLVABLE_TYPE, ResolvableType.forClass(String.class));
+	}
+
 	/**
 	 * Increase method visibility
 	 */