diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index 9fe9a76c..9c2155d5 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -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); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java index ae705657..48d28d21 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/EnableRabbitIntegrationTests.java @@ -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 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 {