andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

HTTP/2 Read Decompression Flow Control Fix

Browse Source 
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
This commit is contained in:
Scott Mitchell 2014-09-17 18:20:13 -04:00 committed by Scott Mitchell
parent aba28d4fef
commit 3807f9fc8e
6 changed files with 393 additions and 283 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

251
codec-http2/src/main/java/io/netty/handler/codec/http2/DecompressorHttp2FrameReader.java
View File

@ -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);
}
}

27
codec-http2/src/main/java/io/netty/handler/codec/http2/DefaultHttp2FrameReader.java
View File

@ -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, private void readDataFrame(ChannelHandlerContext ctx, ByteBuf payload,
Http2FrameListener listener) throws Http2Exception { Http2FrameListener listener) throws Http2Exception {
short padding = readPadding(payload); short padding = readPadding(payload);
@ -396,7 +379,7 @@ public class DefaultHttp2FrameReader implements Http2FrameReader {
} }
ByteBuf data = payload.readSlice(dataLength); 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()); payload.skipBytes(payload.readableBytes());
} }
@ -428,8 +411,8 @@ public class DefaultHttp2FrameReader implements Http2FrameReader {
final HeadersBlockBuilder hdrBlockBuilder = headersBlockBuilder(); final HeadersBlockBuilder hdrBlockBuilder = headersBlockBuilder();
hdrBlockBuilder.addFragment(fragment, ctx.alloc(), endOfHeaders); hdrBlockBuilder.addFragment(fragment, ctx.alloc(), endOfHeaders);
if (endOfHeaders) { if (endOfHeaders) {
notifyListenerOnHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(), listener.onHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(),
streamDependency, weight, exclusive, padding, headersFlags.endOfStream(), listener); streamDependency, weight, exclusive, padding, headersFlags.endOfStream());
close(); close();
} }
} }
@ -454,8 +437,8 @@ public class DefaultHttp2FrameReader implements Http2FrameReader {
final HeadersBlockBuilder hdrBlockBuilder = headersBlockBuilder(); final HeadersBlockBuilder hdrBlockBuilder = headersBlockBuilder();
hdrBlockBuilder.addFragment(fragment, ctx.alloc(), endOfHeaders); hdrBlockBuilder.addFragment(fragment, ctx.alloc(), endOfHeaders);
if (endOfHeaders) { if (endOfHeaders) {
notifyListenerOnHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(), padding, listener.onHeadersRead(ctx, headersStreamId, hdrBlockBuilder.headers(), padding,
headersFlags.endOfStream(), listener); headersFlags.endOfStream());
close(); close();
} }
} }

242
codec-http2/src/main/java/io/netty/handler/codec/http2/DelegatingDecompressorFrameListener.java Normal file
View File

@ -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;
}
}
}

105
codec-http2/src/main/java/io/netty/handler/codec/http2/Http2FrameListenerDecorator.java Normal file
View File

@ -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);
}
}

39
codec-http2/src/test/java/io/netty/handler/codec/http2/DataCompressionHttp2Test.java
View File

@ -80,6 +80,7 @@ public class DataCompressionHttp2Test {
private ServerBootstrap sb; private ServerBootstrap sb;
private Bootstrap cb; private Bootstrap cb;
private Channel serverChannel; private Channel serverChannel;
private Channel serverConnectedChannel;
private Channel clientChannel; private Channel clientChannel;
private CountDownLatch serverLatch; private CountDownLatch serverLatch;
private CountDownLatch clientLatch; private CountDownLatch clientLatch;
@ -105,10 +106,12 @@ public class DataCompressionHttp2Test {
@Override @Override
protected void initChannel(Channel ch) throws Exception { protected void initChannel(Channel ch) throws Exception {
ChannelPipeline p = ch.pipeline(); ChannelPipeline p = ch.pipeline();
serverAdapter = new Http2TestUtil.FrameAdapter(serverConnection, new DecompressorHttp2FrameReader( serverAdapter = new Http2TestUtil.FrameAdapter(serverConnection,
serverConnection), serverListener, serverLatch, false); new DelegatingDecompressorFrameListener(serverConnection, serverListener),
serverLatch, false);
p.addLast("reader", serverAdapter); p.addLast("reader", serverAdapter);
p.addLast(Http2CodecUtil.ignoreSettingsHandler()); p.addLast(Http2CodecUtil.ignoreSettingsHandler());
serverConnectedChannel = ch;
} }
}); });
@ -284,9 +287,10 @@ public class DataCompressionHttp2Test {
} }
@Test @Test
public void deflateEncodingSingleLargeMessage() throws Exception { public void deflateEncodingSingleLargeMessageReducedWindow() throws Exception {
serverLatch(new CountDownLatch(2)); serverLatch(new CountDownLatch(3));
final ByteBuf data = Unpooled.buffer(1 << 16); final int BUFFER_SIZE = 1 << 16;
final ByteBuf data = Unpooled.buffer(BUFFER_SIZE);
final EmbeddedChannel encoder = new EmbeddedChannel(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.ZLIB)); final EmbeddedChannel encoder = new EmbeddedChannel(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.ZLIB));
try { try {
for (int i = 0; i < data.capacity(); ++i) { for (int i = 0; i < data.capacity(); ++i) {
@ -296,12 +300,25 @@ public class DataCompressionHttp2Test {
final Http2Headers headers = final Http2Headers headers =
new DefaultHttp2Headers().method(POST).path(PATH) new DefaultHttp2Headers().method(POST).path(PATH)
.set(CONTENT_ENCODING.toLowerCase(), DEFLATE); .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 // Required because the decompressor intercepts the onXXXRead events before
// our {@link Http2TestUtil$FrameAdapter} does. // our {@link Http2TestUtil$FrameAdapter} does.
Http2TestUtil.FrameAdapter.getOrCreateStream(serverConnection, 3, false); Http2TestUtil.FrameAdapter.getOrCreateStream(serverConnection, 3, false);
runInChannel(clientChannel, new Http2Runnable() { runInChannel(clientChannel, new Http2Runnable() {
@Override @Override
public void run() { public void run() {
frameWriter.writeSettings(ctxClient(), settings, newPromiseClient());
frameWriter.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient()); frameWriter.writeHeaders(ctxClient(), 3, headers, 0, false, newPromiseClient());
frameWriter.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient()); frameWriter.writeData(ctxClient(), 3, encodedData, 0, true, newPromiseClient());
ctxClient().flush(); ctxClient().flush();
@ -364,6 +381,10 @@ public class DataCompressionHttp2Test {
} }
} }
private void awaitClient() throws Exception {
clientLatch.await(5, SECONDS);
}
private void awaitServer() throws Exception { private void awaitServer() throws Exception {
serverLatch.await(5, SECONDS); serverLatch.await(5, SECONDS);
} }
@ -375,4 +396,12 @@ public class DataCompressionHttp2Test {
private ChannelPromise newPromiseClient() { private ChannelPromise newPromiseClient() {
return ctxClient().newPromise(); return ctxClient().newPromise();
} }
private ChannelHandlerContext ctxServer() {
return serverConnectedChannel.pipeline().firstContext();
}
private ChannelPromise newPromiseServer() {
return ctxServer().newPromise();
}
} }

12
example/src/main/java/io/netty/example/http2/client/Http2ClientInitializer.java
View File

@ -25,11 +25,12 @@ import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpClientUpgradeHandler; import io.netty.handler.codec.http.HttpClientUpgradeHandler;
import io.netty.handler.codec.http.HttpMethod; import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpVersion; 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.DefaultHttp2Connection;
import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
import io.netty.handler.codec.http2.DefaultHttp2FrameWriter; import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
import io.netty.handler.codec.http2.DefaultHttp2InboundFlowController; import io.netty.handler.codec.http2.DefaultHttp2InboundFlowController;
import io.netty.handler.codec.http2.DefaultHttp2OutboundFlowController; 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.DelegatingHttp2ConnectionHandler;
import io.netty.handler.codec.http2.DelegatingHttp2HttpConnectionHandler; import io.netty.handler.codec.http2.DelegatingHttp2HttpConnectionHandler;
import io.netty.handler.codec.http2.Http2ClientUpgradeCodec; import io.netty.handler.codec.http2.Http2ClientUpgradeCodec;
@ -66,10 +67,11 @@ public class Http2ClientInitializer extends ChannelInitializer<SocketChannel> {
final Http2Connection connection = new DefaultHttp2Connection(false); final Http2Connection connection = new DefaultHttp2Connection(false);
final Http2FrameWriter frameWriter = frameWriter(); final Http2FrameWriter frameWriter = frameWriter();
connectionHandler = new DelegatingHttp2HttpConnectionHandler(connection, connectionHandler = new DelegatingHttp2HttpConnectionHandler(connection,
frameReader(connection), frameWriter, frameReader(), frameWriter,
new DefaultHttp2InboundFlowController(connection, frameWriter), new DefaultHttp2InboundFlowController(connection, frameWriter),
new DefaultHttp2OutboundFlowController(connection, frameWriter), new DefaultHttp2OutboundFlowController(connection, frameWriter),
InboundHttp2ToHttpAdapter.newInstance(connection, maxContentLength)); new DelegatingDecompressorFrameListener(connection,
InboundHttp2ToHttpAdapter.newInstance(connection, maxContentLength)));
responseHandler = new HttpResponseHandler(); responseHandler = new HttpResponseHandler();
settingsHandler = new Http2SettingsHandler(ch.newPromise()); settingsHandler = new Http2SettingsHandler(ch.newPromise());
if (sslCtx != null) { if (sslCtx != null) {
@ -146,8 +148,8 @@ public class Http2ClientInitializer extends ChannelInitializer<SocketChannel> {
} }
} }
private static Http2FrameReader frameReader(Http2Connection connection) { private static Http2FrameReader frameReader() {
return new Http2InboundFrameLogger(new DecompressorHttp2FrameReader(connection), logger); return new Http2InboundFrameLogger(new DefaultHttp2FrameReader(), logger);
} }
private static Http2FrameWriter frameWriter() { private static Http2FrameWriter frameWriter() {