From d54772f1711691cb48b53ea905cebde96a17f8d4 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 6 Feb 2014 19:28:52 +0200 Subject: [PATCH] INT-3285: Fix AMQP `JsonHeaders` Regression JIRA: https://jira.springsource.org/browse/INT-3285 Spring Integration 3.0 introduced the `JsonHeaders` and populates them from a request AMQP Message to the SI Message. In the case of the `AmqpInboundGateway` and `JsonMessageConverter`, if the SI flow doesn't do anything with `JsonHeaders` they are returned to the `AmqpInboundGateway` for reply Message and may override correct values provided by `JsonMessageConverter` in the `DefaultAmqpHeaderMapper`. This fix provides the check for headers from the reply Message's `MessageProperties` and populates `JsonHeaders` if the `MessageProperties` doesn't contains the `__TypeId__` header already. Note, it is just a fix for regression to restore previous behavior. There is maybe a reason to revise (Another JIRA) all standard AMQP headers and populate them from SI `MessageHeaders` in the `DefaultAmqpHeaderMapper`, only if they haven't been populated by `MessageConverter` before. **Cherry-pick to 3.0.x** --- .../amqp/support/DefaultAmqpHeaderMapper.java | 14 ++- .../amqp/inbound/InboundEndpointTests.java | 86 +++++++++++++++++++ 2 files changed, 98 insertions(+), 2 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 6f76454cc3..f8119f59dd 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -291,6 +291,8 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper jsonHeaders = new HashMap(); + for (String jsonHeader : JsonHeaders.HEADERS) { Object value = getHeaderIfAvailable(headers, jsonHeader, Object.class); if (value != null) { @@ -298,10 +300,18 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper) { value = ((Class) value).getName(); } - amqpMessageProperties.setHeader(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""), value.toString()); + jsonHeaders.put(jsonHeader.replaceFirst(JsonHeaders.PREFIX, ""), value.toString()); } } + /* + * If the MessageProperties already contains JsonHeaders, don't overwrite them here because they were + * set up by a message converter. + */ + if (!amqpMessageProperties.getHeaders().containsKey(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))) { + amqpMessageProperties.getHeaders().putAll(jsonHeaders); + } + String replyCorrelation = getHeaderIfAvailable(headers, AmqpHeaders.SPRING_REPLY_CORRELATION, String.class); if (StringUtils.hasLength(replyCorrelation)) { amqpMessageProperties.setHeader("spring_reply_correlation", replyCorrelation); diff --git a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java index 6a05ef9a53..b4b23f0e68 100644 --- a/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java +++ b/spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java @@ -17,12 +17,18 @@ package org.springframework.integration.amqp.inbound; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.anyBoolean; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.util.Map; + import org.junit.Test; +import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; @@ -30,13 +36,21 @@ import org.springframework.amqp.core.MessageListener; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.connection.Connection; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.amqp.rabbit.support.CorrelationData; import org.springframework.amqp.support.converter.JsonMessageConverter; import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.json.JsonToObjectTransformer; import org.springframework.integration.json.ObjectToJsonTransformer; +import org.springframework.integration.mapping.support.JsonHeaders; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -54,6 +68,7 @@ public class InboundEndpointTests { public void testInt2809JavaTypePropertiesToAmqp() { Connection connection = mock(Connection.class); doAnswer(new Answer() { + @Override public Channel answer(InvocationOnMock invocation) throws Throwable { return mock(Channel.class); } @@ -92,6 +107,7 @@ public class InboundEndpointTests { public void testInt2809JavaTypePropertiesFromAmqp() { Connection connection = mock(Connection.class); doAnswer(new Answer() { + @Override public Channel answer(InvocationOnMock invocation) throws Throwable { return mock(Channel.class); } @@ -123,6 +139,76 @@ public class InboundEndpointTests { assertEquals(payload, result.getPayload()); } + @Test + public void testMessageConverterJsonHeadersHavePrecedenceOverMessageHeaders() { + Connection connection = mock(Connection.class); + doAnswer(new Answer() { + @Override + public Channel answer(InvocationOnMock invocation) throws Throwable { + return mock(Channel.class); + } + }).when(connection).createChannel(anyBoolean()); + ConnectionFactory connectionFactory = mock(ConnectionFactory.class); + when(connectionFactory.createConnection()).thenReturn(connection); + SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); + container.setConnectionFactory(connectionFactory); + + DirectChannel channel = new DirectChannel(); + + channel.subscribe(new MessageTransformingHandler(new Transformer() { + + @Override + public Message transform(Message message) { + return MessageBuilder.fromMessage(message) + .setHeader(JsonHeaders.TYPE_ID, "foo") + .setHeader(JsonHeaders.CONTENT_TYPE_ID, "bar") + .setHeader(JsonHeaders.KEY_TYPE_ID, "baz") + .build(); + } + })); + + AmqpInboundGateway gateway = new AmqpInboundGateway(container); + gateway.setMessageConverter(new JsonMessageConverter()); + + gateway.setRequestChannel(channel); + gateway.afterPropertiesSet(); + + RabbitTemplate rabbitTemplate = Mockito.spy(TestUtils.getPropertyValue(gateway, "amqpTemplate", RabbitTemplate.class)); + + Mockito.doAnswer(new Answer() { + + @Override + public Object answer(InvocationOnMock invocation) throws Throwable { + org.springframework.amqp.core.Message message = (org.springframework.amqp.core.Message) invocation.getArguments()[2]; + Map headers = message.getMessageProperties().getHeaders(); + assertTrue(headers.containsKey(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertNotEquals("foo", headers.get(JsonHeaders.TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertFalse(headers.containsKey(JsonHeaders.CONTENT_TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertFalse(headers.containsKey(JsonHeaders.KEY_TYPE_ID.replaceFirst(JsonHeaders.PREFIX, ""))); + assertFalse(headers.containsKey(JsonHeaders.TYPE_ID)); + assertFalse(headers.containsKey(JsonHeaders.KEY_TYPE_ID)); + assertFalse(headers.containsKey(JsonHeaders.CONTENT_TYPE_ID)); + return null; + } + } + + ).when(rabbitTemplate).send(Mockito.anyString(), Mockito.anyString(), + Mockito.any(org.springframework.amqp.core.Message.class), Mockito.any(CorrelationData.class)); + + DirectFieldAccessor directFieldAccessor = new DirectFieldAccessor(gateway); + directFieldAccessor.setPropertyValue("amqpTemplate", rabbitTemplate); + + Object payload = new Foo("bar1"); + + MessageProperties amqpMessageProperties = new MessageProperties(); + amqpMessageProperties.setReplyTo("test"); + org.springframework.amqp.core.Message amqpMessage = new JsonMessageConverter().toMessage(payload, amqpMessageProperties); + + MessageListener listener = (MessageListener) container.getMessageListener(); + listener.onMessage(amqpMessage); + + } + public static class Foo {