From 3807f9fc8ee292eb35ea6c26f075f0185e2f4994 Mon Sep 17 00:00:00 2001 From: Scott Mitchell Date: Wed, 17 Sep 2014 18:20:13 -0400 Subject: [PATCH] HTTP/2 Read Decompression Flow Control Fix Motivation: The current implementation of the HTTP/2 decompression does not integrate with flow control properly. The decompression code is giving the post-decompression size to the flow control algorithm which results in flow control errors at incorrect times. Modifications: -DecompressorHttp2FrameReader.java will need to change where it hooks into the HTTP/2 codec -Enhance unit tests to test this condition Result: No more flow control errors because of decompression design flaw --- .../http2/DecompressorHttp2FrameReader.java | 251 ------------------ .../codec/http2/DefaultHttp2FrameReader.java | 27 +- .../DelegatingDecompressorFrameListener.java | 242 +++++++++++++++++ .../http2/Http2FrameListenerDecorator.java | 105 ++++++++ .../codec/http2/DataCompressionHttp2Test.java | 39 ++- .../http2/client/Http2ClientInitializer.java | 12 +- 6 files changed, 393 insertions(+), 283 deletions(-) delete mode 100644 codec-http2/src/main/java/io/netty/handler/codec/http2/DecompressorHttp2FrameReader.java create mode 100644 codec-http2/src/main/java/io/netty/handler/codec/http2/DelegatingDecompressorFrameListener.java create mode 100644 codec-http2/src/main/java/io/netty/handler/codec/http2/Http2FrameListenerDecorator.java 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 5de289320f..0000000000 --- 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 - * - */ - 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 259cfa5fdc..18f05c4604 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 @@ public class DefaultHttp2FrameReader implements Http2FrameReader { } } - 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 @@ public class DefaultHttp2FrameReader implements Http2FrameReader { } 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 class DefaultHttp2FrameReader implements Http2FrameReader { 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 class DefaultHttp2FrameReader implements Http2FrameReader { 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 0000000000..4c9df927cd --- /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 0000000000..453fc35a5e --- /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/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 e07baabee0..53357f9b37 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 class DataCompressionHttp2Test { @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 class DataCompressionHttp2Test { } @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 class DataCompressionHttp2Test { 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 @@ public class DataCompressionHttp2Test { } } + private void awaitClient() throws Exception { + clientLatch.await(5, SECONDS); + } + private void awaitServer() throws Exception { serverLatch.await(5, SECONDS); } @@ -375,4 +396,12 @@ public class DataCompressionHttp2Test { private ChannelPromise newPromiseClient() { return ctxClient().newPromise(); } + + private ChannelHandlerContext ctxServer() { + return serverConnectedChannel.pipeline().firstContext(); + } + + private ChannelPromise newPromiseServer() { + return ctxServer().newPromise(); + } } 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 71e2efa33e..77041db03e 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.HttpClientCodec; 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 class Http2ClientInitializer extends ChannelInitializer { 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 class Http2ClientInitializer extends ChannelInitializer { } } - 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() {