Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3285: Fix AMQP JsonHeaders Regression

Browse Source 
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**
This commit is contained in:
Artem Bilan
2014-02-06 19:28:52 +02:00
committed by Gary Russell
parent 15d9f9566d
commit d54772f171
2 changed files with 98 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<MessagePropert
amqpMessageProperties.setUserId(userId);
}
Map<String, String> jsonHeaders = new HashMap<String, String>();
for (String jsonHeader : JsonHeaders.HEADERS) {
Object value = getHeaderIfAvailable(headers, jsonHeader, Object.class);
if (value != null) {
@@ -298,10 +300,18 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessagePropert
if (value instanceof Class<?>) {
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);

86
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/inbound/InboundEndpointTests.java
View File

@@ -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<Channel>() {
@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<Channel>() {
@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<Channel>() {
@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<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
org.springframework.amqp.core.Message message = (org.springframework.amqp.core.Message) invocation.getArguments()[2];
Map<String,Object> 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 {