GH-1094: Add Deflater and Inflator PostProcessors

Fixes #1094

* Add/Fix Author tags and change copyright year.

* Added documentation.

Author: dldiehl77

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DeflaterPostProcessor.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2019 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.support.postprocessor;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.zip.DeflaterOutputStream;
+
+/**
+ * A post processor that uses a {@link DeflaterOutputStream} to compress the message body.
+ * Sets {@link org.springframework.amqp.core.MessageProperties#SPRING_AUTO_DECOMPRESS} to
+ * true by default.
+ *
+ * @author David Diehl
+ * @since 2.2
+ */
+public class DeflaterPostProcessor extends AbstractDeflaterPostProcessor {
+
+	public DeflaterPostProcessor() {
+		super();
+	}
+
+	public DeflaterPostProcessor(boolean autoDecompress) {
+		super(autoDecompress);
+	}
+
+	@Override
+	protected OutputStream getCompressorStream(OutputStream zipped) throws IOException {
+		return new DeflaterPostProcessor.SettableLevelDeflaterOutputStream(zipped, getLevel());
+	}
+
+	@Override
+	protected String getEncoding() {
+		return "deflate";
+	}
+
+	private static final class SettableLevelDeflaterOutputStream extends DeflaterOutputStream {
+
+		SettableLevelDeflaterOutputStream(OutputStream out, int level) {
+			super(out);
+			this.def.setLevel(level);
+		}
+
+	}
+}

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/DelegatingDecompressingPostProcessor.java

@@ -26,9 +26,10 @@
 
 /**
  * A {@link MessagePostProcessor} that delegates to one of its {@link MessagePostProcessor}s
- * depending on the content encoding. Supports {@code gzip, zip} by default.
+ * depending on the content encoding. Supports {@code gzip, zip, deflate} by default.
  *
  * @author Gary Russell
+ * @author David Diehl
  * @since 1.4.2
  */
 public class DelegatingDecompressingPostProcessor implements MessagePostProcessor, Ordered {
@@ -40,6 +41,7 @@ public class DelegatingDecompressingPostProcessor implements MessagePostProcesso
 	public DelegatingDecompressingPostProcessor() {
 		this.decompressors.put("gzip", new GUnzipPostProcessor());
 		this.decompressors.put("zip", new UnzipPostProcessor());
+		this.decompressors.put("deflate", new InflaterPostProcessor());
 	}
 
 	@Override

spring-amqp/src/main/java/org/springframework/amqp/support/postprocessor/InflaterPostProcessor.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2019 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.support.postprocessor;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.zip.InflaterInputStream;
+
+/**
+ * A post processor that uses a {@link InflaterInputStream} to decompress the
+ * message body.
+ *
+ * @author David Diehl
+ * @since 2.2
+ */
+public class InflaterPostProcessor extends AbstractDecompressingPostProcessor {
+	public InflaterPostProcessor() {
+	}
+
+	public InflaterPostProcessor(boolean alwaysDecompress) {
+		super(alwaysDecompress);
+	}
+
+	protected InputStream getDecompressorStream(InputStream zipped) throws IOException {
+		return new InflaterInputStream(zipped);
+	}
+
+	protected String getEncoding() {
+		return "deflate";
+	}
+}

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

@@ -57,9 +57,11 @@
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.amqp.support.postprocessor.AbstractCompressingPostProcessor;
+import org.springframework.amqp.support.postprocessor.DeflaterPostProcessor;
 import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor;
 import org.springframework.amqp.support.postprocessor.GUnzipPostProcessor;
 import org.springframework.amqp.support.postprocessor.GZipPostProcessor;
+import org.springframework.amqp.support.postprocessor.InflaterPostProcessor;
 import org.springframework.amqp.support.postprocessor.UnzipPostProcessor;
 import org.springframework.amqp.support.postprocessor.ZipPostProcessor;
 import org.springframework.amqp.utils.test.TestUtils;
@@ -72,6 +74,7 @@
  * @author Gary Russell
  * @author Artem Bilan
  * @author Mohammad Hewedy
+ * @author David Diehl
  *
  * @since 1.4.1
  *
@@ -517,6 +520,68 @@ public void testSimpleBatchZippedWithEncoding() throws Exception {
 		assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
 	}
 
+	@Test
+	public void testSimpleBatchDeflater() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		DeflaterPostProcessor deflaterPostProcessor = new DeflaterPostProcessor();
+		assertThat(getStreamLevel(deflaterPostProcessor)).isEqualTo(Deflater.BEST_SPEED);
+		template.setBeforePublishPostProcessors(deflaterPostProcessor);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = receive(template);
+		assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("deflate");
+		InflaterPostProcessor inflater = new InflaterPostProcessor();
+		message = inflater.postProcessMessage(message);
+		assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
+	}
+
+	@Test
+	public void testSimpleBatchDeflaterBestCompression() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		DeflaterPostProcessor deflaterPostProcessor = new DeflaterPostProcessor();
+		deflaterPostProcessor.setLevel(Deflater.BEST_COMPRESSION);
+		assertThat(getStreamLevel(deflaterPostProcessor)).isEqualTo(Deflater.BEST_COMPRESSION);
+		template.setBeforePublishPostProcessors(deflaterPostProcessor);
+		MessageProperties props = new MessageProperties();
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = receive(template);
+		assertThat(message.getMessageProperties().getContentEncoding()).isEqualTo("deflate");
+		InflaterPostProcessor inflater = new InflaterPostProcessor();
+		message = inflater.postProcessMessage(message);
+		assertThat(new String(message.getBody())).isEqualTo("\u0000\u0000\u0000\u0003foo\u0000\u0000\u0000\u0003bar");
+	}
+
+	@Test
+	public void testSimpleBatchDeflaterWithEncoding() throws Exception {
+		BatchingStrategy batchingStrategy = new SimpleBatchingStrategy(2, Integer.MAX_VALUE, 30000);
+		BatchingRabbitTemplate template = new BatchingRabbitTemplate(batchingStrategy, this.scheduler);
+		template.setConnectionFactory(this.connectionFactory);
+		DeflaterPostProcessor deflaterPostProcessor = new DeflaterPostProcessor();
+		assertThat(getStreamLevel(deflaterPostProcessor)).isEqualTo(Deflater.BEST_SPEED);
+		template.setBeforePublishPostProcessors(deflaterPostProcessor);
+		MessageProperties props = new MessageProperties();
+		props.setContentEncoding("foo");
+		Message message = new Message("foo".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = new Message("bar".getBytes(), props);
+		template.send("", ROUTE, message);
+		message = receive(template);
+		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");
+	}
+
 	private Message receive(BatchingRabbitTemplate template) throws InterruptedException {
 		Message message = template.receive(ROUTE);
 		int n = 0;

src/reference/asciidoc/amqp.adoc

@@ -3689,7 +3689,7 @@ When using batching (see <<template-batching>>), this is invoked after the batch
 The second is invoked immediately after a message is received.
 
 These extension points are used for such features as compression and, for this purpose, several `MessagePostProcessor` implementations are provided.
-`GZipPostProcessor` and `ZipPostProcessor` compress messages before sending, and `GUnzipPostProcessor` and `UnzipPostProcessor` decompress received messages.
+`GZipPostProcessor`, `ZipPostProcessor` and `DeflaterPostProcessor` compress messages before sending, and `GUnzipPostProcessor`, `UnzipPostProcessor` and `InflaterPostProcessor` decompress received messages.
 
 NOTE: Starting with version 2.1.5, the `GZipPostProcessor` can be configured with the `copyProperties = true` option to make a  copy of the original message properties.
 By default, these properties are reused for performance reasons, and modified with compression content encoding and the optional `MessageProperties.SPRING_AUTO_DECOMPRESS` header.

src/reference/asciidoc/whats-new.adoc

@@ -90,6 +90,10 @@ See <<template-confirms>> for more information.
 
 Also, the publisher confirm type is now specified with the `ConfirmType` enum instead of the two mutually exclusive setter methods.
 
+===== New MessagePostProcessor Classes
+
+Classes `DeflaterPostProcessor` and `InflaterPostProcessor` were added to support compression and decompression, respectively, when the message content-encoding is set to `deflate`.
+
 ===== Other Changes
 
 The `Declarables` object (for declaring multiple queues, exchanges, bindings) now has a filtered getter for each type.