diff --git a/applications/sink/rabbit-sink/README.adoc b/applications/sink/rabbit-sink/README.adoc index b90d5547..a3f9c887 100644 --- a/applications/sink/rabbit-sink/README.adoc +++ b/applications/sink/rabbit-sink/README.adoc @@ -18,6 +18,7 @@ Properties grouped by prefix: $$converter-bean-name$$:: $$The bean name for a custom message converter; if omitted, a SimpleMessageConverter is used. If 'jsonConverter', a Jackson2JsonMessageConverter bean will be created for you.$$ *($$String$$, default: `$$$$`)* $$exchange$$:: $$Exchange name - overridden by exchangeNameExpression, if supplied.$$ *($$String$$, default: `$$$$`)* $$exchange-expression$$:: $$A SpEL expression that evaluates to an exchange name.$$ *($$Expression$$, default: `$$$$`)* +$$headers-mapped-last$$:: $$When mapping headers for the outbound message, determine whether the headers are mapped before the message is converted, or afterwards.$$ *($$Boolean$$, default: `$$true$$`)* $$mapped-request-headers$$:: $$Headers that will be mapped.$$ *($$String[]$$, default: `$$[*]$$`)* $$own-connection$$:: $$When true, use a separate connection based on the boot properties.$$ *($$Boolean$$, default: `$$false$$`)* $$persistent-delivery-mode$$:: $$Default delivery mode when 'amqp_deliveryMode' header is not present, true for PERSISTENT.$$ *($$Boolean$$, default: `$$false$$`)* diff --git a/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/HeadersMappedLastTests.java b/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/HeadersMappedLastTests.java new file mode 100644 index 00000000..9b3b51cf --- /dev/null +++ b/applications/sink/rabbit-sink/src/test/java/org/springframework/cloud/stream/app/sink/rabbit/HeadersMappedLastTests.java @@ -0,0 +1,64 @@ +/* + * Copyright 2016-2021 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.cloud.stream.app.sink.rabbit; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.Message; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.test.context.TestPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests for RabbitSource with headersMappedLast. + * + * @author Nicolas Labrot + */ +public class HeadersMappedLastTests { + + @TestPropertySource(properties = "rabbit.routingKey=scsapp-testq") + public static class HeadersMappedLastTrueTests extends RabbitSinkIntegrationTests { + + @Test + public void test() { + this.channels.send(MessageBuilder.withPayload("{}".getBytes()) + .setHeader("content-type", "application/json") + .build()); + this.rabbitTemplate.setReceiveTimeout(10000); + Message received = this.rabbitTemplate.receive("scsapp-testq"); + assertThat(new String(received.getBody())).isEqualTo("{}"); + assertThat(received.getMessageProperties().getContentType()).isEqualTo("application/json"); + } + } + + @TestPropertySource(properties = { "rabbit.routingKey=scsapp-testq", + "rabbit.headersMappedLast=false" }) + public static class HeadersMappedLastFalseTests extends RabbitSinkIntegrationTests { + + @Test + public void test() { + this.channels.send(MessageBuilder.withPayload("{}".getBytes()) + .setHeader("content-type", "application/json") + .build()); + this.rabbitTemplate.setReceiveTimeout(10000); + Message received = this.rabbitTemplate.receive("scsapp-testq"); + assertThat(new String(received.getBody())).isEqualTo("{}"); + assertThat(received.getMessageProperties().getContentType()).isEqualTo("application/octet-stream"); + } + } +} diff --git a/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java b/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java index dd89486e..ea8a872f 100644 --- a/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java +++ b/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java @@ -78,7 +78,8 @@ public class RabbitConsumerConfiguration implements DisposableBean { .mappedRequestHeaders(properties.getMappedRequestHeaders()) .defaultDeliveryMode(properties.getPersistentDeliveryMode() ? MessageDeliveryMode.PERSISTENT - : MessageDeliveryMode.NON_PERSISTENT); + : MessageDeliveryMode.NON_PERSISTENT) + .headersMappedLast(this.properties.isHeadersMappedLast()); Expression exchangeExpression = this.properties.getExchangeExpression(); if (exchangeExpression != null) { diff --git a/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerProperties.java b/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerProperties.java index df1a15eb..37d7b176 100644 --- a/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerProperties.java +++ b/functions/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerProperties.java @@ -73,6 +73,12 @@ public class RabbitConsumerProperties { */ private boolean ownConnection; + /** + * When mapping headers for the outbound message, determine whether the headers are + * mapped before the message is converted, or afterwards. + */ + private boolean headersMappedLast = true; + public String getExchange() { return this.exchange; } @@ -142,4 +148,11 @@ public class RabbitConsumerProperties { this.ownConnection = ownConnection; } + public boolean isHeadersMappedLast() { + return this.headersMappedLast; + } + + public void setHeadersMappedLast(boolean headersMappedLast) { + this.headersMappedLast = headersMappedLast; + } }