Use stream in compression codecs

And not ByteBuf, better interop.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/compression/CommonsCompressCompressionCodecFactory.java

@@ -24,10 +24,10 @@
  * href="https://commons.apache.org/proper/commons-compress/">Apache Commons Compress</a>
  * implementations.
  *
- * The framed format is used for SNAPPY and LZ4.
+ * <p>The framed format is used for SNAPPY and LZ4.
  *
- * All but ZSTD compression codecs are implemented in Commons Compress. The ZSTD codec
- * uses the <a href="https://github.com/luben/zstd-jni/">zstd-jni</a> library.
+ * <p>All but ZSTD compression codecs are implemented in Commons Compress. The ZSTD codec uses the
+ * <a href="https://github.com/luben/zstd-jni/">zstd-jni</a> library.
  */
 public class CommonsCompressCompressionCodecFactory implements CompressionCodecFactory {
   private final CompressionCodec[] codecs = new CompressionCodec[5];

src/main/java/com/rabbitmq/stream/compression/CompressionCodec.java

@@ -14,7 +14,6 @@
 
 package com.rabbitmq.stream.compression;
 
-import io.netty.buffer.ByteBuf;
 import java.io.InputStream;
 import java.io.OutputStream;
 
@@ -32,18 +31,18 @@ public interface CompressionCodec {
   /**
    * Creates an {@link OutputStream} to compress data.
    *
-   * @param target {@link ByteBuf} to write compressed data to
+   * @param target the stream where compressed data will end up
    * @return output stream to write plain data to
    */
-  OutputStream compress(ByteBuf target);
+  OutputStream compress(OutputStream target);
 
   /**
    * Creates an {@link InputStream} to read decompressed data from.
    *
-   * @param source the {@link ByteBuf} to read compressed from
-   * @return input stream to read decompressed from
+   * @param source the stream to read compressed data from
+   * @return input stream to read decompressed data from
    */
-  InputStream decompress(ByteBuf source);
+  InputStream decompress(InputStream source);
 
   /**
    * Return the code for this type of codec.

src/main/java/com/rabbitmq/stream/compression/CompressionCodecFactory.java

@@ -14,15 +14,14 @@
 
 package com.rabbitmq.stream.compression;
 
-/**
- * Factory to create {@link CompressionCodec} instances.
- */
+/** Factory to create {@link CompressionCodec} instances. */
 public interface CompressionCodecFactory {
 
   /**
    * Get a compression codec for a given type of compression.
+   *
    * @param compression the type of compression codec
    * @return the appropriate compression codec
    */
   CompressionCodec get(Compression compression);
-}
+}

src/main/java/com/rabbitmq/stream/compression/CompressionUtils.java

@@ -3,9 +3,6 @@
 import com.github.luben.zstd.Zstd;
 import com.github.luben.zstd.ZstdInputStream;
 import com.github.luben.zstd.ZstdOutputStream;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufInputStream;
-import io.netty.buffer.ByteBufOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
@@ -30,9 +27,7 @@
 import org.xerial.snappy.SnappyFramedInputStream;
 import org.xerial.snappy.SnappyFramedOutputStream;
 
-/**
- * Implementation of {@link CompressionCodec}s.
- */
+/** Implementation of {@link CompressionCodec}s. */
 public final class CompressionUtils {
 
   private CompressionUtils() {}
@@ -45,18 +40,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new GZIPOutputStream(new ByteBufOutputStream(byteBuf));
+        return new GZIPOutputStream(outputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating GZIP compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new GZIPInputStream(new ByteBufInputStream(byteBuf));
+        return new GZIPInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating GZIP compression input stream", e);
       }
@@ -81,18 +76,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new ZstdOutputStream(new ByteBufOutputStream(byteBuf));
+        return new ZstdOutputStream(outputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating Zstd compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new ZstdInputStream(new ByteBufInputStream(byteBuf));
+        return new ZstdInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating Zstd compression input stream", e);
       }
@@ -120,19 +115,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new LZ4FrameOutputStream(
-            new ByteBufOutputStream(byteBuf), BLOCKSIZE.SIZE_64KB, DEFAULT_FEATURES);
+        return new LZ4FrameOutputStream(outputStream, BLOCKSIZE.SIZE_64KB, DEFAULT_FEATURES);
       } catch (IOException e) {
         throw new CompressionException("Error while creating LZ4 compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new LZ4FrameInputStream(new ByteBufInputStream(byteBuf));
+        return new LZ4FrameInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating LZ4 compression input stream", e);
       }
@@ -157,18 +151,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new SnappyFramedOutputStream(new ByteBufOutputStream(byteBuf));
+        return new SnappyFramedOutputStream(outputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating snappy compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new SnappyFramedInputStream(new ByteBufInputStream(byteBuf));
+        return new SnappyFramedInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating snappy compression input stream", e);
       }
@@ -193,18 +187,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new GzipCompressorOutputStream(new ByteBufOutputStream(byteBuf));
+        return new GzipCompressorOutputStream(outputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating GZIP compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new GzipCompressorInputStream(new ByteBufInputStream(byteBuf));
+        return new GzipCompressorInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating GZIP compression input stream", e);
       }
@@ -230,21 +224,19 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new FramedSnappyCompressorOutputStream(new ByteBufOutputStream(byteBuf));
+        return new FramedSnappyCompressorOutputStream(outputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating Snappy compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
         return new FramedSnappyCompressorInputStream(
-            new ByteBufInputStream(byteBuf),
-            SnappyFramedOutputStream.DEFAULT_BLOCK_SIZE,
-            FramedSnappyDialect.STANDARD);
+            inputStream, SnappyFramedOutputStream.DEFAULT_BLOCK_SIZE, FramedSnappyDialect.STANDARD);
       } catch (IOException e) {
         throw new CompressionException("Error while creating Snappy compression input stream", e);
       }
@@ -272,18 +264,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new FramedLZ4CompressorOutputStream(new ByteBufOutputStream(byteBuf), DEFAULT);
+        return new FramedLZ4CompressorOutputStream(outputStream, DEFAULT);
       } catch (IOException e) {
         throw new CompressionException("Error while creating LZ4 compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new FramedLZ4CompressorInputStream(new ByteBufInputStream(byteBuf));
+        return new FramedLZ4CompressorInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating LZ4 compression input stream", e);
       }
@@ -308,18 +300,18 @@ public int maxCompressedLength(int sourceLength) {
     }
 
     @Override
-    public OutputStream compress(ByteBuf byteBuf) {
+    public OutputStream compress(OutputStream outputStream) {
       try {
-        return new ZstdCompressorOutputStream(new ByteBufOutputStream(byteBuf));
+        return new ZstdCompressorOutputStream(outputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating Zstd compression output stream", e);
       }
     }
 
     @Override
-    public InputStream decompress(ByteBuf byteBuf) {
+    public InputStream decompress(InputStream inputStream) {
       try {
-        return new ZstdCompressorInputStream(new ByteBufInputStream(byteBuf));
+        return new ZstdCompressorInputStream(inputStream);
       } catch (IOException e) {
         throw new CompressionException("Error while creating Zstd compression input stream", e);
       }

src/main/java/com/rabbitmq/stream/impl/Client.java

@@ -73,6 +73,7 @@
 import io.netty.bootstrap.Bootstrap;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufOutputStream;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelHandlerContext;
@@ -1543,7 +1544,7 @@ public void add(Codec.EncodedMessage encodedMessage) {
     public void close() {
       int maxCompressedLength = codec.maxCompressedLength(this.uncompressedByteSize);
       this.buffer = allocator.buffer(maxCompressedLength);
-      OutputStream outputStream = this.codec.compress(buffer);
+      OutputStream outputStream = this.codec.compress(new ByteBufOutputStream(buffer));
       try {
         for (int i = 0; i < messages.size(); i++) {
           final int size = messages.get(i).getSize();

src/main/java/com/rabbitmq/stream/impl/ServerFrameHandler.java

@@ -64,6 +64,7 @@
 import com.rabbitmq.stream.impl.Client.SubscriptionOffset;
 import com.rabbitmq.stream.metrics.MetricsCollector;
 import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufInputStream;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.timeout.IdleStateHandler;
 import java.io.IOException;
@@ -393,7 +394,7 @@ static int handleDeliver(
             CompressionCodec compressionCodec = client.compressionCodecFactory.get(comp);
             ByteBuf outBb = client.channel.alloc().heapBuffer(uncompressedDataSize);
             ByteBuf slice = message.slice(message.readerIndex(), dataSize);
-            InputStream inputStream = compressionCodec.decompress(slice);
+            InputStream inputStream = compressionCodec.decompress(new ByteBufInputStream(slice));
             byte[] inBuffer = new byte[uncompressedDataSize < 1024 ? uncompressedDataSize : 1024];
             int n;
             try {

src/test/java/com/rabbitmq/stream/benchmark/CompressDecompressBenchmark.java

@@ -17,6 +17,8 @@
 import com.rabbitmq.stream.compression.CompressionCodec;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.buffer.ByteBufOutputStream;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
@@ -87,7 +89,7 @@ public void setUp() throws Exception {
 
     int maxCompressedLength = codec.maxCompressedLength(plainData.length);
     ByteBuf bb = allocator.heapBuffer(maxCompressedLength);
-    OutputStream compress = codec.compress(bb);
+    OutputStream compress = codec.compress(new ByteBufOutputStream(bb));
     compress.write(plainData);
     compress.flush();
     compress.close();
@@ -102,7 +104,7 @@ public void setUp() throws Exception {
   public void compress() throws Exception {
     int maxCompressedLength = codec.maxCompressedLength(plainData.length);
     ByteBuf bb = allocator.buffer(maxCompressedLength);
-    OutputStream compress = codec.compress(bb);
+    OutputStream compress = codec.compress(new ByteBufOutputStream(bb));
     compress.write(plainData);
     compress.flush();
     compress.close();
@@ -113,7 +115,7 @@ public void compress() throws Exception {
   public void decodeReadBytePerByte() throws Exception {
     compressedDataBb.readerIndex(0);
     ByteBuf outBb = allocator.buffer(compressedData.length);
-    InputStream inputStream = codec.decompress(compressedDataBb);
+    InputStream inputStream = codec.decompress(new ByteBufInputStream(compressedDataBb));
     int n;
     while (-1 != (n = inputStream.read())) {
       outBb.writeByte(n);
@@ -134,7 +136,7 @@ public void decodeReadBytePerByte() throws Exception {
   @Benchmark
   public void decodePreAllocatedArray() throws Exception {
     compressedDataBb.readerIndex(0);
-    InputStream inputStream = codec.decompress(compressedDataBb);
+    InputStream inputStream = codec.decompress(new ByteBufInputStream(compressedDataBb));
     ByteBuf outBb = allocator.buffer(plainData.length);
     byte[] inBuffer = new byte[compressedData.length];
     int n;
@@ -157,7 +159,7 @@ public void decodePreAllocatedArray() throws Exception {
   @Benchmark
   public void decodeWriteInPreAllocatedHeapByteBuf() throws Exception {
     compressedDataBb.readerIndex(0);
-    InputStream inputStream = codec.decompress(compressedDataBb);
+    InputStream inputStream = codec.decompress(new ByteBufInputStream(compressedDataBb));
     ByteBuf outBb = allocator.heapBuffer(plainData.length);
     inputStream.read(outBb.array(), 0, plainData.length);
 

src/test/java/com/rabbitmq/stream/impl/CompressionCodecsTest.java

@@ -30,6 +30,7 @@
 import com.rabbitmq.stream.impl.Client.EncodedMessageBatch;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.ByteBufInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
@@ -98,7 +99,7 @@ void compressedEncodedMessageBatch(
 
     ByteBuf outBb = allocator.buffer(plainSize);
     destinationBb.readerIndex(0);
-    InputStream inputStream = decompressionCodec.decompress(destinationBb);
+    InputStream inputStream = decompressionCodec.decompress(new ByteBufInputStream(destinationBb));
     int n;
     while (-1 != (n = inputStream.read())) {
       outBb.writeByte(n);