GH-152: RabbitMQ consumer: Add headersMappedLast

Fixes https://github.com/spring-cloud/stream-applications/issues/152

* Add the possibilities to configure the `headersMappedLast` for the RabbitMQ consumer

* Changes following the code review

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: `$$<none>$$`)*
 $$exchange$$:: $$Exchange name - overridden by exchangeNameExpression, if supplied.$$ *($$String$$, default: `$$<empty string>$$`)*
 $$exchange-expression$$:: $$A SpEL expression that evaluates to an exchange name.$$ *($$Expression$$, default: `$$<none>$$`)*
+$$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$$`)*

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");
+		}
+	}
+}

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) {

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;
+	}
 }