From cf221da0dac79efb6d4927ec3a3260aa4f7a79fb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 6 Jul 2020 16:17:38 -0400 Subject: [PATCH] GH-1219: Fix header mapping for replies (@SendTo) Resolves https://github.com/spring-projects/spring-amqp/issues/1219 The headers were mapped after message conversion. This prevented using a `ContentTypeDelegatingMessageConverter` because the content type was not set. Add a header to control whether the user or converter gets to set the content type property in the final message. **cherry-pick to 2.2.x, 2.1.x, 1.7.x** --- .../amqp/support/AmqpHeaders.java | 4 +- .../converter/MessagingMessageConverter.java | 14 +- ...atingMessageConverterIntegrationTests.java | 185 ++++++++++++++++++ src/reference/asciidoc/amqp.adoc | 24 ++- 4 files changed, 222 insertions(+), 5 deletions(-) create mode 100644 spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ContentTypeDelegatingMessageConverterIntegrationTests.java diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java b/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java index bbcbe111..3bc8e5d0 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/AmqpHeaders.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. @@ -49,6 +49,8 @@ public abstract class AmqpHeaders { public static final String CONTENT_TYPE = MessageHeaders.CONTENT_TYPE; + public static final String CONTENT_TYPE_CONVERTER_WINS = PREFIX + "contentTypeConverterWins"; + public static final String CORRELATION_ID = PREFIX + "correlationId"; public static final String DELAY = PREFIX + "delay"; diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessagingMessageConverter.java b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessagingMessageConverter.java index 20ccb4cb..fe03b837 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessagingMessageConverter.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/support/converter/MessagingMessageConverter.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-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. @@ -20,9 +20,11 @@ import java.util.Map; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.support.AmqpHeaderMapper; +import org.springframework.amqp.support.AmqpHeaders; import org.springframework.amqp.support.SimpleAmqpHeaderMapper; import org.springframework.beans.factory.InitializingBean; import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.support.MessageBuilder; import org.springframework.util.Assert; @@ -105,10 +107,16 @@ public class MessagingMessageConverter implements MessageConverter, Initializing Message.class.getName() + "] is handled by this converter"); } Message input = (Message) object; + this.headerMapper.fromHeaders(input.getHeaders(), messageProperties); org.springframework.amqp.core.Message amqpMessage = this.payloadConverter.toMessage( input.getPayload(), messageProperties); - - this.headerMapper.fromHeaders(input.getHeaders(), messageProperties); + // Default previous behavior of mapper wins for backwards compatibility. + if (!Boolean.TRUE.equals(input.getHeaders().get(AmqpHeaders.CONTENT_TYPE_CONVERTER_WINS))) { + String contentType = input.getHeaders().get(MessageHeaders.CONTENT_TYPE, String.class); + if (contentType != null) { + messageProperties.setContentType(contentType); + } + } return amqpMessage; } diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ContentTypeDelegatingMessageConverterIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ContentTypeDelegatingMessageConverterIntegrationTests.java new file mode 100644 index 00000000..a1a25fbb --- /dev/null +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/annotation/ContentTypeDelegatingMessageConverterIntegrationTests.java @@ -0,0 +1,185 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.amqp.rabbit.annotation; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.AnonymousQueue; +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.junit.RabbitAvailable; +import org.springframework.amqp.rabbit.junit.RabbitAvailableCondition; +import org.springframework.amqp.support.AmqpHeaders; +import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter; +import org.springframework.amqp.support.converter.MessageConversionException; +import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.amqp.support.converter.SimpleMessageConverter; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Gary Russell + * @since 2.5 + * + */ +@RabbitAvailable +@SpringJUnitConfig +public class ContentTypeDelegatingMessageConverterIntegrationTests { + + @Autowired + private Config config; + + @Autowired + private RabbitTemplate template; + + @Autowired + private AnonymousQueue queue1; + + @Test + void testReplyContentType() throws InterruptedException { + this.template.convertAndSend(this.queue1.getName(), "foo", msg -> { + msg.getMessageProperties().setContentType("foo/bar"); + return msg; + }); + assertThat(this.config.latch1.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.replyContentType).isEqualTo("baz/qux"); + assertThat(this.config.receivedReplyContentType).isEqualTo("baz/qux"); + + this.template.convertAndSend(this.queue1.getName(), "bar", msg -> { + msg.getMessageProperties().setContentType("foo/bar"); + return msg; + }); + assertThat(this.config.latch2.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.replyContentType).isEqualTo("baz/qux"); + assertThat(this.config.receivedReplyContentType).isEqualTo("foo/bar"); + } + + @Configuration + @EnableRabbit + public static class Config { + + final CountDownLatch latch1 = new CountDownLatch(1); + + final CountDownLatch latch2 = new CountDownLatch(2); + + volatile String replyContentType; + + volatile String receivedReplyContentType; + + @Bean + public ConnectionFactory cf() { + return new CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().getConnectionFactory()); + } + + @Bean + public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() { + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + factory.setConnectionFactory(cf()); + ContentTypeDelegatingMessageConverter converter = + new ContentTypeDelegatingMessageConverter(new SimpleMessageConverter()); + converter.addDelegate("foo/bar", new MessageConverter() { + + @Override + public Message toMessage(Object object, MessageProperties messageProperties) + throws MessageConversionException { + + return new Message("foo".getBytes(), messageProperties); + } + + @Override + public Object fromMessage(Message message) throws MessageConversionException { + return new String(message.getBody()); + } + + }); + converter.addDelegate("baz/qux", new MessageConverter() { + + @Override + public Message toMessage(Object object, MessageProperties messageProperties) + throws MessageConversionException { + + Config.this.replyContentType = messageProperties.getContentType(); + messageProperties.setContentType("foo/bar"); + return new Message("foo".getBytes(), messageProperties); + } + + @Override + public Object fromMessage(Message message) throws MessageConversionException { + return new String(message.getBody()); + } + + }); + factory.setMessageConverter(converter); + return factory; + } + + @Bean + public RabbitTemplate template() { + return new RabbitTemplate(cf()); + } + + @Bean + public RabbitAdmin admin() { + return new RabbitAdmin(cf()); + } + + @Bean + public AnonymousQueue queue1() { + return new AnonymousQueue(); + } + + @Bean + public AnonymousQueue queue2() { + return new AnonymousQueue(); + } + + @RabbitListener(queues = "#{@queue1.name}") + @SendTo("#{@queue2.name}") + public org.springframework.messaging.Message listen1(String in) { + MessageBuilder builder = MessageBuilder.withPayload(in) + .setHeader(MessageHeaders.CONTENT_TYPE, "baz/qux"); + if ("bar".equals(in)) { + builder.setHeader(AmqpHeaders.CONTENT_TYPE_CONVERTER_WINS, true); + } + return builder.build(); + } + + @RabbitListener(queues = "#{@queue2.name}") + public void listen2(Message in) { + this.receivedReplyContentType = in.getMessageProperties().getContentType(); + this.latch1.countDown(); + this.latch2.countDown(); + } + + } + +} diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 42e2d56a..fab732df 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -2717,7 +2717,7 @@ public ReplyPostProcessor echoCustomHeader() { ---- ==== -The `@SendTo` value is assumed as a reply `exchange` and `routingKey` pair that follws the `exchange/routingKey` pattern, +The `@SendTo` value is assumed as a reply `exchange` and `routingKey` pair that follows the `exchange/routingKey` pattern, where one of those parts can be omitted. The valid values are as follows: @@ -2800,6 +2800,28 @@ public String listen(Message in) { ---- ==== +====== Reply ContentType + +If you are using a sophisticated message converter, such as the `ContentTypeDelegatingMessageConverter`, you can control the content type of the reply by returning a `spring-messaging` `Message`: + +==== +[source, java] +---- +@RabbitListener(queues = "q1") +@SendTo("q2") +public Message listen(String in) { + ... + return MessageBuilder.withPayload(in.toUpperCase()) + .setHeader(MessageHeaders.CONTENT_TYPE, "application/xml") + .build(); +} +---- +==== + +This content type will be passed in the `MessageProperties` to the converter. +By default, for backwards compatibility, any content type property set by the converter will be overwritten by this value after conversion. +If you wish to override that behavior, also set the `AmqpHeaders.CONTENT_TYPE_CONVERTER_WINS` to `true` and any value set by the converter will be retained. + [[annotation-method-selection]] ====== Multi-method Listeners