Repository: netty/netty
Issue: HTTP/2 Decompressor flow control interaction error
Labels: defect

The `DecompressorHttp2FrameReader` is intercepting and decompressing `ByteBuf` objects before the `onDataRead` method is called.  However the uncompressed size of the content is being counted against the flow control window instead of the compressed size (which was actually sent over the wire).  This will result in `PROTOCOL_ERROR` or `FLOW_CONTROL_ERROR` messages being incorrectly sent.


Issue comments:
@nmittler Do you see any issues with moving the `Http2InboundFlowController` from `AbstractHttp2ConnectionHandler` to `DefaultHttp2FrameReader`?

If this is OK then for consistency I may also doing the same thing for the `Http2OutboundFlowController` (moving it to `DefaultHttp2FrameWriter`)

@nmittler - On second thought this may get ugly because the stream state hasn't been verified yet.  As a result I may move the functionality from `DecompressorHttp2FrameReader` into `AbstractHttp2ConnectionHandler` after the flow control stuff.  I think this will avoid complicating the flow control (and other) interactions.

@Scottmitch I wonder if the right home for the decompression logic would be in `InboundHttp2ToHttpAdapter.onDataRead`.  WDYT?

@nmittler - I would like to have it be available for the HTTP/2 codec too.  I think it can be but I just put it in the wrong place.

@nmittler - I started working on this but realized this may create a messy merge situation with https://github.com/netty/netty/pull/2892.  I think I will wait until after that is cherry-picked.

@Scottmitch sgtm ... I'll try to get #2892 in soon.

PR https://github.com/netty/netty/pull/2907 was opened to address this issue.

Cherry picked as 3807f9fc8ee292eb35ea6c26f075f0185e2f4994


Pull request review comments:
Maybe we should have a base class called Http2FrameListenerDecorator? WDYT?

maybe just return here ... so no need for else?

Should we do this in a finally block?

May be just my personal preference, so feel free to ignore .... maybe consider if (stream == null ) return ... that way the rest of the method doesn't have to be indented and nested within an if block.

done.

Good idea.  I have a connection listener to clean up just in-case but better to clean up when it is in my control.

Missed this...No problem.

Well it appears I blew away my changes from `Http2FrameAdapter` to `Http2FrameListener`.  I will make this change now ;)


Patch that resolved the issue:
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DecompressorHttp2FrameReader.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DecompressorHttp2FrameReader.java
deleted file mode 100644
index 5de289320f0..00000000000
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DecompressorHttp2FrameReader.java
+++ /dev/null
@@ -1,251 +0,0 @@
-/*
- * Copyright 2014 The Netty Project
- *
- * The Netty Project licenses this file to you 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:
- *
- * http://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 io.netty.handler.codec.http2;
-
-import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_ENCODING;
-import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
-import static io.netty.handler.codec.http.HttpHeaders.Values.DEFLATE;
-import static io.netty.handler.codec.http.HttpHeaders.Values.GZIP;
-import static io.netty.handler.codec.http.HttpHeaders.Values.IDENTITY;
-import static io.netty.handler.codec.http.HttpHeaders.Values.XDEFLATE;
-import static io.netty.handler.codec.http.HttpHeaders.Values.XGZIP;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.embedded.EmbeddedChannel;
-import io.netty.handler.codec.AsciiString;
-import io.netty.handler.codec.ByteToMessageDecoder;
-import io.netty.handler.codec.compression.ZlibCodecFactory;
-import io.netty.handler.codec.compression.ZlibWrapper;
-import io.netty.handler.codec.http.HttpHeaders;
-
-/**
- * A HTTP2 frame reader that will decompress data frames according
- * to the {@code content-encoding} header for each stream.
- */
-public class DecompressorHttp2FrameReader extends DefaultHttp2FrameReader {
-    private static final AsciiString CONTENT_ENCODING_LOWER_CASE = CONTENT_ENCODING.toLowerCase();
-    private static final AsciiString CONTENT_LENGTH_LOWER_CASE = CONTENT_LENGTH.toLowerCase();
-    private static final Http2ConnectionAdapter CLEAN_UP_LISTENER = new Http2ConnectionAdapter() {
-        @Override
-        public void streamRemoved(Http2Stream stream) {
-            final EmbeddedChannel decoder = stream.decompressor();
-            if (decoder != null) {
-                cleanup(stream, decoder);
-            }
-        }
-    };
-
-    private final Http2Connection connection;
-    private final boolean strict;
-
-    /**
-     * Create a new instance with non-strict deflate decoding.
-     * {@link #DecompressorHttp2FrameReader(Http2Connection, boolean)}
-     */
-    public DecompressorHttp2FrameReader(Http2Connection connection) {
-        this(connection, false);
-    }
-
-    /**
-     * Create a new instance.
-     * @param strict
-     * <ul>
-     * <li>{@code true} to use use strict handling of deflate if used</li>
-     * <li>{@code false} be more lenient with decompression</li>
-     * </ul>
-     */
-    public DecompressorHttp2FrameReader(Http2Connection connection, boolean strict) {
-        this.connection = connection;
-        this.strict = strict;
-
-        connection.addListener(CLEAN_UP_LISTENER);
-    }
-
-    /**
-     * Returns a new {@link EmbeddedChannel} that decodes the HTTP2 message
-     * content encoded in the specified {@code contentEncoding}.
-     *
-     * @param contentEncoding the value of the {@code content-encoding} header
-     * @return a new {@link ByteToMessageDecoder} if the specified encoding is supported.
-     *         {@code null} otherwise (alternatively, you can throw a {@link Http2Exception}
-     *         to block unknown encoding).
-     * @throws Http2Exception If the specified encoding is not not supported and warrants an exception
-     */
-    protected EmbeddedChannel newContentDecoder(CharSequence contentEncoding) throws Http2Exception {
-        if (GZIP.equalsIgnoreCase(contentEncoding) ||
-            XGZIP.equalsIgnoreCase(contentEncoding)) {
-            return new EmbeddedChannel(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
-        }
-        if (DEFLATE.equalsIgnoreCase(contentEncoding) ||
-            XDEFLATE.equalsIgnoreCase(contentEncoding)) {
-            final ZlibWrapper wrapper = strict ? ZlibWrapper.ZLIB : ZlibWrapper.ZLIB_OR_NONE;
-            // To be strict, 'deflate' means ZLIB, but some servers were not implemented correctly.
-            return new EmbeddedChannel(ZlibCodecFactory.newZlibDecoder(wrapper));
-        }
-        // 'identity' or unsupported
-        return null;
-    }
-
-    /**
-     * Returns the expected content encoding of the decoded content.
-     * This getMethod returns {@code "identity"} by default, which is the case for
-     * most decoders.
-     *
-     * @param contentEncoding the value of the {@code content-encoding} header
-     * @return the expected content encoding of the new content.
-     * @throws Http2Exception if the {@code contentEncoding} is not supported and warrants an exception
-     */
-    protected AsciiString getTargetContentEncoding(
-            @SuppressWarnings("UnusedParameters") CharSequence contentEncoding) throws Http2Exception {
-        return HttpHeaders.Values.IDENTITY;
-    }
-
-    /**
-     * Checks if a new decoder object is needed for the stream identified by {@code streamId}.
-     * This method will modify the {@code content-encoding} header contained in {@code builder}.
-     * @param streamId The identifier for the headers inside {@code builder}
-     * @param builder Object representing headers which have been read
-     * @param endOfStream Indicates if the stream has ended
-     * @throws Http2Exception If the {@code content-encoding} is not supported
-     */
-    private void initDecoder(int streamId, Http2Headers headers, boolean endOfStream)
-            throws Http2Exception {
-        // Convert the names into a case-insensitive map.
-        final Http2Stream stream = connection.stream(streamId);
-        if (stream != null) {
-            EmbeddedChannel decoder = stream.decompressor();
-            if (decoder == null) {
-                if (!endOfStream) {
-                    // Determine the content encoding.
-                    AsciiString contentEncoding = headers.get(CONTENT_ENCODING_LOWER_CASE);
-                    if (contentEncoding == null) {
-                        contentEncoding = IDENTITY;
-                    }
-                    decoder = newContentDecoder(contentEncoding);
-                    if (decoder != null) {
-                        stream.decompressor(decoder);
-                        // Decode the content and remove or replace the existing headers
-                        // so that the message looks like a decoded message.
-                        AsciiString targetContentEncoding = getTargetContentEncoding(contentEncoding);
-                        if (IDENTITY.equalsIgnoreCase(targetContentEncoding)) {
-                            headers.remove(CONTENT_ENCODING_LOWER_CASE);
-                        } else {
-                            headers.set(CONTENT_ENCODING_LOWER_CASE, targetContentEncoding);
-                        }
-                    }
-                }
-            } else if (endOfStream) {
-                cleanup(stream, decoder);
-            }
-            if (decoder != null) {
-                // The content length will be for the compressed data.  Since we will decompress the data
-                // this content-length will not be correct.  Instead of queuing messages or delaying sending
-                // header frames...just remove the content-length header
-                headers.remove(CONTENT_LENGTH_LOWER_CASE);
-            }
-        }
-    }
-
-    /**
-     * Release remaining content from the {@link EmbeddedChannel} and remove the decoder from the {@link Http2Stream}.
-     * @param stream The stream for which {@code decoder} is the decompressor for
-     * @param decoder The decompressor for {@code stream}
-     */
-    private static void cleanup(Http2Stream stream, EmbeddedChannel decoder) {
-        if (decoder.finish()) {
-            for (;;) {
-                final ByteBuf buf = decoder.readInbound();
-                if (buf == null) {
-                    break;
-                }
-                buf.release();
-            }
-        }
-        stream.decompressor(null);
-    }
-
-    /**
-     * Read the next decoded {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist.
-     * @param decoder The channel to read from
-     * @return The next decoded {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist
-     */
-    private static ByteBuf nextReadableBuf(EmbeddedChannel decoder) {
-        for (;;) {
-            final ByteBuf buf = decoder.readInbound();
-            if (buf == null) {
-                return null;
-            }
-            if (!buf.isReadable()) {
-                buf.release();
-                continue;
-            }
-            return buf;
-        }
-    }
-
-    @Override
-    protected void notifyListenerOnDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
-            boolean endOfStream, Http2FrameListener listener) throws Http2Exception {
-        final Http2Stream stream = connection.stream(streamId);
-        final EmbeddedChannel decoder = stream == null ? null : stream.decompressor();
-        if (decoder == null) {
-            super.notifyListenerOnDataRead(ctx, streamId, data, padding, endOfStream, listener);
-        } else {
-            // call retain here as it will call release after its written to the channel
-            decoder.writeInbound(data.retain());
-            ByteBuf buf = nextReadableBuf(decoder);
-            if (buf == null) {
-                if (endOfStream) {
-                    super.notifyListenerOnDataRead(ctx, streamId, Unpooled.EMPTY_BUFFER, padding, true, listener);
-                }
-                // END_STREAM is not set and the data could not be decoded yet.
-                // The assumption has to be there will be more data frames to complete the decode.
-                // We don't have enough information here to know if this is an error.
-            } else {
-                for (;;) {
-                    final ByteBuf nextBuf = nextReadableBuf(decoder);
-                    if (nextBuf == null) {
-                        super.notifyListenerOnDataRead(ctx, streamId, buf, padding, endOfStream, listener);
-                        break;
-                    } else {
-                        super.notifyListenerOnDataRead(ctx, streamId, buf, padding, false, listener);
-                    }
-                    buf = nextBuf;
-                }
-            }
-
-            if (endOfStream) {
-                cleanup(stream, decoder);
-            }
-        }
-    }
-
-    @Override
-    protected void notifyListenerOnHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
-            int streamDependency, short weight, boolean exclusive, int padding, boolean endOfStream,
-            Http2FrameListener listener) throws Http2Exception {
-        initDecoder(streamId, headers, endOfStream);
-        super.notifyListenerOnHeadersRead(ctx, streamId, headers, streamDependency, weight,
-                exclusive, padding, endOfStream, listener);
-    }
-
-    @Override
-    protected void notifyListenerOnHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
-            int padding, boolean endOfStream, Http2FrameListener listener) throws Http2Exception {
-        initDecoder(streamId, headers, endOfStream);
-        super.notifyListenerOnHeadersRead(ctx, streamId, headers, padding, endOfStream, listener);
-    }
-}
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameReader.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameReader.java
index 259cfa5fdcc..18f05c4604d 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameReader.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameReader.java
@@ -367,23 +367,6 @@ private void verifyContinuationFrame() throws Http2Exception {
         }
     }
 
-    protected void notifyListenerOnDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data,
-            int padding, boolean endOfStream, Http2FrameListener listener) throws Http2Exception {
-        listener.onDataRead(ctx, streamId, data, padding, endOfStream);
-    }
-
-    protected void notifyListenerOnHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
-            int streamDependency, short weight, boolean exclusive, int padding,
-            boolean endOfStream, Http2FrameListener listener) throws Http2Exception {
-        listener.onHeadersRead(ctx, streamId, headers, streamDependency,
-                weight, exclusive, padding, endOfStream);
-    }
-
-    protected void notifyListenerOnHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
-            int padding, boolean endOfStream, Http2FrameListener listener) throws Http2Exception {
-        listener.onHeadersRead(ctx, streamId, headers, padding, endOfStream);
-    }
-
     private void readDataFrame(ChannelHandlerContext ctx, ByteBuf payload,
             Http2FrameListener listener) throws Http2Exception {
         short padding = readPadding(payload);
@@ -396,7 +379,7 @@ private void readDataFrame(ChannelHandlerContext ctx, ByteBuf payload,
         }
 
         ByteBuf data = payload.readSlice(dataLength);
-        notifyListenerOnDataRead(ctx, streamId, data, padding, flags.endOfStream(), listener);
+        listener.onDataRead(ctx, streamId, data, padding, flags.endOfStream());
         payload.skipBytes(payload.readableBytes());
     }
 
@@ -428,8 +411,8 @@ public void processFragment(boolean endOfHeaders, ByteBuf fragment,
                     final HeadersBlockBuilder hdrBlockBuilder = headersBlockBuilder();
                     hdrBlockBuilder.addFragment(fragment, ctx.alloc(), endOfHeaders);
                     if (endOfHeaders) {
-                        notifyListenerOnHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(),
-                                streamDependency, weight, exclusive, padding, headersFlags.endOfStream(), listener);
+                        listener.onHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(),
+                                        streamDependency, weight, exclusive, padding, headersFlags.endOfStream());
                         close();
                     }
                 }
@@ -454,8 +437,8 @@ public void processFragment(boolean endOfHeaders, ByteBuf fragment,
                 final HeadersBlockBuilder hdrBlockBuilder = headersBlockBuilder();
                 hdrBlockBuilder.addFragment(fragment, ctx.alloc(), endOfHeaders);
                 if (endOfHeaders) {
-                    notifyListenerOnHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(), padding,
-                            headersFlags.endOfStream(), listener);
+                    listener.onHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(), padding,
+                                    headersFlags.endOfStream());
                     close();
                 }
             }
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/DelegatingDecompressorFrameListener.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DelegatingDecompressorFrameListener.java
new file mode 100644
index 00000000000..4c9df927cdc
--- /dev/null
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DelegatingDecompressorFrameListener.java
@@ -0,0 +1,242 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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:
+ *
+ * http://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 io.netty.handler.codec.http2;
+
+import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_ENCODING;
+import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH;
+import static io.netty.handler.codec.http.HttpHeaders.Values.DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaders.Values.GZIP;
+import static io.netty.handler.codec.http.HttpHeaders.Values.IDENTITY;
+import static io.netty.handler.codec.http.HttpHeaders.Values.XDEFLATE;
+import static io.netty.handler.codec.http.HttpHeaders.Values.XGZIP;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.handler.codec.AsciiString;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.compression.ZlibCodecFactory;
+import io.netty.handler.codec.compression.ZlibWrapper;
+
+/**
+ * A HTTP2 frame listener that will decompress data frames according to the {@code content-encoding} header for each
+ * stream.
+ */
+public class DelegatingDecompressorFrameListener extends Http2FrameListenerDecorator {
+    private static final AsciiString CONTENT_ENCODING_LOWER_CASE = CONTENT_ENCODING.toLowerCase();
+    private static final AsciiString CONTENT_LENGTH_LOWER_CASE = CONTENT_LENGTH.toLowerCase();
+    private static final Http2ConnectionAdapter CLEAN_UP_LISTENER = new Http2ConnectionAdapter() {
+        @Override
+        public void streamRemoved(Http2Stream stream) {
+            final EmbeddedChannel decompressor = stream.decompressor();
+            if (decompressor != null) {
+                cleanup(stream, decompressor);
+            }
+        }
+    };
+
+    private final Http2Connection connection;
+    private final boolean strict;
+
+    public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener) {
+        this(connection, listener, true);
+    }
+
+    public DelegatingDecompressorFrameListener(Http2Connection connection, Http2FrameListener listener,
+                    boolean strict) {
+        super(listener);
+        this.connection = connection;
+        this.strict = strict;
+
+        connection.addListener(CLEAN_UP_LISTENER);
+    }
+
+    @Override
+    public void onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
+                    throws Http2Exception {
+        final Http2Stream stream = connection.stream(streamId);
+        final EmbeddedChannel decompressor = stream == null ? null : stream.decompressor();
+        if (decompressor == null) {
+            listener.onDataRead(ctx, streamId, data, padding, endOfStream);
+            return;
+        }
+
+        try {
+            // call retain here as it will call release after its written to the channel
+            decompressor.writeInbound(data.retain());
+            ByteBuf buf = nextReadableBuf(decompressor);
+            if (buf == null) {
+                if (endOfStream) {
+                    listener.onDataRead(ctx, streamId, Unpooled.EMPTY_BUFFER, padding, true);
+                }
+                // END_STREAM is not set and the data could not be decoded yet.
+                // The assumption has to be there will be more data frames to complete the decode.
+                // We don't have enough information here to know if this is an error.
+            } else {
+                for (;;) {
+                    final ByteBuf nextBuf = nextReadableBuf(decompressor);
+                    if (nextBuf == null) {
+                        listener.onDataRead(ctx, streamId, buf, padding, endOfStream);
+                        break;
+                    } else {
+                        listener.onDataRead(ctx, streamId, buf, padding, false);
+                    }
+                    buf = nextBuf;
+                }
+            }
+        } finally {
+            if (endOfStream) {
+                cleanup(stream, decompressor);
+            }
+        }
+    }
+
+    @Override
+    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
+                    boolean endStream) throws Http2Exception {
+        initDecompressor(streamId, headers, endStream);
+        listener.onHeadersRead(ctx, streamId, headers, padding, endStream);
+    }
+
+    @Override
+    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
+                    short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception {
+        initDecompressor(streamId, headers, endStream);
+        listener.onHeadersRead(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endStream);
+    }
+
+    /**
+     * Returns a new {@link EmbeddedChannel} that decodes the HTTP2 message content encoded in the specified
+     * {@code contentEncoding}.
+     *
+     * @param contentEncoding the value of the {@code content-encoding} header
+     * @return a new {@link ByteToMessageDecoder} if the specified encoding is supported. {@code null} otherwise
+     *         (alternatively, you can throw a {@link Http2Exception} to block unknown encoding).
+     * @throws Http2Exception If the specified encoding is not not supported and warrants an exception
+     */
+    protected EmbeddedChannel newContentDecompressor(AsciiString contentEncoding) throws Http2Exception {
+        if (GZIP.equalsIgnoreCase(contentEncoding) || XGZIP.equalsIgnoreCase(contentEncoding)) {
+            return new EmbeddedChannel(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
+        }
+        if (DEFLATE.equalsIgnoreCase(contentEncoding) || XDEFLATE.equalsIgnoreCase(contentEncoding)) {
+            final ZlibWrapper wrapper = strict ? ZlibWrapper.ZLIB : ZlibWrapper.ZLIB_OR_NONE;
+            // To be strict, 'deflate' means ZLIB, but some servers were not implemented correctly.
+            return new EmbeddedChannel(ZlibCodecFactory.newZlibDecoder(wrapper));
+        }
+        // 'identity' or unsupported
+        return null;
+    }
+
+    /**
+     * Returns the expected content encoding of the decoded content. This getMethod returns {@code "identity"} by
+     * default, which is the case for most decompressors.
+     *
+     * @param contentEncoding the value of the {@code content-encoding} header
+     * @return the expected content encoding of the new content.
+     * @throws Http2Exception if the {@code contentEncoding} is not supported and warrants an exception
+     */
+    protected AsciiString getTargetContentEncoding(@SuppressWarnings("UnusedParameters") AsciiString contentEncoding)
+                    throws Http2Exception {
+        return IDENTITY;
+    }
+
+    /**
+     * Checks if a new decompressor object is needed for the stream identified by {@code streamId}.
+     * This method will modify the {@code content-encoding} header contained in {@code headers}.
+     *
+     * @param streamId The identifier for the headers inside {@code headers}
+     * @param headers Object representing headers which have been read
+     * @param endOfStream Indicates if the stream has ended
+     * @throws Http2Exception If the {@code content-encoding} is not supported
+     */
+    private void initDecompressor(int streamId, Http2Headers headers, boolean endOfStream) throws Http2Exception {
+        final Http2Stream stream = connection.stream(streamId);
+        if (stream == null) {
+            return;
+        }
+
+        EmbeddedChannel decompressor = stream.decompressor();
+        if (decompressor == null) {
+            if (!endOfStream) {
+                // Determine the content encoding.
+                AsciiString contentEncoding = headers.get(CONTENT_ENCODING_LOWER_CASE);
+                if (contentEncoding == null) {
+                    contentEncoding = IDENTITY;
+                }
+                decompressor = newContentDecompressor(contentEncoding);
+                if (decompressor != null) {
+                    stream.decompressor(decompressor);
+                    // Decode the content and remove or replace the existing headers
+                    // so that the message looks like a decoded message.
+                    AsciiString targetContentEncoding = getTargetContentEncoding(contentEncoding);
+                    if (IDENTITY.equalsIgnoreCase(targetContentEncoding)) {
+                        headers.remove(CONTENT_ENCODING_LOWER_CASE);
+                    } else {
+                        headers.set(CONTENT_ENCODING_LOWER_CASE, targetContentEncoding);
+                    }
+                }
+            }
+        } else if (endOfStream) {
+            cleanup(stream, decompressor);
+        }
+        if (decompressor != null) {
+            // The content length will be for the compressed data. Since we will decompress the data
+            // this content-length will not be correct. Instead of queuing messages or delaying sending
+            // header frames...just remove the content-length header
+            headers.remove(CONTENT_LENGTH_LOWER_CASE);
+        }
+    }
+
+    /**
+     * Release remaining content from the {@link EmbeddedChannel} and remove the decompressor
+     * from the {@link Http2Stream}.
+     *
+     * @param stream The stream for which {@code decompressor} is the decompressor for
+     * @param decompressor The decompressor for {@code stream}
+     */
+    private static void cleanup(Http2Stream stream, EmbeddedChannel decompressor) {
+        if (decompressor.finish()) {
+            for (;;) {
+                final ByteBuf buf = decompressor.readInbound();
+                if (buf == null) {
+                    break;
+                }
+                buf.release();
+            }
+        }
+        stream.decompressor(null);
+    }
+
+    /**
+     * Read the next decompressed {@link ByteBuf} from the {@link EmbeddedChannel}
+     * or {@code null} if one does not exist.
+     *
+     * @param decompressor The channel to read from
+     * @return The next decoded {@link ByteBuf} from the {@link EmbeddedChannel} or {@code null} if one does not exist
+     */
+    private static ByteBuf nextReadableBuf(EmbeddedChannel decompressor) {
+        for (;;) {
+            final ByteBuf buf = decompressor.readInbound();
+            if (buf == null) {
+                return null;
+            }
+            if (!buf.isReadable()) {
+                buf.release();
+                continue;
+            }
+            return buf;
+        }
+    }
+}
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2FrameListenerDecorator.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2FrameListenerDecorator.java
new file mode 100644
index 00000000000..453fc35a5e3
--- /dev/null
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2FrameListenerDecorator.java
@@ -0,0 +1,105 @@
+/*
+ * Copyright 2014 The Netty Project
+ *
+ * The Netty Project licenses this file to you 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:
+ *
+ * http://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 io.netty.handler.codec.http2;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+/**
+ * Provides a decorator around a {@link Http2FrameListener} and delegates all method calls
+ */
+public class Http2FrameListenerDecorator implements Http2FrameListener {
+    protected final Http2FrameListener listener;
+
+    public Http2FrameListenerDecorator(Http2FrameListener listener) {
+        if (listener == null) {
+            throw new NullPointerException("listener");
+        }
+        this.listener = listener;
+    }
+
+    @Override
+    public void onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
+            throws Http2Exception {
+        listener.onDataRead(ctx, streamId, data, padding, endOfStream);
+    }
+
+    @Override
+    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
+            boolean endStream) throws Http2Exception {
+        listener.onHeadersRead(ctx, streamId, headers, padding, endStream);
+    }
+
+    @Override
+    public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
+            short weight, boolean exclusive, int padding, boolean endStream) throws Http2Exception {
+        listener.onHeadersRead(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endStream);
+    }
+
+    @Override
+    public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
+            boolean exclusive) throws Http2Exception {
+        listener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
+    }
+
+    @Override
+    public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
+        listener.onRstStreamRead(ctx, streamId, errorCode);
+    }
+
+    @Override
+    public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
+        listener.onSettingsAckRead(ctx);
+    }
+
+    @Override
+    public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
+        listener.onSettingsRead(ctx, settings);
+    }
+
+    @Override
+    public void onPingRead(ChannelHandlerContext ctx, ByteBuf data) throws Http2Exception {
+        listener.onPingRead(ctx, data);
+    }
+
+    @Override
+    public void onPingAckRead(ChannelHandlerContext ctx, ByteBuf data) throws Http2Exception {
+        listener.onPingAckRead(ctx, data);
+    }
+
+    @Override
+    public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId, Http2Headers headers,
+            int padding) throws Http2Exception {
+        listener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
+    }
+
+    @Override
+    public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
+            throws Http2Exception {
+        listener.onGoAwayRead(ctx, lastStreamId, errorCode, debugData);
+    }
+
+    @Override
+    public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
+            throws Http2Exception {
+        listener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
+    }
+
+    @Override
+    public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
+            ByteBuf payload) {
+        listener.onUnknownFrame(ctx, frameType, streamId, flags, payload);
+    }
+}
diff --git a/example/src/main/java/io/netty/example/http2/client/Http2ClientInitializer.java b/example/src/main/java/io/netty/example/http2/client/Http2ClientInitializer.java
index 71e2efa33ec..77041db03e6 100644
--- a/example/src/main/java/io/netty/example/http2/client/Http2ClientInitializer.java
+++ b/example/src/main/java/io/netty/example/http2/client/Http2ClientInitializer.java
@@ -25,11 +25,12 @@
 import io.netty.handler.codec.http.HttpClientUpgradeHandler;
 import io.netty.handler.codec.http.HttpMethod;
 import io.netty.handler.codec.http.HttpVersion;
-import io.netty.handler.codec.http2.DecompressorHttp2FrameReader;
 import io.netty.handler.codec.http2.DefaultHttp2Connection;
+import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
 import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
 import io.netty.handler.codec.http2.DefaultHttp2InboundFlowController;
 import io.netty.handler.codec.http2.DefaultHttp2OutboundFlowController;
+import io.netty.handler.codec.http2.DelegatingDecompressorFrameListener;
 import io.netty.handler.codec.http2.DelegatingHttp2ConnectionHandler;
 import io.netty.handler.codec.http2.DelegatingHttp2HttpConnectionHandler;
 import io.netty.handler.codec.http2.Http2ClientUpgradeCodec;
@@ -66,10 +67,11 @@ public void initChannel(SocketChannel ch) throws Exception {
         final Http2Connection connection = new DefaultHttp2Connection(false);
         final Http2FrameWriter frameWriter = frameWriter();
         connectionHandler = new DelegatingHttp2HttpConnectionHandler(connection,
-                        frameReader(connection), frameWriter,
+                        frameReader(), frameWriter,
                         new DefaultHttp2InboundFlowController(connection, frameWriter),
                         new DefaultHttp2OutboundFlowController(connection, frameWriter),
-                        InboundHttp2ToHttpAdapter.newInstance(connection, maxContentLength));
+                        new DelegatingDecompressorFrameListener(connection,
+                                        InboundHttp2ToHttpAdapter.newInstance(connection, maxContentLength)));
         responseHandler = new HttpResponseHandler();
         settingsHandler = new Http2SettingsHandler(ch.newPromise());
         if (sslCtx != null) {
@@ -146,8 +148,8 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exc
         }
     }
 
-    private static Http2FrameReader frameReader(Http2Connection connection) {
-        return new Http2InboundFrameLogger(new DecompressorHttp2FrameReader(connection), logger);
+    private static Http2FrameReader frameReader() {
+        return new Http2InboundFrameLogger(new DefaultHttp2FrameReader(), logger);
     }
 
     private static Http2FrameWriter frameWriter() {


Test patch:
diff --git a/codec-http2/src/test/java/io/netty/handler/codec/http2/DataCompressionHttp2Test.java b/codec-http2/src/test/java/io/netty/handler/codec/http2/DataCompressionHttp2Test.java
index e07baabee0d..53357f9b37e 100644
--- a/codec-http2/src/test/java/io/netty/handler/codec/http2/DataCompressionHttp2Test.java
+++ b/codec-http2/src/test/java/io/netty/handler/codec/http2/DataCompressionHttp2Test.java
@@ -80,6 +80,7 @@ public class DataCompressionHttp2Test {
     private ServerBootstrap sb;
     private Bootstrap cb;
     private Channel serverChannel;
+    private Channel serverConnectedChannel;
     private Channel clientChannel;
     private CountDownLatch serverLatch;
     private CountDownLatch clientLatch;
@@ -105,10 +106,12 @@ public void setup() throws InterruptedException {
             @Override
             protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
-                serverAdapter = new Http2TestUtil.FrameAdapter(serverConnection, new DecompressorHttp2FrameReader(
-                                serverConnection), serverListener, serverLatch, false);
+                serverAdapter = new Http2TestUtil.FrameAdapter(serverConnection,
+                                new DelegatingDecompressorFrameListener(serverConnection, serverListener),
+                                serverLatch, false);
                 p.addLast("reader", serverAdapter);
                 p.addLast(Http2CodecUtil.ignoreSettingsHandler());
+                serverConnectedChannel = ch;
             }
         });
 
@@ -284,9 +287,10 @@ public void run() {
     }
 
     @Test
-    public void deflateEncodingSingleLargeMessage() throws Exception {
-        serverLatch(new CountDownLatch(2));
-        final ByteBuf data = Unpooled.buffer(1 << 16);
+    public void deflateEncodingSingleLargeMessageReducedWindow() throws Exception {
+        serverLatch(new CountDownLatch(3));
+        final int BUFFER_SIZE = 1 << 16;
+        final ByteBuf data = Unpooled.buffer(BUFFER_SIZE);
         final EmbeddedChannel encoder = new EmbeddedChannel(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.ZLIB));
         try {
             for (int i = 0; i < data.capacity(); ++i) {
@@ -296,12 +300,25 @@ public void deflateEncodingSingleLargeMessage() throws Exception {
             final Http2Headers headers =
                     new DefaultHttp2Headers().method(POST).path(PATH)
                             .set(CONTENT_ENCODING.toLowerCase(), DEFLATE);
+            final Http2Settings settings = new Http2Settings();
+            // Assume the compression operation will reduce the size by at least 10 bytes
+            settings.initialWindowSize(BUFFER_SIZE - 10);
+            runInChannel(serverConnectedChannel, new Http2Runnable() {
+                @Override
+                public void run() {
+                    frameWriter.writeSettings(ctxServer(), settings, newPromiseServer());
+                    ctxServer().flush();
+                }
+            });
+            awaitClient();
+
             // Required because the decompressor intercepts the onXXXRead events before
             // our {@link Http2TestUtil$FrameAdapter} does.
             Http2TestUtil.FrameAdapter.getOrCreateStream(serverConnection, 3, false);
             runInChannel(clientChannel, new Http2Runnable() {
                 @Override
                 public void run() {
+                    frameWriter.writeSettings(ctxClient(), settings, newPromiseClient());
                     frameWriter.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
                     frameWriter.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
                     ctxClient().flush();
@@ -364,6 +381,10 @@ private void clientLatch(CountDownLatch latch) {
         }
     }
 
+    private void awaitClient() throws Exception {
+        clientLatch.await(5, SECONDS);
+    }
+
     private void awaitServer() throws Exception {
         serverLatch.await(5, SECONDS);
     }
@@ -375,4 +396,12 @@ private ChannelHandlerContext ctxClient() {
     private ChannelPromise newPromiseClient() {
         return ctxClient().newPromise();
     }
+
+    private ChannelHandlerContext ctxServer() {
+        return serverConnectedChannel.pipeline().firstContext();
+    }
+
+    private ChannelPromise newPromiseServer() {
+        return ctxServer().newPromise();
+    }
 }