mirror of https://github.com/grpc/grpc-java.git
nmittler
parent
caad0294b9
commit
5cef321b78
|
|
@ -1,338 +0,0 @@
|
|||
/*
|
||||
* Copyright 2015, Google Inc. All rights reserved.
|
||||
*
|
||||
* Redistribution and use in source and binary forms, with or without
|
||||
* modification, are permitted provided that the following conditions are
|
||||
* met:
|
||||
*
|
||||
* * Redistributions of source code must retain the above copyright
|
||||
* notice, this list of conditions and the following disclaimer.
|
||||
* * Redistributions in binary form must reproduce the above
|
||||
* copyright notice, this list of conditions and the following disclaimer
|
||||
* in the documentation and/or other materials provided with the
|
||||
* distribution.
|
||||
*
|
||||
* * Neither the name of Google Inc. nor the names of its
|
||||
* contributors may be used to endorse or promote products derived from
|
||||
* this software without specific prior written permission.
|
||||
*
|
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
*/
|
||||
|
||||
package io.grpc.netty;
|
||||
|
||||
import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
|
||||
import static io.netty.handler.codec.http2.Http2Exception.connectionError;
|
||||
|
||||
import io.netty.buffer.ByteBuf;
|
||||
import io.netty.buffer.ByteBufUtil;
|
||||
import io.netty.channel.ChannelFuture;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.channel.ChannelPromise;
|
||||
import io.netty.handler.codec.http2.DecoratingHttp2ConnectionEncoder;
|
||||
import io.netty.handler.codec.http2.Http2CodecUtil;
|
||||
import io.netty.handler.codec.http2.Http2Connection;
|
||||
import io.netty.handler.codec.http2.Http2ConnectionAdapter;
|
||||
import io.netty.handler.codec.http2.Http2ConnectionEncoder;
|
||||
import io.netty.handler.codec.http2.Http2Headers;
|
||||
import io.netty.handler.codec.http2.Http2Stream;
|
||||
import io.netty.util.ReferenceCountUtil;
|
||||
|
||||
import java.util.ArrayDeque;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
import java.util.Queue;
|
||||
import java.util.TreeMap;
|
||||
|
||||
/**
|
||||
* Implementation of a {@link Http2ConnectionEncoder} that dispatches all method call to
|
||||
* another {@link Http2ConnectionEncoder}, except for when the maximum number of
|
||||
* concurrent streams limit is reached.
|
||||
*
|
||||
* <p>When this limit is hit, instead of rejecting any new streams this implementation
|
||||
* buffers newly created streams and their corresponding frames. Once an active stream
|
||||
* gets closed or the maximum number of concurrent streams is increased, this encoder will
|
||||
* automatically try to empty its buffer and create as many new streams as possible.
|
||||
*
|
||||
* <p>This implementation makes the buffering mostly transparent and can be used as a drop
|
||||
* in replacement for {@link io.netty.handler.codec.http2.DefaultHttp2ConnectionEncoder}.
|
||||
*/
|
||||
class BufferingHttp2ConnectionEncoder extends DecoratingHttp2ConnectionEncoder {
|
||||
/**
|
||||
* The assumed minimum value for {@code SETTINGS_MAX_CONCURRENT_STREAMS} as
|
||||
* recommended by the HTTP/2 spec.
|
||||
*/
|
||||
static final int SMALLEST_MAX_CONCURRENT_STREAMS = 100;
|
||||
|
||||
/**
|
||||
* Buffer for any streams and corresponding frames that could not be created
|
||||
* due to the maximum concurrent stream limit being hit.
|
||||
*/
|
||||
private final TreeMap<Integer, PendingStream> pendingStreams =
|
||||
new TreeMap<Integer, PendingStream>();
|
||||
private final int initialMaxConcurrentStreams;
|
||||
// Smallest stream id whose corresponding frames do not get buffered.
|
||||
private int largestCreatedStreamId;
|
||||
private boolean receivedSettings;
|
||||
|
||||
protected BufferingHttp2ConnectionEncoder(Http2ConnectionEncoder delegate) {
|
||||
this(delegate, SMALLEST_MAX_CONCURRENT_STREAMS);
|
||||
}
|
||||
|
||||
protected BufferingHttp2ConnectionEncoder(Http2ConnectionEncoder delegate,
|
||||
int initialMaxConcurrentStreams) {
|
||||
super(delegate);
|
||||
this.initialMaxConcurrentStreams = initialMaxConcurrentStreams;
|
||||
connection().addListener(new Http2ConnectionAdapter() {
|
||||
|
||||
@Override
|
||||
public void onGoAwayReceived(int lastStreamId, long errorCode, ByteBuf debugData) {
|
||||
cancelGoAwayStreams(lastStreamId, errorCode, debugData);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onStreamClosed(Http2Stream stream) {
|
||||
tryCreatePendingStreams();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Indicates the number of streams that are currently buffered, awaiting creation.
|
||||
*/
|
||||
public int numBufferedStreams() {
|
||||
return pendingStreams.size();
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
|
||||
int padding, boolean endStream, ChannelPromise promise) {
|
||||
return writeHeaders(ctx, streamId, headers, 0, Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT, false,
|
||||
padding, endStream, promise);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChannelFuture writeHeaders(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
|
||||
int streamDependency, short weight, boolean exclusive,
|
||||
int padding, boolean endOfStream, ChannelPromise promise) {
|
||||
if (existingStream(streamId) || connection().goAwayReceived()) {
|
||||
return super.writeHeaders(ctx, streamId, headers, streamDependency, weight,
|
||||
exclusive, padding, endOfStream, promise);
|
||||
}
|
||||
if (canCreateStream()) {
|
||||
assert streamId > largestCreatedStreamId;
|
||||
largestCreatedStreamId = streamId;
|
||||
return super.writeHeaders(ctx, streamId, headers, streamDependency, weight,
|
||||
exclusive, padding, endOfStream, promise);
|
||||
}
|
||||
PendingStream pendingStream = pendingStreams.get(streamId);
|
||||
if (pendingStream == null) {
|
||||
pendingStream = new PendingStream(ctx, streamId);
|
||||
pendingStreams.put(streamId, pendingStream);
|
||||
}
|
||||
pendingStream.frames.add(new HeadersFrame(headers, streamDependency, weight, exclusive,
|
||||
padding, endOfStream, promise));
|
||||
return promise;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChannelFuture writeRstStream(ChannelHandlerContext ctx, int streamId, long errorCode,
|
||||
ChannelPromise promise) {
|
||||
if (existingStream(streamId)) {
|
||||
return super.writeRstStream(ctx, streamId, errorCode, promise);
|
||||
}
|
||||
// Since the delegate doesn't know about any buffered streams we have to handle cancellation
|
||||
// of the promises and releasing of the ByteBufs here.
|
||||
PendingStream stream = pendingStreams.remove(streamId);
|
||||
if (stream != null) {
|
||||
// Sending a RST_STREAM to a buffered stream will succeed the promise of all frames
|
||||
// associated with the stream, as sending a RST_STREAM means that someone "doesn't care"
|
||||
// about the stream anymore and thus there is not point in failing the promises and invoking
|
||||
// error handling routines.
|
||||
stream.close(null);
|
||||
promise.setSuccess();
|
||||
} else {
|
||||
promise.setFailure(connectionError(PROTOCOL_ERROR, "Stream does not exist %d", streamId));
|
||||
}
|
||||
return promise;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChannelFuture writeData(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
|
||||
boolean endOfStream, ChannelPromise promise) {
|
||||
if (existingStream(streamId)) {
|
||||
return super.writeData(ctx, streamId, data, padding, endOfStream, promise);
|
||||
}
|
||||
PendingStream pendingStream = pendingStreams.get(streamId);
|
||||
if (pendingStream != null) {
|
||||
pendingStream.frames.add(new DataFrame(data, padding, endOfStream, promise));
|
||||
} else {
|
||||
ReferenceCountUtil.safeRelease(data);
|
||||
promise.setFailure(connectionError(PROTOCOL_ERROR, "Stream does not exist %d", streamId));
|
||||
}
|
||||
return promise;
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChannelFuture writeSettingsAck(ChannelHandlerContext ctx, ChannelPromise promise) {
|
||||
receivedSettings = true;
|
||||
ChannelFuture future = super.writeSettingsAck(ctx, promise);
|
||||
// After having received a SETTINGS frame, the maximum number of concurrent streams
|
||||
// might have changed. So try to create some buffered streams.
|
||||
tryCreatePendingStreams();
|
||||
return future;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
super.close();
|
||||
cancelPendingStreams();
|
||||
}
|
||||
|
||||
private void tryCreatePendingStreams() {
|
||||
while (!pendingStreams.isEmpty() && canCreateStream()) {
|
||||
Map.Entry<Integer, PendingStream> entry = pendingStreams.pollFirstEntry();
|
||||
PendingStream pendingStream = entry.getValue();
|
||||
pendingStream.sendFrames();
|
||||
largestCreatedStreamId = pendingStream.streamId;
|
||||
}
|
||||
}
|
||||
|
||||
private void cancelPendingStreams() {
|
||||
Exception e = new Exception("Connection closed.");
|
||||
while (!pendingStreams.isEmpty()) {
|
||||
PendingStream stream = pendingStreams.pollFirstEntry().getValue();
|
||||
stream.close(e);
|
||||
}
|
||||
}
|
||||
|
||||
private void cancelGoAwayStreams(int lastStreamId, long errorCode, ByteBuf debugData) {
|
||||
Iterator<PendingStream> iter = pendingStreams.values().iterator();
|
||||
Exception e = new GoAwayClosedStreamException(lastStreamId, errorCode,
|
||||
ByteBufUtil.getBytes(debugData));
|
||||
while (iter.hasNext()) {
|
||||
PendingStream stream = iter.next();
|
||||
if (stream.streamId > lastStreamId) {
|
||||
iter.remove();
|
||||
stream.close(e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines whether or not we're allowed to create a new stream right now.
|
||||
*/
|
||||
private boolean canCreateStream() {
|
||||
Http2Connection.Endpoint<?> local = connection().local();
|
||||
return (receivedSettings || local.numActiveStreams() < initialMaxConcurrentStreams)
|
||||
&& local.canOpenStream() && !local.isExhausted();
|
||||
}
|
||||
|
||||
private boolean existingStream(int streamId) {
|
||||
return streamId <= largestCreatedStreamId;
|
||||
}
|
||||
|
||||
private static class PendingStream {
|
||||
final ChannelHandlerContext ctx;
|
||||
final int streamId;
|
||||
final Queue<Frame> frames = new ArrayDeque<Frame>(2);
|
||||
|
||||
PendingStream(ChannelHandlerContext ctx, int streamId) {
|
||||
this.ctx = ctx;
|
||||
this.streamId = streamId;
|
||||
}
|
||||
|
||||
void sendFrames() {
|
||||
for (Frame frame : frames) {
|
||||
frame.send(ctx, streamId);
|
||||
}
|
||||
}
|
||||
|
||||
void close(Throwable t) {
|
||||
for (Frame frame : frames) {
|
||||
frame.release(t);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private abstract static class Frame {
|
||||
final ChannelPromise promise;
|
||||
|
||||
Frame(ChannelPromise promise) {
|
||||
this.promise = promise;
|
||||
}
|
||||
|
||||
/**
|
||||
* Release any resources (features, buffers, ...) associated with the frame.
|
||||
*/
|
||||
void release(Throwable t) {
|
||||
if (t == null) {
|
||||
promise.setSuccess();
|
||||
} else {
|
||||
promise.setFailure(t);
|
||||
}
|
||||
}
|
||||
|
||||
abstract void send(ChannelHandlerContext ctx, int streamId);
|
||||
}
|
||||
|
||||
private class HeadersFrame extends Frame {
|
||||
final Http2Headers headers;
|
||||
final int streamDependency;
|
||||
final short weight;
|
||||
final boolean exclusive;
|
||||
final int padding;
|
||||
final boolean endOfStream;
|
||||
|
||||
HeadersFrame(Http2Headers headers, int streamDependency, short weight, boolean exclusive,
|
||||
int padding, boolean endOfStream, ChannelPromise promise) {
|
||||
super(promise);
|
||||
this.headers = headers;
|
||||
this.streamDependency = streamDependency;
|
||||
this.weight = weight;
|
||||
this.exclusive = exclusive;
|
||||
this.padding = padding;
|
||||
this.endOfStream = endOfStream;
|
||||
}
|
||||
|
||||
@Override
|
||||
void send(ChannelHandlerContext ctx, int streamId) {
|
||||
writeHeaders(ctx, streamId, headers, streamDependency, weight, exclusive, padding,
|
||||
endOfStream, promise);
|
||||
}
|
||||
}
|
||||
|
||||
private class DataFrame extends Frame {
|
||||
final ByteBuf data;
|
||||
final int padding;
|
||||
final boolean endOfStream;
|
||||
|
||||
DataFrame(ByteBuf data, int padding, boolean endOfStream, ChannelPromise promise) {
|
||||
super(promise);
|
||||
this.data = data;
|
||||
this.padding = padding;
|
||||
this.endOfStream = endOfStream;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void release(Throwable t) {
|
||||
super.release(t);
|
||||
ReferenceCountUtil.safeRelease(data);
|
||||
}
|
||||
|
||||
@Override
|
||||
void send(ChannelHandlerContext ctx, int streamId) {
|
||||
writeData(ctx, streamId, data, padding, endOfStream, promise);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
/*
|
||||
* Copyright 2015, Google Inc. All rights reserved.
|
||||
*
|
||||
* Redistribution and use in source and binary forms, with or without
|
||||
* modification, are permitted provided that the following conditions are
|
||||
* met:
|
||||
*
|
||||
* * Redistributions of source code must retain the above copyright
|
||||
* notice, this list of conditions and the following disclaimer.
|
||||
* * Redistributions in binary form must reproduce the above
|
||||
* copyright notice, this list of conditions and the following disclaimer
|
||||
* in the documentation and/or other materials provided with the
|
||||
* distribution.
|
||||
*
|
||||
* * Neither the name of Google Inc. nor the names of its
|
||||
* contributors may be used to endorse or promote products derived from
|
||||
* this software without specific prior written permission.
|
||||
*
|
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
*/
|
||||
|
||||
package io.grpc.netty;
|
||||
|
||||
class GoAwayClosedStreamException extends Exception {
|
||||
private static final long serialVersionUID = 1326785622777291198L;
|
||||
private final int lastStreamId;
|
||||
private final long errorCode;
|
||||
private final byte[] debugData;
|
||||
|
||||
GoAwayClosedStreamException(int lastStreamId, long errorCode, byte[] debugData) {
|
||||
this.lastStreamId = lastStreamId;
|
||||
this.errorCode = errorCode;
|
||||
this.debugData = debugData;
|
||||
}
|
||||
|
||||
int lastStreamId() {
|
||||
return lastStreamId;
|
||||
}
|
||||
|
||||
long errorCode() {
|
||||
return errorCode;
|
||||
}
|
||||
|
||||
byte[] debugData() {
|
||||
return debugData;
|
||||
}
|
||||
}
|
||||
|
|
@ -78,6 +78,7 @@ import io.netty.handler.codec.http2.Http2OutboundFrameLogger;
|
|||
import io.netty.handler.codec.http2.Http2Settings;
|
||||
import io.netty.handler.codec.http2.Http2Stream;
|
||||
import io.netty.handler.codec.http2.Http2StreamVisitor;
|
||||
import io.netty.handler.codec.http2.StreamBufferingEncoder;
|
||||
import io.netty.handler.logging.LogLevel;
|
||||
|
||||
import java.util.Random;
|
||||
|
|
@ -144,7 +145,7 @@ class NettyClientHandler extends AbstractNettyHandler {
|
|||
frameReader = new Http2InboundFrameLogger(frameReader, frameLogger);
|
||||
frameWriter = new Http2OutboundFrameLogger(frameWriter, frameLogger);
|
||||
|
||||
BufferingHttp2ConnectionEncoder encoder = new BufferingHttp2ConnectionEncoder(
|
||||
StreamBufferingEncoder encoder = new StreamBufferingEncoder(
|
||||
new DefaultHttp2ConnectionEncoder(connection, frameWriter)) {
|
||||
private boolean firstSettings = true;
|
||||
|
||||
|
|
@ -174,7 +175,7 @@ class NettyClientHandler extends AbstractNettyHandler {
|
|||
}
|
||||
|
||||
private NettyClientHandler(Http2ConnectionDecoder decoder,
|
||||
BufferingHttp2ConnectionEncoder encoder, Http2Settings settings,
|
||||
StreamBufferingEncoder encoder, Http2Settings settings,
|
||||
Ticker ticker) {
|
||||
super(decoder, encoder, settings);
|
||||
this.ticker = ticker;
|
||||
|
|
@ -323,7 +324,7 @@ class NettyClientHandler extends AbstractNettyHandler {
|
|||
protected boolean isGracefulShutdownComplete() {
|
||||
// Only allow graceful shutdown to complete after all pending streams have completed.
|
||||
return super.isGracefulShutdownComplete()
|
||||
&& ((BufferingHttp2ConnectionEncoder) encoder()).numBufferedStreams() == 0;
|
||||
&& ((StreamBufferingEncoder) encoder()).numBufferedStreams() == 0;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -385,8 +386,9 @@ class NettyClientHandler extends AbstractNettyHandler {
|
|||
promise.setSuccess();
|
||||
} else {
|
||||
final Throwable cause = future.cause();
|
||||
if (cause instanceof GoAwayClosedStreamException) {
|
||||
GoAwayClosedStreamException e = (GoAwayClosedStreamException) cause;
|
||||
if (cause instanceof StreamBufferingEncoder.Http2GoAwayException) {
|
||||
StreamBufferingEncoder.Http2GoAwayException e =
|
||||
(StreamBufferingEncoder.Http2GoAwayException) cause;
|
||||
goAwayStatus(statusFromGoAway(e.errorCode(), e.debugData()));
|
||||
promise.setFailure(goAwayStatusThrowable);
|
||||
} else {
|
||||
|
|
|
|||
|
|
@ -1,446 +0,0 @@
|
|||
/*
|
||||
* Copyright 2015, Google Inc. All rights reserved.
|
||||
*
|
||||
* Redistribution and use in source and binary forms, with or without
|
||||
* modification, are permitted provided that the following conditions are
|
||||
* met:
|
||||
*
|
||||
* * Redistributions of source code must retain the above copyright
|
||||
* notice, this list of conditions and the following disclaimer.
|
||||
* * Redistributions in binary form must reproduce the above
|
||||
* copyright notice, this list of conditions and the following disclaimer
|
||||
* in the documentation and/or other materials provided with the
|
||||
* distribution.
|
||||
*
|
||||
* * Neither the name of Google Inc. nor the names of its
|
||||
* contributors may be used to endorse or promote products derived from
|
||||
* this software without specific prior written permission.
|
||||
*
|
||||
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
||||
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
||||
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
||||
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
||||
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
||||
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
||||
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
||||
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
||||
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
||||
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
||||
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
||||
*/
|
||||
|
||||
package io.grpc.netty;
|
||||
|
||||
import static io.grpc.internal.GrpcUtil.Http2Error.CANCEL;
|
||||
import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
|
||||
import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_LOCAL;
|
||||
import static io.netty.util.CharsetUtil.UTF_8;
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertFalse;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Matchers.eq;
|
||||
import static org.mockito.Mockito.atLeastOnce;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.spy;
|
||||
import static org.mockito.Mockito.times;
|
||||
import static org.mockito.Mockito.verify;
|
||||
|
||||
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.embedded.EmbeddedChannel;
|
||||
import io.netty.handler.codec.http2.DefaultHttp2Connection;
|
||||
import io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder;
|
||||
import io.netty.handler.codec.http2.DefaultHttp2ConnectionEncoder;
|
||||
import io.netty.handler.codec.http2.DefaultHttp2FrameReader;
|
||||
import io.netty.handler.codec.http2.DefaultHttp2FrameWriter;
|
||||
import io.netty.handler.codec.http2.DefaultHttp2Headers;
|
||||
import io.netty.handler.codec.http2.Http2Connection;
|
||||
import io.netty.handler.codec.http2.Http2ConnectionHandler;
|
||||
import io.netty.handler.codec.http2.Http2ConnectionHandlerBuilder;
|
||||
import io.netty.handler.codec.http2.Http2Exception;
|
||||
import io.netty.handler.codec.http2.Http2FrameListener;
|
||||
import io.netty.handler.codec.http2.Http2FrameWriter;
|
||||
import io.netty.handler.codec.http2.Http2Headers;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.runners.JUnit4;
|
||||
import org.mockito.Mock;
|
||||
import org.mockito.MockitoAnnotations;
|
||||
import org.mockito.verification.VerificationMode;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Tests for {@link BufferingHttp2ConnectionEncoder}.
|
||||
*/
|
||||
@RunWith(JUnit4.class)
|
||||
public class BufferingHttp2ConnectionEncoderTest {
|
||||
private static final byte[] DEBUG_DATA = "test exception".getBytes(UTF_8);
|
||||
|
||||
private BufferingHttp2ConnectionEncoder encoder;
|
||||
|
||||
private Http2Connection connection;
|
||||
|
||||
private Http2FrameWriter writer;
|
||||
|
||||
@Mock
|
||||
private Http2FrameListener frameListener;
|
||||
|
||||
private ChannelHandlerContext ctx;
|
||||
|
||||
private EmbeddedChannel channel;
|
||||
|
||||
/**
|
||||
* Init fields and do mocking.
|
||||
*/
|
||||
@Before
|
||||
public void setup() throws Exception {
|
||||
MockitoAnnotations.initMocks(this);
|
||||
|
||||
connection = new DefaultHttp2Connection(false);
|
||||
|
||||
writer = spy(new DefaultHttp2FrameWriter());
|
||||
DefaultHttp2ConnectionEncoder defaultEncoder =
|
||||
new DefaultHttp2ConnectionEncoder(connection, writer);
|
||||
encoder = new BufferingHttp2ConnectionEncoder(defaultEncoder);
|
||||
DefaultHttp2ConnectionDecoder decoder = new DefaultHttp2ConnectionDecoder(connection, encoder,
|
||||
new DefaultHttp2FrameReader());
|
||||
decoder.frameListener(frameListener);
|
||||
|
||||
Http2ConnectionHandler handler =
|
||||
new Http2ConnectionHandlerBuilder().codec(decoder, encoder).build();
|
||||
channel = new EmbeddedChannel(handler);
|
||||
ctx = channel.pipeline().context(handler);
|
||||
}
|
||||
|
||||
@After
|
||||
public void teardown() {
|
||||
channel.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void multipleWritesToActiveStream() throws Http2Exception {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
encoderWriteHeaders(3);
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
encoder.writeData(ctx, 3, data(), 0, false, newPromise());
|
||||
encoder.writeData(ctx, 3, data(), 0, false, newPromise());
|
||||
encoder.writeData(ctx, 3, data(), 0, false, newPromise());
|
||||
encoderWriteHeaders(3);
|
||||
|
||||
flush();
|
||||
|
||||
writeVerifyWriteHeaders(times(2), 3);
|
||||
verify(writer, atLeastOnce()).writeData(eq(ctx), eq(3), any(ByteBuf.class), eq(0), eq(false),
|
||||
any(ChannelPromise.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void ensureCanCreateNextStreamWhenStreamCloses() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(1);
|
||||
|
||||
encoderWriteHeaders(3);
|
||||
flush();
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
|
||||
// This one gets buffered.
|
||||
encoderWriteHeaders(5);
|
||||
flush();
|
||||
assertEquals(1, connection.numActiveStreams());
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
|
||||
// Now prevent us from creating another stream.
|
||||
connection.local().maxActiveStreams(0);
|
||||
|
||||
// Close the previous stream.
|
||||
connection.stream(3).close();
|
||||
|
||||
flush();
|
||||
|
||||
// Ensure that no streams are currently active and that only the HEADERS from the first
|
||||
// stream were written.
|
||||
writeVerifyWriteHeaders(times(1), 3);
|
||||
writeVerifyWriteHeaders(never(), 5);
|
||||
assertEquals(0, connection.numActiveStreams());
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void alternatingWritesToActiveAndBufferedStreams() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(1);
|
||||
|
||||
encoderWriteHeaders(3);
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
|
||||
encoderWriteHeaders(5);
|
||||
assertEquals(1, connection.numActiveStreams());
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
|
||||
encoder.writeData(ctx, 3, Unpooled.buffer(0), 0, false, newPromise());
|
||||
flush();
|
||||
writeVerifyWriteHeaders(times(1), 3);
|
||||
encoder.writeData(ctx, 5, Unpooled.buffer(0), 0, false, newPromise());
|
||||
flush();
|
||||
verify(writer, never()).writeData(eq(ctx), eq(5), any(ByteBuf.class), eq(0), eq(false),
|
||||
any(ChannelPromise.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void bufferingNewStreamFailsAfterGoAwayReceived() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(0);
|
||||
connection.goAwayReceived(1, 8, Unpooled.wrappedBuffer(DEBUG_DATA));
|
||||
|
||||
ChannelFuture future = encoderWriteHeaders(3);
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
assertTrue(future.isDone());
|
||||
assertFalse(future.isSuccess());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void receivingGoAwayFailsBufferedStreams() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
|
||||
int maxStreams = 5;
|
||||
connection.local().maxActiveStreams(maxStreams);
|
||||
|
||||
List<ChannelFuture> futures = new ArrayList<ChannelFuture>();
|
||||
int streamId = 3;
|
||||
for (int i = 0; i < 9; i++) {
|
||||
futures.add(encoderWriteHeaders(streamId));
|
||||
streamId += 2;
|
||||
}
|
||||
flush();
|
||||
assertEquals(4, encoder.numBufferedStreams());
|
||||
|
||||
connection.goAwayReceived(11, 8, Unpooled.wrappedBuffer(DEBUG_DATA));
|
||||
|
||||
assertEquals(5, connection.numActiveStreams());
|
||||
// The 4 buffered streams must have been failed.
|
||||
|
||||
int failed = 0;
|
||||
for (ChannelFuture future : futures) {
|
||||
if (future.isDone() && !future.isSuccess()) {
|
||||
failed++;
|
||||
}
|
||||
}
|
||||
assertEquals(4, failed);
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void sendingGoAwayShouldNotFailStreams() throws Exception {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(1);
|
||||
|
||||
List<ChannelFuture> futures = new ArrayList<ChannelFuture>();
|
||||
futures.add(encoderWriteHeaders(3));
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
futures.add(encoderWriteHeaders(5));
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
futures.add(encoderWriteHeaders(7));
|
||||
assertEquals(2, encoder.numBufferedStreams());
|
||||
|
||||
ByteBuf empty = Unpooled.buffer(0);
|
||||
encoder.writeGoAway(ctx, 3, CANCEL.code(), empty, newPromise());
|
||||
|
||||
assertEquals(1, connection.numActiveStreams());
|
||||
assertEquals(2, encoder.numBufferedStreams());
|
||||
|
||||
for (ChannelFuture future : futures) {
|
||||
assertNull(future.cause());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void endStreamDoesNotFailBufferedStream() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(0);
|
||||
|
||||
encoderWriteHeaders(3);
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
|
||||
ByteBuf empty = Unpooled.buffer(0);
|
||||
encoder.writeData(ctx, 3, empty, 0, true, newPromise());
|
||||
flush();
|
||||
|
||||
assertEquals(0, connection.numActiveStreams());
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
|
||||
// Simulate that we received a SETTINGS frame which
|
||||
// increased MAX_CONCURRENT_STREAMS to 1.
|
||||
connection.local().maxActiveStreams(1);
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
flush();
|
||||
|
||||
assertEquals(1, connection.numActiveStreams());
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
assertEquals(HALF_CLOSED_LOCAL, connection.stream(3).state());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void rstStreamClosesBufferedStream() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(0);
|
||||
|
||||
ChannelFuture future = encoderWriteHeaders(3);
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
|
||||
assertFalse(future.isDone());
|
||||
ChannelPromise rstStreamPromise = mock(ChannelPromise.class);
|
||||
encoder.writeRstStream(ctx, 3, CANCEL.code(), rstStreamPromise);
|
||||
assertTrue(future.isSuccess());
|
||||
verify(rstStreamPromise).setSuccess();
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void bufferUntilActiveStreamsAreReset() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(1);
|
||||
|
||||
encoderWriteHeaders(3);
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
encoderWriteHeaders(5);
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
encoderWriteHeaders(7);
|
||||
assertEquals(2, encoder.numBufferedStreams());
|
||||
|
||||
flush();
|
||||
|
||||
writeVerifyWriteHeaders(times(1), 3);
|
||||
writeVerifyWriteHeaders(never(), 5);
|
||||
writeVerifyWriteHeaders(never(), 7);
|
||||
|
||||
encoder.writeRstStream(ctx, 3, CANCEL.code(), newPromise());
|
||||
flush();
|
||||
assertEquals(1, connection.numActiveStreams());
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
|
||||
encoder.writeRstStream(ctx, 5, CANCEL.code(), newPromise());
|
||||
flush();
|
||||
assertEquals(1, connection.numActiveStreams());
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
|
||||
encoder.writeRstStream(ctx, 7, CANCEL.code(), newPromise());
|
||||
flush();
|
||||
assertEquals(0, connection.numActiveStreams());
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void bufferUntilMaxStreamsIncreased() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(2);
|
||||
|
||||
encoderWriteHeaders(3);
|
||||
encoderWriteHeaders(5);
|
||||
encoderWriteHeaders(7);
|
||||
encoderWriteHeaders(9);
|
||||
assertEquals(2, encoder.numBufferedStreams());
|
||||
flush();
|
||||
|
||||
writeVerifyWriteHeaders(times(1), 3);
|
||||
writeVerifyWriteHeaders(times(1), 5);
|
||||
writeVerifyWriteHeaders(never(), 7);
|
||||
writeVerifyWriteHeaders(never(), 9);
|
||||
|
||||
// Simulate that we received a SETTINGS frame which
|
||||
// increased MAX_CONCURRENT_STREAMS to 5.
|
||||
connection.local().maxActiveStreams(5);
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
flush();
|
||||
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
writeVerifyWriteHeaders(times(1), 7);
|
||||
writeVerifyWriteHeaders(times(1), 9);
|
||||
|
||||
encoderWriteHeaders(11);
|
||||
flush();
|
||||
|
||||
writeVerifyWriteHeaders(times(1), 11);
|
||||
|
||||
assertEquals(5, connection.local().numActiveStreams());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void bufferUntilSettingsReceived() {
|
||||
int initialLimit = BufferingHttp2ConnectionEncoder.SMALLEST_MAX_CONCURRENT_STREAMS;
|
||||
int numStreams = initialLimit * 2;
|
||||
for (int ix = 0, nextStreamId = 3; ix < numStreams; ++ix, nextStreamId += 2) {
|
||||
encoderWriteHeaders(nextStreamId);
|
||||
flush();
|
||||
if (ix < initialLimit) {
|
||||
writeVerifyWriteHeaders(times(1), nextStreamId);
|
||||
} else {
|
||||
writeVerifyWriteHeaders(never(), nextStreamId);
|
||||
}
|
||||
}
|
||||
assertEquals(numStreams / 2, encoder.numBufferedStreams());
|
||||
|
||||
// Simulate that we received a SETTINGS frame.
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
assertEquals(numStreams, connection.local().numActiveStreams());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void closedBufferedStreamReleasesByteBuf() {
|
||||
encoder.writeSettingsAck(ctx, newPromise());
|
||||
connection.local().maxActiveStreams(0);
|
||||
ByteBuf data = mock(ByteBuf.class);
|
||||
|
||||
ChannelFuture headersFuture = encoderWriteHeaders(3);
|
||||
assertEquals(1, encoder.numBufferedStreams());
|
||||
ChannelFuture dataFuture = encoder.writeData(ctx, 3, data, 0, false, newPromise());
|
||||
|
||||
ChannelPromise rstPromise = mock(ChannelPromise.class);
|
||||
encoder.writeRstStream(ctx, 3, CANCEL.code(), rstPromise);
|
||||
|
||||
assertEquals(0, encoder.numBufferedStreams());
|
||||
verify(rstPromise).setSuccess();
|
||||
assertTrue(headersFuture.isSuccess());
|
||||
assertTrue(dataFuture.isSuccess());
|
||||
verify(data).release();
|
||||
}
|
||||
|
||||
private ChannelFuture encoderWriteHeaders(int streamId) {
|
||||
return encoder.writeHeaders(ctx, streamId, new DefaultHttp2Headers(), 0,
|
||||
DEFAULT_PRIORITY_WEIGHT, false, 0, false, newPromise());
|
||||
}
|
||||
|
||||
private void flush() {
|
||||
channel.flush();
|
||||
}
|
||||
|
||||
private void writeVerifyWriteHeaders(VerificationMode mode, int streamId) {
|
||||
verify(writer, mode).writeHeaders(eq(ctx), eq(streamId), any(Http2Headers.class), eq(0),
|
||||
eq(DEFAULT_PRIORITY_WEIGHT), eq(false), eq(0),
|
||||
eq(false), any(ChannelPromise.class));
|
||||
}
|
||||
|
||||
private static ByteBuf data() {
|
||||
ByteBuf buf = Unpooled.buffer(10);
|
||||
for (int i = 0; i < buf.writableBytes(); i++) {
|
||||
buf.writeByte(i);
|
||||
}
|
||||
return buf;
|
||||
}
|
||||
|
||||
private ChannelPromise newPromise() {
|
||||
return channel.newPromise();
|
||||
}
|
||||
}
|
||||
|
|
@ -173,7 +173,8 @@ public class NettyClientHandlerTest extends NettyHandlerTestBase<NettyClientHand
|
|||
|
||||
@Test
|
||||
public void cancelWhileBufferedShouldSucceed() throws Exception {
|
||||
connection().local().maxActiveStreams(0);
|
||||
// Force the stream to be buffered.
|
||||
receiveMaxConcurrentStreams(0);
|
||||
|
||||
ChannelFuture createFuture = createStream();
|
||||
assertFalse(createFuture.isDone());
|
||||
|
|
|
|||
Loading…
Reference in New Issue