diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java index 5bd765ef..006975db 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractCompressingPostProcessor.java +++ b/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.MessagePostProcessor; 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,19 @@ public abstract class AbstractCompressingPostProcessor implements MessagePostPro this.copyProperties = copyProperties; } + /** + * Set a delimiter to be added between the compression type and the original encoding, + * if any. Defaults to {@code ", "} (since 2.3); for compatibility with consumers + * using versions of spring-amqp earlier than 2.2.12, set it to {@code ":"} (no + * trailing space). + * @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 +126,9 @@ public abstract class AbstractCompressingPostProcessor implements MessagePostPro MessageProperties messageProperties = messagePropertiesBuilder.setContentEncoding(getEncoding() + - (originalProperties.getContentEncoding() == null + (!StringUtils.hasText(originalProperties.getContentEncoding()) ? "" - : ":" + originalProperties.getContentEncoding())) + : this.encodingDelimiter + originalProperties.getContentEncoding())) .build(); return new Message(compressed, messageProperties); diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractDecompressingPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractDecompressingPostProcessor.java index 5e56bc59..a8274c11 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/AbstractDecompressingPostProcessor.java +++ b/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 abstract class AbstractDecompressingPostProcessor implements MessagePostP 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); diff --git a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java b/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java index c11a96eb..ed44cfb1 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java +++ b/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 class DelegatingDecompressingPostProcessor implements MessagePostProcesso 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) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java index c55c6216..12cf4666 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/BatchingRabbitTemplateTests.java @@ -378,6 +378,7 @@ public class BatchingRabbitTemplateTests { BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler); template.setConnectionFactory(this.connectionFactory); GZipPostProcessor gZipPostProcessor = new GZipPostProcessor(); + gZipPostProcessor.setEncodingDelimiter(":"); // unzip messages from older versions assertThat(getStreamLevel(gZipPostProcessor)).isEqualTo(Deflater.BEST_SPEED); template.setBeforePublishPostProcessors(gZipPostProcessor); MessageProperties props = new MessageProperties(); @@ -489,7 +490,7 @@ public class BatchingRabbitTemplateTests { message = new Message("bar".getBytes(), props); template.send("", ROUTE, message); message = receive(template); - assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("gzip:foo"); + assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("gzip, foo"); GUnzipPostProcessor unzipper = new GUnzipPostProcessor(); message = unzipper.postProcessMessage(message); assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar"); @@ -500,10 +501,14 @@ public class BatchingRabbitTemplateTests { 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 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); @@ -512,6 +517,7 @@ public class BatchingRabbitTemplateTests { 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 @@ -550,7 +556,7 @@ public class BatchingRabbitTemplateTests { 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"); @@ -612,7 +618,7 @@ public class BatchingRabbitTemplateTests { 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"); diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java index 7d95d463..f4e5baaa 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/core/RabbitTemplateIntegrationTests.java @@ -1393,7 +1393,7 @@ public class RabbitTemplateIntegrationTests { Message message = new Message("foo".getBytes(), props); Message reply = template.sendAndReceive("", ROUTE, message); assertThat(reply).isNotNull(); - assertThat(reply.getMessageProperties().getContentEncoding()).isEqualTo("gzip:UTF-8"); + assertThat(reply.getMessageProperties().getContentEncoding()).isEqualTo("gzip, UTF-8"); GUnzipPostProcessor unzipper = new GUnzipPostProcessor(); reply = unzipper.postProcessMessage(reply); assertThat(new String(reply.getBody())).isEqualTo("FOO"); diff --git a/src/reference/asciidoc/amqp.adoc b/src/reference/asciidoc/amqp.adoc index 48b41ab1..3252cf6a 100644 --- a/src/reference/asciidoc/amqp.adoc +++ b/src/reference/asciidoc/amqp.adoc @@ -3946,6 +3946,12 @@ 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. +The decompressors will work with both delimiters. +However, if you publish messages with 2.3 or later and consume with 2.2.11 or earlier, you MUST set the `encodingDelimiter` property on the compressor(s) to `:`. +When your consumers are upgraded to 2.2.11 or later, you can revert to the default of `, `. + 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. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 50a7bdf7..7313332c 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -38,3 +38,9 @@ See <> 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 <> for more information. + +==== MessagePostProcessor Changes + +The compressing `MessagePostProcessor` s now use a comma to separate multiple content encodings instead of a colon. +The decompressors can handle both formats but, if you produce messages with this version that are consumed by versions earlier than 2.2.12, you should configure the compressor to use the old delimiter. +See the IMPORTANT note in <> for more information.