Reduce the memory copies in JdkZlibEncoder

Motivation:

At the moment we use a lot of unnecessary memory copies in JdkZlibEncoder. This is caused by either allocate a to small ByteBuf and expand it later or using a temporary byte array.
Beside this the memory footprint of JdkZlibEncoder is pretty high because of the byte[] used for compressing.

Modification:

- Override allocateBuffer(...) and calculate the estimatedsize in there, this reduce expanding of the ByteBuf later
- Not use byte[] in the instance itself but allocate a heap ByteBuf and write directly into the byte array

Result:

Less memory copies and smaller memory footprint

codec/src/main/java/io/netty/handler/codec/compression/JdkZlibEncoder.java

@@ -34,7 +34,6 @@ import java.util.zip.Deflater;
 public class JdkZlibEncoder extends ZlibEncoder {
 
     private final ZlibWrapper wrapper;
-    private final byte[] encodeBuf = new byte[8192];
     private final Deflater deflater;
     private volatile boolean finished;
     private volatile ChannelHandlerContext ctx;
@@ -211,23 +210,11 @@ public class JdkZlibEncoder extends ZlibEncoder {
             offset = 0;
         }
 
-        int sizeEstimate = (int) Math.ceil(inAry.length * 1.001) + 12;
-
         if (writeHeader) {
             writeHeader = false;
-            switch (wrapper) {
+            if (wrapper == ZlibWrapper.GZIP) {
-                case GZIP:
-                    out.ensureWritable(sizeEstimate + gzipHeader.length);
                 out.writeBytes(gzipHeader);
-                    break;
-                case ZLIB:
-                    out.ensureWritable(sizeEstimate + 2); // first two magic bytes
-                    break;
-                default:
-                    out.ensureWritable(sizeEstimate);
             }
-        } else {
-            out.ensureWritable(sizeEstimate);
         }
 
         if (wrapper == ZlibWrapper.GZIP) {
@@ -240,6 +227,23 @@ public class JdkZlibEncoder extends ZlibEncoder {
         }
     }
 
+    @Override
+    protected final ByteBuf allocateBuffer(ChannelHandlerContext ctx, ByteBuf msg,
+                                           boolean preferDirect) throws Exception {
+        int sizeEstimate = (int) Math.ceil(msg.readableBytes() * 1.001) + 12;
+        if (writeHeader) {
+            switch (wrapper) {
+                case GZIP:
+                    sizeEstimate += gzipHeader.length;
+                    break;
+                case ZLIB:
+                    sizeEstimate += 2; // first two magic bytes
+                    break;
+            }
+        }
+        return ctx.alloc().heapBuffer(sizeEstimate);
+    }
+
     @Override
     public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
         ChannelFuture f = finishEncode(ctx, ctx.newPromise());
@@ -268,8 +272,7 @@ public class JdkZlibEncoder extends ZlibEncoder {
         }
 
         finished = true;
-
+        ByteBuf footer = ctx.alloc().heapBuffer();
-        ByteBuf footer = ctx.alloc().buffer();
         if (writeHeader && wrapper == ZlibWrapper.GZIP) {
             // Write the GZIP header first if not written yet. (i.e. user wrote nothing.)
             writeHeader = false;
@@ -277,8 +280,14 @@ public class JdkZlibEncoder extends ZlibEncoder {
         }
 
         deflater.finish();
+
         while (!deflater.finished()) {
             deflate(footer);
+            if (!footer.isWritable()) {
+                // no more space so write it to the channel and continue
+                ctx.write(footer);
+                footer = ctx.alloc().heapBuffer();
+            }
         }
         if (wrapper == ZlibWrapper.GZIP) {
             int crcValue = (int) crc.getValue();
@@ -299,8 +308,10 @@ public class JdkZlibEncoder extends ZlibEncoder {
     private void deflate(ByteBuf out) {
         int numBytes;
         do {
-            numBytes = deflater.deflate(encodeBuf, 0, encodeBuf.length, Deflater.SYNC_FLUSH);
+            int writerIndex = out.writerIndex();
-            out.writeBytes(encodeBuf, 0, numBytes);
+            numBytes = deflater.deflate(
+                    out.array(), out.arrayOffset() + writerIndex, out.writableBytes(), Deflater.SYNC_FLUSH);
+            out.writerIndex(writerIndex + numBytes);
         } while (numBytes > 0);
     }