Fixes #7566 by handling concatenated GZIP streams.

Motivation:
According to RFC 1952, concatenation of valid gzip streams is also a valid gzip stream. JdkZlibDecoder only processed the first and discarded the rest.

Modifications:
- Introduced a constructor argument decompressConcatenated that if true, JdkZlibDecoder would continue to process the stream.

Result:
- If 'decompressConcatenated = true', concatenated streams would be processed in
compliance to RFC 1952.
- If 'decompressConcatenated = false' (default), existing behavior would remain.

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

@@ -39,6 +39,7 @@ public class JdkZlibDecoder extends ZlibDecoder {
 
     // GZIP related
     private final ByteBufChecksum crc;
+    private final boolean decompressConcatenated;
 
     private enum GzipState {
         HEADER_START,
@@ -63,7 +64,7 @@ public class JdkZlibDecoder extends ZlibDecoder {
      * Creates a new instance with the default wrapper ({@link ZlibWrapper#ZLIB}).
      */
     public JdkZlibDecoder() {
-        this(ZlibWrapper.ZLIB, null);
+        this(ZlibWrapper.ZLIB, null, false);
     }
 
     /**
@@ -72,7 +73,7 @@ public class JdkZlibDecoder extends ZlibDecoder {
      * supports the preset dictionary.
      */
     public JdkZlibDecoder(byte[] dictionary) {
-        this(ZlibWrapper.ZLIB, dictionary);
+        this(ZlibWrapper.ZLIB, dictionary, false);
     }
 
     /**
@@ -81,13 +82,22 @@ public class JdkZlibDecoder extends ZlibDecoder {
      * supported atm.
      */
     public JdkZlibDecoder(ZlibWrapper wrapper) {
-        this(wrapper, null);
+        this(wrapper, null, false);
     }
 
-    private JdkZlibDecoder(ZlibWrapper wrapper, byte[] dictionary) {
+    public JdkZlibDecoder(ZlibWrapper wrapper, boolean decompressConcatenated) {
+        this(wrapper, null, decompressConcatenated);
+    }
+
+    public JdkZlibDecoder(boolean decompressConcatenated) {
+        this(ZlibWrapper.GZIP, null, decompressConcatenated);
+    }
+
+    private JdkZlibDecoder(ZlibWrapper wrapper, byte[] dictionary, boolean decompressConcatenated) {
         if (wrapper == null) {
             throw new NullPointerException("wrapper");
         }
+        this.decompressConcatenated = decompressConcatenated;
         switch (wrapper) {
             case GZIP:
                 inflater = new Inflater(true);
@@ -207,7 +217,13 @@ public class JdkZlibDecoder extends ZlibDecoder {
             if (readFooter) {
                 gzipState = GzipState.FOOTER_START;
                 if (readGZIPFooter(in)) {
-                    finished = true;
+                    finished = !decompressConcatenated;
+
+                    if (!finished) {
+                        inflater.reset();
+                        crc.reset();
+                        gzipState = GzipState.HEADER_START;
+                    }
                 }
             }
         } catch (DataFormatException e) {

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

@@ -113,7 +113,7 @@ public final class ZlibCodecFactory {
         if (PlatformDependent.javaVersion() < 7 || noJdkZlibDecoder) {
             return new JZlibDecoder();
         } else {
-            return new JdkZlibDecoder();
+            return new JdkZlibDecoder(true);
         }
     }
 
@@ -121,7 +121,7 @@ public final class ZlibCodecFactory {
         if (PlatformDependent.javaVersion() < 7 || noJdkZlibDecoder) {
             return new JZlibDecoder(wrapper);
         } else {
-            return new JdkZlibDecoder(wrapper);
+            return new JdkZlibDecoder(wrapper, true);
         }
     }
 

codec/src/test/java/io/netty/handler/codec/compression/JdkZlibTest.java

@@ -15,8 +15,20 @@
  */
 package io.netty.handler.codec.compression;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.util.CharsetUtil;
+import io.netty.util.ReferenceCountUtil;
+import org.apache.commons.compress.utils.IOUtils;
 import org.junit.Test;
 
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Queue;
+
+import static org.junit.Assert.*;
+
 
 public class JdkZlibTest extends ZlibTest {
 
@@ -35,4 +47,47 @@ public class JdkZlibTest extends ZlibTest {
     public void testZLIB_OR_NONE3() throws Exception {
         super.testZLIB_OR_NONE3();
     }
+
+    @Test
+    // verifies backward compatibility
+    public void testConcatenatedStreamsReadFirstOnly() throws IOException {
+        EmbeddedChannel chDecoderGZip = new EmbeddedChannel(createDecoder(ZlibWrapper.GZIP));
+
+        try {
+            byte[] bytes = IOUtils.toByteArray(getClass().getResourceAsStream("/multiple.gz"));
+
+            assertTrue(chDecoderGZip.writeInbound(Unpooled.copiedBuffer(bytes)));
+            Queue<Object> messages = chDecoderGZip.inboundMessages();
+            assertEquals(1, messages.size());
+
+            ByteBuf msg = (ByteBuf) messages.poll();
+            assertEquals("a", msg.toString(CharsetUtil.UTF_8));
+            ReferenceCountUtil.release(msg);
+        } finally {
+            assertFalse(chDecoderGZip.finish());
+            chDecoderGZip.close();
+        }
+    }
+
+    @Test
+    public void testConcatenatedStreamsReadFully() throws IOException {
+        EmbeddedChannel chDecoderGZip = new EmbeddedChannel(new JdkZlibDecoder(true));
+
+        try {
+            byte[] bytes = IOUtils.toByteArray(getClass().getResourceAsStream("/multiple.gz"));
+
+            assertTrue(chDecoderGZip.writeInbound(Unpooled.copiedBuffer(bytes)));
+            Queue<Object> messages = chDecoderGZip.inboundMessages();
+            assertEquals(2, messages.size());
+
+            for (String s : Arrays.asList("a", "b")) {
+                ByteBuf msg = (ByteBuf) messages.poll();
+                assertEquals(s, msg.toString(CharsetUtil.UTF_8));
+                ReferenceCountUtil.release(msg);
+            }
+        } finally {
+            assertFalse(chDecoderGZip.finish());
+            chDecoderGZip.close();
+        }
+    }
 }