Repository: netty/netty
Issue: HTTP/2 data write compression support
Labels: feature

The HTTP/2 codec currently does not support compression on data writes. There should be an interface which facilitates compression in the native HTTP/2 codec and also supported by the HTTP/1.x translation layer.

After https://github.com/netty/netty/issues/2793 is resolved their will be an opportunity to follow existing practices and possibly extract out commonalities (or leverage unit tests) between the two features.


Issue comments:
Note that https://github.com/netty/netty/pull/2907 was introduced to fix bugs in the original decompression code.

@nmittler - So I have looked at this briefly and have come up with the following options:
1. Create a class which inherits from `DefaultHttp2OutboundFlowController` that overrides `writeData()`, compresses the data, then calls `super.writeData()`
2. Create a new interface which is passed into `AbstractHttp2ConnectionHandler` and transforms data in `writeData()` before calling `outboundFlow.writeData()`.

| Option | Pros | Cons |
| --- | --- | --- |
| 1 | Existing interfaces don't change. Compression (and associated code) only used if runtime time is instantiated (no if checks) | Maybe not the most obvious choice to have a `Http2OutboundFlowController` also provide compression |
| 2 | Clear resposibility of classes | Interfaces would change (new set of constructors).  This will impact code even if no compression is used (require a call to a default `transformation` class or `if != null` check every `writeData` call). |

What are your thoughts on this?  Do you have any alternative approaches, or do you need any clarification on the above options?

@Scottmitch I think we want to keep compression out of the flow controller ... it already has a lot of responsibilities and is complex enough as it is :).

Compression can take a lot of forms and is not formally defined at the HTTP/2 framing layer, which is where flow control is applied.  Instead, I think that compression is a concern of the HTTP/1->HTTP/2 translation layer, which would compress the data contained within a FullHttpMessage before writing it to the HTTP/2 layer. So I think this isn't quite equivalent to either of your options ... I guess this is number 3?  WDYT?

@nmittler - I agree that even though extending `DefaultHttp2OutboundFlowController` is convenient it is likely not the best solution.

Although it is not formally defined in the http/2 spec (outside of the [warnings](https://tools.ietf.org/html/draft-ietf-httpbis-http2-14#section-10.6)) http/2 is just a proper framing layer around http and should retain all core functionality of http (see [abstract](https://tools.ietf.org/html/draft-ietf-httpbis-http2-14) and [section 2](https://tools.ietf.org/html/draft-ietf-httpbis-http2-14#section-2)).  In http you can have chunked responses where each chunk is encoded according to the content-encoding header.  The "chunked" response use case is inherently supported by http/2 with the DATA FRAME.  So just as the recently added decompression code operates on the data portion of the DATA FRAME the compression would operate on the same section.  The difference would be the decompression code operates post flow control and the compression operates pre flow control.

I think it makes sense for the HTTP/1->HTTP/2 layer to be able to do compression/decompression however I think the responsibility to provide the functionality is in the HTTP/2 proper code.  This would allow all use cases involving the HTTP2-codec to take advantage of this functionality.

Do you agree with this?

@Scottmitch I think we're basically on the same page here.  Just to clarify where I'm coming from ... 

HTTP/1 is just one protocol that can be layered atop HTTP/2 framing.  Other (e.g. new) protocols may have their own compression schemes that may "chunk" in different ways from HTTP/1 (i..e. defining where the compression boundaries are).  For example, it may be the case that a protocol may not map compression boundaries to a single call to connectionHandler.writeData().  Maybe the protocol doesn't have all of the data yet and is waiting for more data before it can write the next compression block.  Whatever the protocol decides to do, the details of compression are really a function of the upper-level protocol and not the framing layer.

With that said, I don't feel that AbstractHttp2ConnectionHandler is the right place for compression.  Some HTTP/1 translation object will be responsible for calling connectionHandler.writeData() and I think that is the object that should be responsible for applying compression.  I definitely think we should abstract this utility behind an interface and provide a default implementation - that makes a lot of sense.

@nmittler - Hum I can't quite tell if we are on the same page...seems like we are close though ;)

In general I agree with your first paragraph.  There can be many different compression schemes desired in the application layer.  However I do think we can generalize a common use case which is the application calling some form of `writeData()` and having this data compressed before `flowControl.writeData()` is invoked.  This compression is based upon some encoding (i.e. gzip, deflate, my-app-encoding, etc...) which was previously specified with a `connectionHandler.writeHeader()` call.  This interface can be made generic so that they can implement queuing if they need to wait for more data before actually doing the compression and calling `connectionHandler.writeData()`.  I think the default implementation (just like the decompression) would be to implement streaming compression based upon gzip and deflate with room for extension.

Is the HTTP/1 translation layer the desired place because it takes a FullHttpMessage and has all the content when making a call to `connectionHandler.writeData()`?  Either way isn't this just a subset of the use-case from my description in the first paragraph?

@nmittler - I opened PR https://github.com/netty/netty/pull/2935 to help demo an approach.

The re-factoring allowed for easy access in PR https://github.com/netty/netty/pull/3106.

Cherry-picked as 685075f1a058c0d5d8fc59072278bc822dce26d8


Pull request review comments:
@trustin - I think these got moved during the header back port (and then forward port).  I think it is better to keep a common definition in the HttpHeaderValue class rather than an independent definition in http/2 and http codec (and potentially spdy).  If you agree I will back port the portion related to this change after this PR is resolved.

I think you could combine the two if's: if (compressor == null && !endOfStream)...

Maybe add a comment here indicating why compressor might be null for a particular stream.

Doesn't need to be an else.

Shouldn't this be writing "buf", not "data"?

maybe make a utility method newCompressionChannel(ZlibWrapper)?

May be worth adding a comment that this does compression on the entire stream as a whole.  Message-based applications will have to roll their own compression to demarcate compressed "chunks" within a single stream.

Done.

Done.

Good catch.

Yes!  Great catch..knew I was going to do that...too much cpy/paste.

Maybe combine the two writes into one to reduce the duplication...

```
boolean endOfStreamForBuffer = nextBuf == null ? endOfStream : false;
super.writeData(ctx, streamId, buf, padding, endOfStreamForBuffer, newPromise);
if (endOfStreamForBuffer) {
    break;
}
```

Done.

Any suggestions?  I think `getTargetContentEncoding` should allow an extension point if they have another encoding on top of the compression encoding.

Humm I'm not sure this will work.  endOfStreamForBuffer can still be false (if endOfStream is false) even if there is no more next buffer.

Actually, I don't see this class being extended to support message-based compression.  I was just suggesting that we comment this class to indicate that when we're talking about compression, we mean for the entire stream. 

Ah ok, so the if would change:

```
boolean endOfStreamForBuffer = nextBuf == null ? endOfStream : false;
super.writeData(ctx, streamId, buf, padding, endOfStreamForBuffer, newPromise);
if (nextBuf == null) {
    break;
}
```

I think that will work. Thanks.

Oops I don't think I can do this.  This may lead to `compressor` null pointer exception down at the else condition when `cleanup` is called.

Check the comments I added (most recent commit) and let me know if this is what you had in mind.

ah good point.


Patch that resolved the issue:
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecompressor.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecompressor.java
index 45344677cd0..5fea256a4c0 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecompressor.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpContentDecompressor.java
@@ -15,8 +15,11 @@
  */
 package io.netty.handler.codec.http;
 
+import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
+import static io.netty.handler.codec.http.HttpHeaderValues.X_DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaderValues.X_GZIP;
 import io.netty.channel.embedded.EmbeddedChannel;
-import io.netty.handler.codec.AsciiString;
 import io.netty.handler.codec.compression.ZlibCodecFactory;
 import io.netty.handler.codec.compression.ZlibWrapper;
 
@@ -26,16 +29,6 @@
  * handler modifies the message, please refer to {@link HttpContentDecoder}.
  */
 public class HttpContentDecompressor extends HttpContentDecoder {
-
-    /**
-     * {@code "x-deflate"}
-     */
-    private static final AsciiString X_DEFLATE = new AsciiString("x-deflate");
-    /**
-     * {@code "x-gzip"}
-     */
-    private static final AsciiString X_GZIP = new AsciiString("x-gzip");
-
     private final boolean strict;
 
     /**
@@ -57,11 +50,11 @@ public HttpContentDecompressor(boolean strict) {
 
     @Override
     protected EmbeddedChannel newContentDecoder(String contentEncoding) throws Exception {
-        if (HttpHeaderValues.GZIP.equalsIgnoreCase(contentEncoding) ||
+        if (GZIP.equalsIgnoreCase(contentEncoding) ||
             X_GZIP.equalsIgnoreCase(contentEncoding)) {
             return new EmbeddedChannel(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
         }
-        if (HttpHeaderValues.DEFLATE.equalsIgnoreCase(contentEncoding) ||
+        if (DEFLATE.equalsIgnoreCase(contentEncoding) ||
             X_DEFLATE.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.
diff --git a/codec-http/src/main/java/io/netty/handler/codec/http/HttpHeaderValues.java b/codec-http/src/main/java/io/netty/handler/codec/http/HttpHeaderValues.java
index ea907cba2bd..ce4534337b5 100644
--- a/codec-http/src/main/java/io/netty/handler/codec/http/HttpHeaderValues.java
+++ b/codec-http/src/main/java/io/netty/handler/codec/http/HttpHeaderValues.java
@@ -76,6 +76,10 @@ public final class HttpHeaderValues {
      * {@code "deflate"}
      */
     public static final AsciiString DEFLATE = new AsciiString("deflate");
+    /**
+     * {@code "x-deflate"}
+     */
+    public static final AsciiString X_DEFLATE = new AsciiString("x-deflate");
     /**
      * {@code "file"}
      * See {@link HttpHeaderNames#CONTENT_DISPOSITION}
@@ -95,6 +99,10 @@ public final class HttpHeaderValues {
      * {@code "gzip"}
      */
     public static final AsciiString GZIP = new AsciiString("gzip");
+    /**
+     * {@code "x-gzip"}
+     */
+    public static final AsciiString X_GZIP = new AsciiString("x-gzip");
     /**
      * {@code "identity"}
      */
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/CompressorHttp2ConnectionEncoder.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/CompressorHttp2ConnectionEncoder.java
new file mode 100644
index 00000000000..b237c86aa2e
--- /dev/null
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/CompressorHttp2ConnectionEncoder.java
@@ -0,0 +1,289 @@
+/*
+ * 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.HttpHeaderNames.CONTENT_ENCODING;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
+import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
+import static io.netty.handler.codec.http.HttpHeaderValues.IDENTITY;
+import static io.netty.handler.codec.http.HttpHeaderValues.X_DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaderValues.X_GZIP;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.ChannelPromiseAggregator;
+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 encoder that will compress data frames according to the {@code content-encoding} header for each stream.
+ * The compression provided by this class will be applied to the data for the entire stream.
+ */
+public class CompressorHttp2ConnectionEncoder extends DefaultHttp2ConnectionEncoder {
+    private static final Http2ConnectionAdapter CLEAN_UP_LISTENER = new Http2ConnectionAdapter() {
+        @Override
+        public void streamRemoved(Http2Stream stream) {
+            final EmbeddedChannel compressor = stream.compressor();
+            if (compressor != null) {
+                cleanup(stream, compressor);
+            }
+        }
+    };
+
+    private final int compressionLevel;
+    private final int windowBits;
+    private final int memLevel;
+
+    /**
+     * Builder for new instances of {@link CompressorHttp2ConnectionEncoder}
+     */
+    public static class Builder extends DefaultHttp2ConnectionEncoder.Builder {
+        protected int compressionLevel = 6;
+        protected int windowBits = 15;
+        protected int memLevel = 8;
+
+        public Builder compressionLevel(int compressionLevel) {
+            this.compressionLevel = compressionLevel;
+            return this;
+        }
+
+        public Builder windowBits(int windowBits) {
+            this.windowBits = windowBits;
+            return this;
+        }
+
+        public Builder memLevel(int memLevel) {
+            this.memLevel = memLevel;
+            return this;
+        }
+
+        @Override
+        public CompressorHttp2ConnectionEncoder build() {
+            return new CompressorHttp2ConnectionEncoder(this);
+        }
+    }
+
+    protected CompressorHttp2ConnectionEncoder(Builder builder) {
+        super(builder);
+        if (builder.compressionLevel < 0 || builder.compressionLevel > 9) {
+            throw new IllegalArgumentException("compressionLevel: " + builder.compressionLevel + " (expected: 0-9)");
+        }
+        if (builder.windowBits < 9 || builder.windowBits > 15) {
+            throw new IllegalArgumentException("windowBits: " + builder.windowBits + " (expected: 9-15)");
+        }
+        if (builder.memLevel < 1 || builder.memLevel > 9) {
+            throw new IllegalArgumentException("memLevel: " + builder.memLevel + " (expected: 1-9)");
+        }
+        this.compressionLevel = builder.compressionLevel;
+        this.windowBits = builder.windowBits;
+        this.memLevel = builder.memLevel;
+
+        connection().addListener(CLEAN_UP_LISTENER);
+    }
+
+    @Override
+    public ChannelFuture writeData(final ChannelHandlerContext ctx, final int streamId, ByteBuf data, int padding,
+            final boolean endOfStream, ChannelPromise promise) {
+        final Http2Stream stream = connection().stream(streamId);
+        final EmbeddedChannel compressor = stream == null ? null : stream.compressor();
+        if (compressor == null) {
+            // The compressor may be null if no compatible encoding type was found in this stream's headers
+            return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
+        }
+
+        try {
+            // call retain here as it will call release after its written to the channel
+            compressor.writeOutbound(data.retain());
+            ByteBuf buf = nextReadableBuf(compressor);
+            if (buf == null) {
+                if (endOfStream) {
+                    return super.writeData(ctx, streamId, Unpooled.EMPTY_BUFFER, padding, endOfStream, promise);
+                }
+                // END_STREAM is not set and the assumption is data is still forthcoming.
+                promise.setSuccess();
+                return promise;
+            }
+
+            ChannelPromiseAggregator aggregator = new ChannelPromiseAggregator(promise);
+            for (;;) {
+                final ByteBuf nextBuf = nextReadableBuf(compressor);
+                final boolean endOfStreamForBuf = nextBuf == null ? endOfStream : false;
+                ChannelPromise newPromise = ctx.newPromise();
+                aggregator.add(newPromise);
+
+                super.writeData(ctx, streamId, buf, padding, endOfStreamForBuf, newPromise);
+                if (nextBuf == null) {
+                    break;
+                }
+
+                buf = nextBuf;
+            }
+            return promise;
+        } finally {
+            if (endOfStream) {
+                cleanup(stream, compressor);
+            }
+        }
+    }
+
+    @Override
+    public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
+            boolean endStream, ChannelPromise promise) {
+        initCompressor(streamId, headers, endStream);
+        return super.writeHeaders(ctx, streamId, headers, padding, endStream, promise);
+    }
+
+    @Override
+    public ChannelFuture writeHeaders(final ChannelHandlerContext ctx, final int streamId, final Http2Headers headers,
+            final int streamDependency, final short weight, final boolean exclusive, final int padding,
+            final boolean endOfStream, final ChannelPromise promise) {
+        initCompressor(streamId, headers, endOfStream);
+        return super.writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive, padding, endOfStream,
+                promise);
+    }
+
+    /**
+     * Returns a new {@link EmbeddedChannel} that encodes 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 newContentCompressor(AsciiString contentEncoding) throws Http2Exception {
+        if (GZIP.equalsIgnoreCase(contentEncoding) || X_GZIP.equalsIgnoreCase(contentEncoding)) {
+            return newCompressionChannel(ZlibWrapper.GZIP);
+        }
+        if (DEFLATE.equalsIgnoreCase(contentEncoding) || X_DEFLATE.equalsIgnoreCase(contentEncoding)) {
+            return newCompressionChannel(ZlibWrapper.ZLIB);
+        }
+        // 'identity' or unsupported
+        return null;
+    }
+
+    /**
+     * Returns the expected content encoding of the decoded content. Returning {@code contentEncoding} is the default
+     * behavior, which is the case for most compressors.
+     *
+     * @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(AsciiString contentEncoding) throws Http2Exception {
+        return contentEncoding;
+    }
+
+    /**
+     * Generate a new instance of an {@link EmbeddedChannel} capable of compressing data
+     * @param wrapper Defines what type of encoder should be used
+     */
+    private EmbeddedChannel newCompressionChannel(ZlibWrapper wrapper) {
+        return new EmbeddedChannel(ZlibCodecFactory.newZlibEncoder(wrapper, compressionLevel, windowBits,
+                memLevel));
+    }
+
+    /**
+     * Checks if a new compressor 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 are to be written
+     * @param endOfStream Indicates if the stream has ended
+     */
+    private void initCompressor(int streamId, Http2Headers headers, boolean endOfStream) {
+        final Http2Stream stream = connection().stream(streamId);
+        if (stream == null) {
+            return;
+        }
+
+        EmbeddedChannel compressor = stream.compressor();
+        if (compressor == null) {
+            if (!endOfStream) {
+                AsciiString encoding = headers.get(CONTENT_ENCODING);
+                if (encoding == null) {
+                    encoding = IDENTITY;
+                }
+                try {
+                    compressor = newContentCompressor(encoding);
+                    if (compressor != null) {
+                        AsciiString targetContentEncoding = getTargetContentEncoding(encoding);
+                        if (IDENTITY.equalsIgnoreCase(targetContentEncoding)) {
+                            headers.remove(CONTENT_ENCODING);
+                        } else {
+                            headers.set(CONTENT_ENCODING, targetContentEncoding);
+                        }
+                    }
+                } catch (Throwable ignored) {
+                    // Ignore
+                }
+            }
+        } else if (endOfStream) {
+            cleanup(stream, compressor);
+        }
+
+        if (compressor != null) {
+            // The content length will be for the decompressed data. Since we will compress 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);
+        }
+    }
+
+    /**
+     * Release remaining content from {@link EmbeddedChannel} and remove the compressor from the {@link Http2Stream}.
+     *
+     * @param stream The stream for which {@code compressor} is the compressor for
+     * @param decompressor The compressor for {@code stream}
+     */
+    private static void cleanup(Http2Stream stream, EmbeddedChannel compressor) {
+        if (compressor.finish()) {
+            for (;;) {
+                final ByteBuf buf = compressor.readOutbound();
+                if (buf == null) {
+                    break;
+                }
+                buf.release();
+            }
+        }
+        stream.compressor(null);
+    }
+
+    /**
+     * Read the next compressed {@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 compressor) {
+        for (;;) {
+            final ByteBuf buf = compressor.readOutbound();
+            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/DefaultHttp2Connection.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
index 635d755dbc1..f78b35cefa5 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2Connection.java
@@ -221,6 +221,7 @@ private class DefaultStream implements Http2Stream {
         private FlowState inboundFlow;
         private FlowState outboundFlow;
         private EmbeddedChannel decompressor;
+        private EmbeddedChannel compressor;
         private Object data;
 
         DefaultStream(int id) {
@@ -310,6 +311,19 @@ public EmbeddedChannel decompressor() {
             return decompressor;
         }
 
+        @Override
+        public void compressor(EmbeddedChannel compressor) {
+            if (this.compressor != null && compressor != null) {
+                throw new IllegalStateException("compressor can not be reassigned");
+            }
+            this.compressor = compressor;
+        }
+
+        @Override
+        public EmbeddedChannel compressor() {
+            return compressor;
+        }
+
         @Override
         public FlowState inboundFlow() {
             return inboundFlow;
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
index bdb0c4a0194..88b94885e06 100644
--- 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
@@ -14,6 +14,13 @@
  */
 package io.netty.handler.codec.http2;
 
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_ENCODING;
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
+import static io.netty.handler.codec.http.HttpHeaderValues.DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaderValues.GZIP;
+import static io.netty.handler.codec.http.HttpHeaderValues.IDENTITY;
+import static io.netty.handler.codec.http.HttpHeaderValues.X_DEFLATE;
+import static io.netty.handler.codec.http.HttpHeaderValues.X_GZIP;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
@@ -22,24 +29,12 @@
 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.HttpHeaderNames;
-import io.netty.handler.codec.http.HttpHeaderValues;
 
 /**
  * A HTTP2 frame listener that will decompress data frames according to the {@code content-encoding} header for each
- * stream.
+ * stream. The decompression provided by this class will be applied to the data for the entire stream.
  */
 public class DelegatingDecompressorFrameListener extends Http2FrameListenerDecorator {
-
-    /**
-     * {@code "x-deflate"}
-     */
-    private static final AsciiString X_DEFLATE = new AsciiString("x-deflate");
-    /**
-     * {@code "x-gzip"}
-     */
-    private static final AsciiString X_GZIP = new AsciiString("x-gzip");
-
     private static final Http2ConnectionAdapter CLEAN_UP_LISTENER = new Http2ConnectionAdapter() {
         @Override
         public void streamRemoved(Http2Stream stream) {
@@ -72,6 +67,7 @@ public void onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, in
         final Http2Stream stream = connection.stream(streamId);
         final EmbeddedChannel decompressor = stream == null ? null : stream.decompressor();
         if (decompressor == null) {
+            // The decompressor may be null if no compatible encoding type was found in this stream's headers
             listener.onDataRead(ctx, streamId, data, padding, endOfStream);
             return;
         }
@@ -90,12 +86,13 @@ public void onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, in
             } else {
                 for (;;) {
                     final ByteBuf nextBuf = nextReadableBuf(decompressor);
+                    final boolean endOfStreamForBuf = nextBuf == null ? endOfStream : false;
+
+                    listener.onDataRead(ctx, streamId, buf, padding, endOfStreamForBuf);
                     if (nextBuf == null) {
-                        listener.onDataRead(ctx, streamId, buf, padding, endOfStream);
                         break;
                     }
 
-                    listener.onDataRead(ctx, streamId, buf, padding, false);
                     buf = nextBuf;
                 }
             }
@@ -130,11 +127,11 @@ public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers
      * @throws Http2Exception If the specified encoding is not not supported and warrants an exception
      */
     protected EmbeddedChannel newContentDecompressor(AsciiString contentEncoding) throws Http2Exception {
-        if (HttpHeaderValues.GZIP.equalsIgnoreCase(contentEncoding) ||
+        if (GZIP.equalsIgnoreCase(contentEncoding) ||
             X_GZIP.equalsIgnoreCase(contentEncoding)) {
             return new EmbeddedChannel(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
         }
-        if (HttpHeaderValues.DEFLATE.equalsIgnoreCase(contentEncoding) ||
+        if (DEFLATE.equalsIgnoreCase(contentEncoding) ||
             X_DEFLATE.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.
@@ -154,7 +151,7 @@ protected EmbeddedChannel newContentDecompressor(AsciiString contentEncoding) th
      */
     protected AsciiString getTargetContentEncoding(@SuppressWarnings("UnusedParameters") AsciiString contentEncoding)
                     throws Http2Exception {
-        return HttpHeaderValues.IDENTITY;
+        return IDENTITY;
     }
 
     /**
@@ -176,9 +173,9 @@ private void initDecompressor(int streamId, Http2Headers headers, boolean endOfS
         if (decompressor == null) {
             if (!endOfStream) {
                 // Determine the content encoding.
-                AsciiString contentEncoding = headers.get(HttpHeaderNames.CONTENT_ENCODING);
+                AsciiString contentEncoding = headers.get(CONTENT_ENCODING);
                 if (contentEncoding == null) {
-                    contentEncoding = HttpHeaderValues.IDENTITY;
+                    contentEncoding = IDENTITY;
                 }
                 decompressor = newContentDecompressor(contentEncoding);
                 if (decompressor != null) {
@@ -186,21 +183,22 @@ private void initDecompressor(int streamId, Http2Headers headers, boolean endOfS
                     // Decode the content and remove or replace the existing headers
                     // so that the message looks like a decoded message.
                     AsciiString targetContentEncoding = getTargetContentEncoding(contentEncoding);
-                    if (HttpHeaderValues.IDENTITY.equalsIgnoreCase(targetContentEncoding)) {
-                        headers.remove(HttpHeaderNames.CONTENT_ENCODING);
+                    if (IDENTITY.equalsIgnoreCase(targetContentEncoding)) {
+                        headers.remove(CONTENT_ENCODING);
                     } else {
-                        headers.set(HttpHeaderNames.CONTENT_ENCODING, targetContentEncoding);
+                        headers.set(CONTENT_ENCODING, 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(HttpHeaderNames.CONTENT_LENGTH);
+            headers.remove(CONTENT_LENGTH);
         }
     }
 
diff --git a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Stream.java b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Stream.java
index cabbdb8f6f5..34f5c3971eb 100644
--- a/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Stream.java
+++ b/codec-http2/src/main/java/io/netty/handler/codec/http2/Http2Stream.java
@@ -153,6 +153,16 @@ enum State {
      */
     EmbeddedChannel decompressor();
 
+    /**
+     * Associate an object responsible for compressing data frames for this stream
+     */
+    void compressor(EmbeddedChannel decompressor);
+
+    /**
+     * Get the object capable of compressing data frames for this stream
+     */
+    EmbeddedChannel compressor();
+
     /**
      * Gets the in-bound flow control state for this stream.
      */


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 ca5fb29f8b2..876f824eac8 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
@@ -14,6 +14,17 @@
  */
 package io.netty.handler.codec.http2;
 
+import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
+import static io.netty.handler.codec.http2.Http2TestUtil.as;
+import static io.netty.handler.codec.http2.Http2TestUtil.runInChannel;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.buffer.ByteBuf;
@@ -35,8 +46,15 @@
 import io.netty.handler.codec.compression.ZlibWrapper;
 import io.netty.handler.codec.http.HttpHeaderNames;
 import io.netty.handler.codec.http.HttpHeaderValues;
+import io.netty.handler.codec.http2.Http2TestUtil.FrameAdapter;
+import io.netty.handler.codec.http2.Http2TestUtil.Http2Runnable;
 import io.netty.util.NetUtil;
 import io.netty.util.concurrent.Future;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
@@ -44,15 +62,6 @@
 import org.mockito.Mock;
 import org.mockito.MockitoAnnotations;
 
-import java.net.InetSocketAddress;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-
-import static io.netty.handler.codec.http2.Http2TestUtil.*;
-import static java.util.concurrent.TimeUnit.*;
-import static org.junit.Assert.*;
-import static org.mockito.Mockito.*;
-
 /**
  * Test for data decompression in the HTTP/2 codec.
  */
@@ -66,9 +75,14 @@ public class DataCompressionHttp2Test {
     private Http2FrameListener serverListener;
     @Mock
     private Http2FrameListener clientListener;
+    @Mock
+    private Http2LifecycleManager serverLifeCycleManager;
+    @Mock
+    private Http2LifecycleManager clientLifeCycleManager;
 
     private ByteBufAllocator alloc;
-    private Http2FrameWriter frameWriter;
+    private Http2ConnectionEncoder serverEncoder;
+    private Http2ConnectionEncoder clientEncoder;
     private ServerBootstrap sb;
     private Bootstrap cb;
     private Channel serverChannel;
@@ -107,22 +121,22 @@ public void teardown() throws InterruptedException {
 
     @Test
     public void justHeadersNoData() throws Exception {
-        bootstrapEnv(2, 1);
-        final Http2Headers headers =
-                new DefaultHttp2Headers().method(GET).path(PATH)
-                                         .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
+        bootstrapEnv(1, 1);
+        final Http2Headers headers = new DefaultHttp2Headers().method(GET).path(PATH)
+                .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
         // Required because the decompressor intercepts the onXXXRead events before
         // our {@link Http2TestUtil$FrameAdapter} does.
         FrameAdapter.getOrCreateStream(serverConnection, 3, false);
         runInChannel(clientChannel, new Http2Runnable() {
             @Override
             public void run() {
-                frameWriter.writeHeaders(ctxClient(), 3, headers, 0, true, newPromiseClient());
+                clientEncoder.writeHeaders(ctxClient(), 3, headers, 0, true, newPromiseClient());
                 ctxClient().flush();
             }
         });
         awaitServer();
-        verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(headers), eq(0), eq(true));
+        verify(serverListener).onHeadersRead(any(ChannelHandlerContext.class), eq(3), eq(headers), eq(0),
+                eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0), eq(true));
     }
 
     @Test
@@ -133,17 +147,16 @@ public void gzipEncodingSingleEmptyMessage() throws Exception {
         final EmbeddedChannel encoder = new EmbeddedChannel(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP));
         try {
             final ByteBuf encodedData = encodeData(data, encoder);
-            final Http2Headers headers =
-                    new DefaultHttp2Headers().method(POST).path(PATH)
-                                             .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
+            final Http2Headers headers = new DefaultHttp2Headers().method(POST).path(PATH)
+                    .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
             // Required because the decompressor intercepts the onXXXRead events before
             // our {@link Http2TestUtil$FrameAdapter} does.
             FrameAdapter.getOrCreateStream(serverConnection, 3, false);
             runInChannel(clientChannel, new Http2Runnable() {
                 @Override
                 public void run() {
-                    frameWriter.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
-                    frameWriter.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
+                    clientEncoder.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
+                    clientEncoder.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
                     ctxClient().flush();
                 }
             });
@@ -151,7 +164,7 @@ public void run() {
             data.resetReaderIndex();
             ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), dataCaptor.capture(), eq(0),
-                            eq(true));
+                    eq(true));
             dataCapture = dataCaptor.getAllValues();
             assertEquals(data, dataCapture.get(0));
         } finally {
@@ -168,17 +181,16 @@ public void gzipEncodingSingleMessage() throws Exception {
         final EmbeddedChannel encoder = new EmbeddedChannel(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP));
         try {
             final ByteBuf encodedData = encodeData(data, encoder);
-            final Http2Headers headers =
-                    new DefaultHttp2Headers().method(POST).path(PATH)
-                                             .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
+            final Http2Headers headers = new DefaultHttp2Headers().method(POST).path(PATH)
+                    .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
             // Required because the decompressor intercepts the onXXXRead events before
             // our {@link Http2TestUtil$FrameAdapter} does.
             FrameAdapter.getOrCreateStream(serverConnection, 3, false);
             runInChannel(clientChannel, new Http2Runnable() {
                 @Override
                 public void run() {
-                    frameWriter.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
-                    frameWriter.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
+                    clientEncoder.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
+                    clientEncoder.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
                     ctxClient().flush();
                 }
             });
@@ -186,7 +198,7 @@ public void run() {
             data.resetReaderIndex();
             ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), dataCaptor.capture(), eq(0),
-                            eq(true));
+                    eq(true));
             dataCapture = dataCaptor.getAllValues();
             assertEquals(data, dataCapture.get(0));
         } finally {
@@ -206,18 +218,17 @@ public void gzipEncodingMultipleMessages() throws Exception {
         try {
             final ByteBuf encodedData1 = encodeData(data1, encoder);
             final ByteBuf encodedData2 = encodeData(data2, encoder);
-            final Http2Headers headers =
-                    new DefaultHttp2Headers().method(POST).path(PATH)
-                                             .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
+            final Http2Headers headers = new DefaultHttp2Headers().method(POST).path(PATH)
+                    .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.GZIP);
             // Required because the decompressor intercepts the onXXXRead events before
             // our {@link Http2TestUtil$FrameAdapter} does.
             FrameAdapter.getOrCreateStream(serverConnection, 3, false);
             runInChannel(clientChannel, new Http2Runnable() {
                 @Override
                 public void run() {
-                    frameWriter.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
-                    frameWriter.writeData(ctxClient(), 3, encodedData1, 0, false, newPromiseClient());
-                    frameWriter.writeData(ctxClient(), 3, encodedData2, 0, true, newPromiseClient());
+                    clientEncoder.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
+                    clientEncoder.writeData(ctxClient(), 3, encodedData1, 0, false, newPromiseClient());
+                    clientEncoder.writeData(ctxClient(), 3, encodedData2, 0, true, newPromiseClient());
                     ctxClient().flush();
                 }
             });
@@ -227,7 +238,7 @@ public void run() {
             ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             ArgumentCaptor<Boolean> endStreamCaptor = ArgumentCaptor.forClass(Boolean.class);
             verify(serverListener, times(2)).onDataRead(any(ChannelHandlerContext.class), eq(3), dataCaptor.capture(),
-                            eq(0), endStreamCaptor.capture());
+                    eq(0), endStreamCaptor.capture());
             dataCapture = dataCaptor.getAllValues();
             assertEquals(data1, dataCapture.get(0));
             assertEquals(data2, dataCapture.get(1));
@@ -252,16 +263,15 @@ public void deflateEncodingSingleLargeMessageReducedWindow() throws Exception {
                 data.writeByte((byte) 'a');
             }
             final ByteBuf encodedData = encodeData(data, encoder);
-            final Http2Headers headers =
-                    new DefaultHttp2Headers().method(POST).path(PATH)
-                                             .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.DEFLATE);
+            final Http2Headers headers = new DefaultHttp2Headers().method(POST).path(PATH)
+                    .set(HttpHeaderNames.CONTENT_ENCODING, HttpHeaderValues.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());
+                    serverEncoder.writeSettings(ctxServer(), settings, newPromiseServer());
                     ctxServer().flush();
                 }
             });
@@ -273,9 +283,9 @@ public void run() {
             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());
+                    clientEncoder.writeSettings(ctxClient(), settings, newPromiseClient());
+                    clientEncoder.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
+                    clientEncoder.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
                     ctxClient().flush();
                 }
             });
@@ -283,7 +293,7 @@ public void run() {
             data.resetReaderIndex();
             ArgumentCaptor<ByteBuf> dataCaptor = ArgumentCaptor.forClass(ByteBuf.class);
             verify(serverListener).onDataRead(any(ChannelHandlerContext.class), eq(3), dataCaptor.capture(), eq(0),
-                            eq(true));
+                    eq(true));
             dataCapture = dataCaptor.getAllValues();
             assertEquals(data, dataCapture.get(0));
         } finally {
@@ -329,7 +339,6 @@ private void bootstrapEnv(int serverCountDown, int clientCountDown) throws Excep
 
         serverLatch = new CountDownLatch(serverCountDown);
         clientLatch = new CountDownLatch(clientCountDown);
-        frameWriter = new DefaultHttp2FrameWriter();
         serverConnection = new DefaultHttp2Connection(true);
 
         final CountDownLatch latch = new CountDownLatch(1);
@@ -339,9 +348,13 @@ private void bootstrapEnv(int serverCountDown, int clientCountDown) throws Excep
             @Override
             protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
-                serverAdapter = new FrameAdapter(serverConnection,
-                                new DelegatingDecompressorFrameListener(serverConnection, serverListener),
-                                serverLatch);
+                CompressorHttp2ConnectionEncoder.Builder builder = new CompressorHttp2ConnectionEncoder.Builder();
+                Http2FrameWriter writer = new DefaultHttp2FrameWriter();
+                serverEncoder = builder.connection(serverConnection).frameWriter(writer)
+                        .outboundFlow(new DefaultHttp2OutboundFlowController(serverConnection, writer))
+                        .lifecycleManager(serverLifeCycleManager).build();
+                serverAdapter = new FrameAdapter(serverConnection, new DelegatingDecompressorFrameListener(
+                        serverConnection, serverListener), serverLatch);
                 p.addLast("reader", serverAdapter);
                 p.addLast(Http2CodecUtil.ignoreSettingsHandler());
                 serverConnectedChannel = ch;
@@ -355,7 +368,13 @@ protected void initChannel(Channel ch) throws Exception {
             @Override
             protected void initChannel(Channel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
-                clientAdapter = new FrameAdapter(clientListener, clientLatch);
+                Http2Connection connection = new DefaultHttp2Connection(false);
+                Http2FrameWriter writer = new DefaultHttp2FrameWriter();
+                CompressorHttp2ConnectionEncoder.Builder builder = new CompressorHttp2ConnectionEncoder.Builder();
+                clientEncoder = builder.connection(connection).frameWriter(writer)
+                        .outboundFlow(new DefaultHttp2OutboundFlowController(connection, writer))
+                        .lifecycleManager(clientLifeCycleManager).build();
+                clientAdapter = new FrameAdapter(connection, clientListener, clientLatch);
                 p.addLast("reader", clientAdapter);
                 p.addLast(Http2CodecUtil.ignoreSettingsHandler());
             }