HttpContentDecoder must continue read when it did not produce any mes… (#8922)
Motivation: When HttpContentDecoder (and so HttpContentDecompressor) does not produce any message we need to make sure it calls ctx.read() if auto read is false to not stale. Modifications: - Keep track if we need to call ctx.read() or not - Add unit test Result: Fixes https://github.com/netty/netty/issues/8915.
This commit is contained in:
Norman Maurer
parent
2689f28cb7
commit
10f06a04e6
|
|
@ -51,102 +51,107 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
|
|||
protected ChannelHandlerContext ctx;
|
||||
private EmbeddedChannel decoder;
|
||||
private boolean continueResponse;
|
||||
private boolean needRead = true;
|
||||
|
||||
@Override
|
||||
protected void decode(ChannelHandlerContext ctx, HttpObject msg, List<Object> out) throws Exception {
|
||||
if (msg instanceof HttpResponse && ((HttpResponse) msg).status().code() == 100) {
|
||||
try {
|
||||
if (msg instanceof HttpResponse && ((HttpResponse) msg).status().code() == 100) {
|
||||
|
||||
if (!(msg instanceof LastHttpContent)) {
|
||||
continueResponse = true;
|
||||
}
|
||||
// 100-continue response must be passed through.
|
||||
out.add(ReferenceCountUtil.retain(msg));
|
||||
return;
|
||||
}
|
||||
|
||||
if (continueResponse) {
|
||||
if (msg instanceof LastHttpContent) {
|
||||
continueResponse = false;
|
||||
}
|
||||
// 100-continue response must be passed through.
|
||||
out.add(ReferenceCountUtil.retain(msg));
|
||||
return;
|
||||
}
|
||||
|
||||
if (msg instanceof HttpMessage) {
|
||||
cleanup();
|
||||
final HttpMessage message = (HttpMessage) msg;
|
||||
final HttpHeaders headers = message.headers();
|
||||
|
||||
// Determine the content encoding.
|
||||
String contentEncoding = headers.get(HttpHeaderNames.CONTENT_ENCODING);
|
||||
if (contentEncoding != null) {
|
||||
contentEncoding = contentEncoding.trim();
|
||||
} else {
|
||||
contentEncoding = IDENTITY;
|
||||
}
|
||||
decoder = newContentDecoder(contentEncoding);
|
||||
|
||||
if (decoder == null) {
|
||||
if (message instanceof HttpContent) {
|
||||
((HttpContent) message).retain();
|
||||
if (!(msg instanceof LastHttpContent)) {
|
||||
continueResponse = true;
|
||||
}
|
||||
out.add(message);
|
||||
// 100-continue response must be passed through.
|
||||
out.add(ReferenceCountUtil.retain(msg));
|
||||
return;
|
||||
}
|
||||
|
||||
// Remove content-length header:
|
||||
// the correct value can be set only after all chunks are processed/decoded.
|
||||
// If buffering is not an issue, add HttpObjectAggregator down the chain, it will set the header.
|
||||
// Otherwise, rely on LastHttpContent message.
|
||||
if (headers.contains(HttpHeaderNames.CONTENT_LENGTH)) {
|
||||
headers.remove(HttpHeaderNames.CONTENT_LENGTH);
|
||||
headers.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
|
||||
}
|
||||
// Either it is already chunked or EOF terminated.
|
||||
// See https://github.com/netty/netty/issues/5892
|
||||
|
||||
// set new content encoding,
|
||||
CharSequence targetContentEncoding = getTargetContentEncoding(contentEncoding);
|
||||
if (HttpHeaderValues.IDENTITY.contentEquals(targetContentEncoding)) {
|
||||
// Do NOT set the 'Content-Encoding' header if the target encoding is 'identity'
|
||||
// as per: http://tools.ietf.org/html/rfc2616#section-14.11
|
||||
headers.remove(HttpHeaderNames.CONTENT_ENCODING);
|
||||
} else {
|
||||
headers.set(HttpHeaderNames.CONTENT_ENCODING, targetContentEncoding);
|
||||
}
|
||||
|
||||
if (message instanceof HttpContent) {
|
||||
// If message is a full request or response object (headers + data), don't copy data part into out.
|
||||
// Output headers only; data part will be decoded below.
|
||||
// Note: "copy" object must not be an instance of LastHttpContent class,
|
||||
// as this would (erroneously) indicate the end of the HttpMessage to other handlers.
|
||||
HttpMessage copy;
|
||||
if (message instanceof HttpRequest) {
|
||||
HttpRequest r = (HttpRequest) message; // HttpRequest or FullHttpRequest
|
||||
copy = new DefaultHttpRequest(r.protocolVersion(), r.method(), r.uri());
|
||||
} else if (message instanceof HttpResponse) {
|
||||
HttpResponse r = (HttpResponse) message; // HttpResponse or FullHttpResponse
|
||||
copy = new DefaultHttpResponse(r.protocolVersion(), r.status());
|
||||
} else {
|
||||
throw new CodecException("Object of class " + message.getClass().getName() +
|
||||
" is not a HttpRequest or HttpResponse");
|
||||
if (continueResponse) {
|
||||
if (msg instanceof LastHttpContent) {
|
||||
continueResponse = false;
|
||||
}
|
||||
copy.headers().set(message.headers());
|
||||
copy.setDecoderResult(message.decoderResult());
|
||||
out.add(copy);
|
||||
} else {
|
||||
out.add(message);
|
||||
// 100-continue response must be passed through.
|
||||
out.add(ReferenceCountUtil.retain(msg));
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (msg instanceof HttpContent) {
|
||||
final HttpContent c = (HttpContent) msg;
|
||||
if (decoder == null) {
|
||||
out.add(c.retain());
|
||||
} else {
|
||||
decodeContent(c, out);
|
||||
if (msg instanceof HttpMessage) {
|
||||
cleanup();
|
||||
final HttpMessage message = (HttpMessage) msg;
|
||||
final HttpHeaders headers = message.headers();
|
||||
|
||||
// Determine the content encoding.
|
||||
String contentEncoding = headers.get(HttpHeaderNames.CONTENT_ENCODING);
|
||||
if (contentEncoding != null) {
|
||||
contentEncoding = contentEncoding.trim();
|
||||
} else {
|
||||
contentEncoding = IDENTITY;
|
||||
}
|
||||
decoder = newContentDecoder(contentEncoding);
|
||||
|
||||
if (decoder == null) {
|
||||
if (message instanceof HttpContent) {
|
||||
((HttpContent) message).retain();
|
||||
}
|
||||
out.add(message);
|
||||
return;
|
||||
}
|
||||
|
||||
// Remove content-length header:
|
||||
// the correct value can be set only after all chunks are processed/decoded.
|
||||
// If buffering is not an issue, add HttpObjectAggregator down the chain, it will set the header.
|
||||
// Otherwise, rely on LastHttpContent message.
|
||||
if (headers.contains(HttpHeaderNames.CONTENT_LENGTH)) {
|
||||
headers.remove(HttpHeaderNames.CONTENT_LENGTH);
|
||||
headers.set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
|
||||
}
|
||||
// Either it is already chunked or EOF terminated.
|
||||
// See https://github.com/netty/netty/issues/5892
|
||||
|
||||
// set new content encoding,
|
||||
CharSequence targetContentEncoding = getTargetContentEncoding(contentEncoding);
|
||||
if (HttpHeaderValues.IDENTITY.contentEquals(targetContentEncoding)) {
|
||||
// Do NOT set the 'Content-Encoding' header if the target encoding is 'identity'
|
||||
// as per: http://tools.ietf.org/html/rfc2616#section-14.11
|
||||
headers.remove(HttpHeaderNames.CONTENT_ENCODING);
|
||||
} else {
|
||||
headers.set(HttpHeaderNames.CONTENT_ENCODING, targetContentEncoding);
|
||||
}
|
||||
|
||||
if (message instanceof HttpContent) {
|
||||
// If message is a full request or response object (headers + data), don't copy data part into out.
|
||||
// Output headers only; data part will be decoded below.
|
||||
// Note: "copy" object must not be an instance of LastHttpContent class,
|
||||
// as this would (erroneously) indicate the end of the HttpMessage to other handlers.
|
||||
HttpMessage copy;
|
||||
if (message instanceof HttpRequest) {
|
||||
HttpRequest r = (HttpRequest) message; // HttpRequest or FullHttpRequest
|
||||
copy = new DefaultHttpRequest(r.protocolVersion(), r.method(), r.uri());
|
||||
} else if (message instanceof HttpResponse) {
|
||||
HttpResponse r = (HttpResponse) message; // HttpResponse or FullHttpResponse
|
||||
copy = new DefaultHttpResponse(r.protocolVersion(), r.status());
|
||||
} else {
|
||||
throw new CodecException("Object of class " + message.getClass().getName() +
|
||||
" is not a HttpRequest or HttpResponse");
|
||||
}
|
||||
copy.headers().set(message.headers());
|
||||
copy.setDecoderResult(message.decoderResult());
|
||||
out.add(copy);
|
||||
} else {
|
||||
out.add(message);
|
||||
}
|
||||
}
|
||||
|
||||
if (msg instanceof HttpContent) {
|
||||
final HttpContent c = (HttpContent) msg;
|
||||
if (decoder == null) {
|
||||
out.add(c.retain());
|
||||
} else {
|
||||
decodeContent(c, out);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
needRead = out.isEmpty();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -170,6 +175,20 @@ public abstract class HttpContentDecoder extends MessageToMessageDecoder<HttpObj
|
|||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
||||
boolean needRead = this.needRead;
|
||||
this.needRead = true;
|
||||
|
||||
try {
|
||||
ctx.fireChannelReadComplete();
|
||||
} finally {
|
||||
if (needRead && !ctx.channel().config().isAutoRead()) {
|
||||
ctx.read();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a new {@link EmbeddedChannel} that decodes the HTTP message
|
||||
* content encoded in the specified <tt>contentEncoding</tt>.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,70 @@
|
|||
/*
|
||||
* Copyright 2019 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.http;
|
||||
|
||||
import io.netty.buffer.Unpooled;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelInboundHandlerAdapter;
|
||||
import io.netty.channel.ChannelOutboundHandlerAdapter;
|
||||
import io.netty.channel.embedded.EmbeddedChannel;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
public class HttpContentDecompressorTest {
|
||||
|
||||
// See https://github.com/netty/netty/issues/8915.
|
||||
@Test
|
||||
public void testInvokeReadWhenNotProduceMessage() {
|
||||
final AtomicInteger readCalled = new AtomicInteger();
|
||||
EmbeddedChannel channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter() {
|
||||
@Override
|
||||
public void read(ChannelHandlerContext ctx) {
|
||||
readCalled.incrementAndGet();
|
||||
ctx.read();
|
||||
}
|
||||
}, new HttpContentDecompressor(), new ChannelInboundHandlerAdapter() {
|
||||
@Override
|
||||
public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
||||
ctx.fireChannelRead(msg);
|
||||
ctx.read();
|
||||
}
|
||||
});
|
||||
|
||||
channel.config().setAutoRead(false);
|
||||
|
||||
readCalled.set(0);
|
||||
HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
|
||||
response.headers().set(HttpHeaderNames.CONTENT_ENCODING, "gzip");
|
||||
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json;charset=UTF-8");
|
||||
response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
|
||||
|
||||
Assert.assertTrue(channel.writeInbound(response));
|
||||
|
||||
// we triggered read explicitly
|
||||
Assert.assertEquals(1, readCalled.get());
|
||||
|
||||
Assert.assertTrue(channel.readInbound() instanceof HttpResponse);
|
||||
|
||||
Assert.assertFalse(channel.writeInbound(new DefaultHttpContent(Unpooled.EMPTY_BUFFER)));
|
||||
|
||||
// read was triggered by the HttpContentDecompressor itself as it did not produce any message to the next
|
||||
// inbound handler.
|
||||
Assert.assertEquals(2, readCalled.get());
|
||||
Assert.assertFalse(channel.finishAndReleaseAll());
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user