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

AMQP-827: Fix @RL reply Message<?> conversion

Browse Source 
JIRA: https://jira.spring.io/browse/AMQP-827

Use the correct `payloadConverter` in the `MessagingMessageConverter`
to support `@RabbitListener` `Message<?>` return types.

**cherry-pick to 2.0.x, 1.7.x**

# Conflicts:
#	spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
This commit is contained in:
Gary Russell
2018-08-07 16:04:27 -04:00
committed by Artem Bilan
parent 97552bc87c
commit 0e597c98de
2 changed files with 50 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java
View File

@@ -97,6 +97,12 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis
return this.messagingMessageConverter;
}
@Override
public void setMessageConverter(MessageConverter messageConverter) {
super.setMessageConverter(messageConverter);
this.messagingMessageConverter.setPayloadConverter(messageConverter);
}
@Override
public void onMessage(org.springframework.amqp.core.Message amqpMessage, Channel channel) throws Exception {
Message<?> message = toMessagingMessage(amqpMessage);

45
spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java
View File

@@ -18,7 +18,10 @@ package org.springframework.amqp.rabbit.annotation;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -36,6 +39,7 @@ import java.lang.annotation.Target;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@@ -65,6 +69,7 @@ import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -111,6 +116,7 @@ import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.test.annotation.DirtiesContext;
@@ -153,7 +159,8 @@ public class EnableRabbitIntegrationTests {
"test.converted", "test.converted.list", "test.converted.array", "test.converted.args1",
"test.converted.args2", "test.converted.message", "test.notconverted.message",
"test.notconverted.channel", "test.notconverted.messagechannel", "test.notconverted.messagingmessage",
"test.converted.foomessage", "test.notconverted.messagingmessagenotgeneric", "amqp656dlq");
"test.converted.foomessage", "test.notconverted.messagingmessagenotgeneric", "amqp656dlq",
"test.messaging.message", "test.amqp.message");
@Autowired
private RabbitTemplate rabbitTemplate;
@@ -577,6 +584,26 @@ public class EnableRabbitIntegrationTests {
assertSame(value, typeCache.get(Foo1.class));
}
@Test
public void messagingMessageReturned() {
Message message = org.springframework.amqp.core.MessageBuilder.withBody("\"messaging\"".getBytes())
.andProperties(MessagePropertiesBuilder.newInstance().setContentType("application/json").build()).build();
message = this.rabbitTemplate.sendAndReceive("test.messaging.message", message);
assertThat(message, is(notNullValue()));
assertThat(new String(message.getBody()), equalTo("{\"field\":\"MESSAGING\"}"));
assertThat(message.getMessageProperties().getHeaders().get("foo"), equalTo("bar"));
}
@Test
public void amqpMessageReturned() {
Message message = org.springframework.amqp.core.MessageBuilder.withBody("amqp".getBytes())
.andProperties(MessagePropertiesBuilder.newInstance().setContentType("text/plain").build()).build();
message = this.rabbitTemplate.sendAndReceive("test.amqp.message", message);
assertThat(message, is(notNullValue()));
assertThat(new String(message.getBody()), equalTo("AMQP"));
assertThat(message.getMessageProperties().getHeaders().get("foo"), equalTo("bar"));
}
interface TxService {
@Transactional
@@ -811,6 +838,22 @@ public class EnableRabbitIntegrationTests {
public String handleWithDeadLetterDefaultExchange(String foo) {
throw new AmqpRejectAndDontRequeueException("dlq");
}
@RabbitListener(queues = "test.messaging.message", containerFactory = "simpleJsonListenerContainerFactory")
public org.springframework.messaging.Message<Bar> messagingMessage(String in) {
Bar bar = new Bar();
bar.field = in.toUpperCase();
return new GenericMessage<>(bar, Collections.singletonMap("foo", "bar"));
}
@RabbitListener(queues = "test.amqp.message")
public Message amqpMessage(String in) {
return org.springframework.amqp.core.MessageBuilder.withBody(in.toUpperCase().getBytes())
.andProperties(MessagePropertiesBuilder.newInstance().setContentType("text/plain")
.setHeader("foo", "bar")
.build())
.build();
}
}
public static class Foo1 {