Fix Compressed contentEncoding Delimiter

Resolves #1251

Delimiter should be a comma and whitespace trimmeed.

Handle both delimiters in decompressors and add property for backwards compatibility.

**I will backport to 2.2.x with default `:`**

* Do not add a delimiter if the original encoding is an empty String.

Author: garyrussell

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.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.
@@ -31,7 +31,9 @@
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.core.MessagePropertiesBuilder;
 import org.springframework.core.Ordered;
+import org.springframework.util.Assert;
 import org.springframework.util.FileCopyUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Base class for post processors that compress the message body. The content encoding is
@@ -53,6 +55,8 @@ public abstract class AbstractCompressingPostProcessor implements MessagePostPro
 
 	private boolean copyProperties = false;
 
+	private String encodingDelimiter = ":";
+
 	/**
 	 * Construct a post processor that will include the
 	 * {@link MessageProperties#SPRING_AUTO_DECOMPRESS} header set to 'true'.
@@ -85,6 +89,18 @@ public void setCopyProperties(boolean copyProperties) {
 		this.copyProperties = copyProperties;
 	}
 
+	/**
+	 * Set a delimiter to be added between the compression type and the original encoding,
+	 * if any. Defaults to {@code ":"} for compatibility with consumers using versions of
+	 * spring-amqp earlier than 2.2.12. In version 2.3, it will default to {@code ", "}.
+	 * @param encodingDelimiter the delimiter.
+	 * @since 2.2.12
+	 */
+	public void setEncodingDelimiter(String encodingDelimiter) {
+		Assert.notNull(encodingDelimiter, "'encodingDelimiter' cannot be null");
+		this.encodingDelimiter = encodingDelimiter;
+	}
+
 	@Override
 	public Message postProcessMessage(Message message) throws AmqpException {
 		try {
@@ -109,9 +125,9 @@ public Message postProcessMessage(Message message) throws AmqpException {
 
 			MessageProperties messageProperties =
 					messagePropertiesBuilder.setContentEncoding(getEncoding() +
-							(originalProperties.getContentEncoding() == null
+							(!StringUtils.hasText(originalProperties.getContentEncoding())
 									? ""
-									: ":" + originalProperties.getContentEncoding()))
+									: this.encodingDelimiter + originalProperties.getContentEncoding()))
 							.build();
 
 			return new Message(compressed, messageProperties);

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractDecompressingPostProcessor.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.
@@ -91,17 +91,22 @@ public Message postProcessMessage(Message message) throws AmqpException {
 				FileCopyUtils.copy(unzipper, out);
 				MessageProperties messageProperties = message.getMessageProperties();
 				String encoding = messageProperties.getContentEncoding();
-				int colonAt = encoding.indexOf(':');
-				if (colonAt > 0) {
-					encoding = encoding.substring(0, colonAt);
+				int delimAt = encoding.indexOf(':');
+				if (delimAt < 0) {
+					delimAt = encoding.indexOf(',');
+				}
+				if (delimAt > 0) {
+					encoding = encoding.substring(0, delimAt);
 				}
 				Assert.state(getEncoding().equals(encoding), "Content encoding must be:" + getEncoding() + ", was:"
 						+ encoding);
-				if (colonAt < 0) {
+				if (delimAt < 0) {
 					messageProperties.setContentEncoding(null);
 				}
 				else {
-					messageProperties.setContentEncoding(messageProperties.getContentEncoding().substring(colonAt + 1));
+					messageProperties.setContentEncoding(messageProperties.getContentEncoding()
+							.substring(delimAt + 1)
+							.trim());
 				}
 				messageProperties.getHeaders().remove(MessageProperties.SPRING_AUTO_DECOMPRESS);
 				return new Message(out.toByteArray(), messageProperties);

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.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.
@@ -94,9 +94,12 @@ public Message postProcessMessage(Message message) throws AmqpException {
 			return message;
 		}
 		else {
-			int colonAt = encoding.indexOf(':');
-			if (colonAt > 0) {
-				encoding = encoding.substring(0, colonAt);
+			int delimAt = encoding.indexOf(':');
+			if (delimAt < 0) {
+				delimAt = encoding.indexOf(',');
+			}
+			if (delimAt > 0) {
+				encoding = encoding.substring(0, delimAt);
 			}
 			MessagePostProcessor decompressor = this.decompressors.get(encoding);
 			if (decompressor != null) {

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java

@@ -494,10 +494,14 @@ public void testSimpleBatchGZippedWithEncodingInflated() throws Exception {
 		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
 		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
 		template.setConnectionFactory(this.connectionFactory);
-		template.setBeforePublishPostProcessors(new GZipPostProcessor());
+		AtomicReference<String> encoding = new AtomicReference<>();
+		template.setBeforePublishPostProcessors(new GZipPostProcessor(), msg -> {
+			encoding.set(msg.getMessageProperties().getContentEncoding());
+			return msg;
+		});
 		template.setAfterReceivePostProcessors(new DelegatingDecompressingPostProcessor());
 		MessageProperties props = new MessageProperties();
-		props.setContentEncoding("foo");
+		props.setContentEncoding("");
 		Message message = new Message("foo".getBytes(), props);
 		template.send("", ROUTE, message);
 		message = new Message("bar".getBytes(), props);
@@ -506,6 +510,7 @@ public void testSimpleBatchGZippedWithEncodingInflated() throws Exception {
 		byte[] out = (byte[]) template.receiveAndConvert(ROUTE);
 		assertThat(out).isNotNull();
 		assertThat(new String(out)).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
+		assertThat(encoding.get()).isEqualTo("gzip");
 	}
 
 	@Test
@@ -535,6 +540,7 @@ public void testSimpleBatchZippedWithEncoding() throws Exception {
 		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
 		template.setConnectionFactory(this.connectionFactory);
 		ZipPostProcessor zipPostProcessor = new ZipPostProcessor();
+		zipPostProcessor.setEncodingDelimiter(", ");
 		assertThat(getStreamLevel(zipPostProcessor)).isEqualTo(Deflater.BEST_SPEED);
 		template.setBeforePublishPostProcessors(zipPostProcessor);
 		MessageProperties props = new MessageProperties();
@@ -544,7 +550,7 @@ public void testSimpleBatchZippedWithEncoding() throws Exception {
 		message = new Message("bar".getBytes(), props);
 		template.send("", ROUTE, message);
 		message = receive(template);
-		assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("zip:foo");
+		assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("zip, foo");
 		UnzipPostProcessor unzipper = new UnzipPostProcessor();
 		message = unzipper.postProcessMessage(message);
 		assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
@@ -597,6 +603,7 @@ public void testSimpleBatchDeflaterWithEncoding() throws Exception {
 		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
 		template.setConnectionFactory(this.connectionFactory);
 		DeflaterPostProcessor deflaterPostProcessor = new DeflaterPostProcessor();
+		deflaterPostProcessor.setEncodingDelimiter(", ");
 		assertThat(getStreamLevel(deflaterPostProcessor)).isEqualTo(Deflater.BEST_SPEED);
 		template.setBeforePublishPostProcessors(deflaterPostProcessor);
 		MessageProperties props = new MessageProperties();
@@ -606,7 +613,7 @@ public void testSimpleBatchDeflaterWithEncoding() throws Exception {
 		message = new Message("bar".getBytes(), props);
 		template.send("", ROUTE, message);
 		message = receive(template);
-		assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("deflate:foo");
+		assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("deflate, foo");
 		InflaterPostProcessor inflater = new InflaterPostProcessor();
 		message = inflater.postProcessMessage(message);
 		assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");

src/reference/asciidoc/amqp.adoc

@@ -3828,6 +3828,10 @@ By default, these properties are reused for performance reasons, and modified wi
 If you retain a reference to the original outbound message, its properties will change as well.
 So, if your application retains a copy of an outbound message with these message post processors, consider turning the `copyProperties` option on.
 
+IMPORTANT: Starting with version 2.2.12, you can configure the delimiter that the compressing post processors use between content encoding elements.
+With versions 2.2.11 and before, this was hard-coded as `:`; it is now set to `:` by default but can be overridden to a standard `,` by setting the `encodingDelimiter` property; this will be the default in 2.3 and above.
+The decompressors will work with both delimiters.
+
 Similarly, the `SimpleMessageListenerContainer` also has a `setAfterReceivePostProcessors()` method, letting the decompression be performed after messages are received by the container.
 
 Starting with version 2.1.4, `addBeforePublishPostProcessors()` and `addAfterReceivePostProcessors()` have been added to the `RabbitTemplate` to allow appending new post processors to the list of before publish and after receive post processors respectively.

src/reference/asciidoc/whats-new.adoc

@@ -127,3 +127,11 @@ See <<message-properties-converters>> for more information.
 
 Recovery of failed producer-created batches is now supported.
 See <<batch-retry>> for more information.
+
+A new listener container property `consumeDelay` is now available; it is helpful when using the https://github.com/rabbitmq/rabbitmq-sharding[RabbitMQ Sharding Plugin].
+See <<containerAttributes>> for more information.
+
+==== MessagePostProcessor Changes
+
+The compressing `MessagePostProcessor` s now can use a comma to separate multiple content encodings instead of a colon.
+See the IMPORTANT note in <<post-processing>> for more information.