2015-01-20 01:48:11 +01:00
|
|
|
/*
|
|
|
|
* 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;
|
2015-03-23 18:52:11 +01:00
|
|
|
import io.netty.buffer.ByteBufUtil;
|
2015-01-20 01:48:11 +01:00
|
|
|
import io.netty.channel.ChannelFuture;
|
|
|
|
import io.netty.channel.ChannelFutureListener;
|
|
|
|
import io.netty.channel.ChannelHandlerContext;
|
|
|
|
import io.netty.channel.ChannelOutboundHandler;
|
|
|
|
import io.netty.channel.ChannelPromise;
|
|
|
|
import io.netty.handler.codec.ByteToMessageDecoder;
|
|
|
|
import io.netty.handler.codec.http2.Http2Exception.CompositeStreamException;
|
|
|
|
import io.netty.handler.codec.http2.Http2Exception.StreamException;
|
2015-08-18 19:36:16 +02:00
|
|
|
import io.netty.util.concurrent.ScheduledFuture;
|
|
|
|
import io.netty.util.internal.OneTimeTask;
|
2015-03-20 02:36:24 +01:00
|
|
|
import io.netty.util.internal.logging.InternalLogger;
|
|
|
|
import io.netty.util.internal.logging.InternalLoggerFactory;
|
2015-01-20 01:48:11 +01:00
|
|
|
|
|
|
|
import java.net.SocketAddress;
|
|
|
|
import java.util.List;
|
2015-08-18 19:36:16 +02:00
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
|
|
|
|
import static io.netty.buffer.ByteBufUtil.hexDump;
|
|
|
|
import static io.netty.handler.codec.http2.Http2CodecUtil.HTTP_UPGRADE_STREAM_ID;
|
|
|
|
import static io.netty.handler.codec.http2.Http2CodecUtil.connectionPrefaceBuf;
|
|
|
|
import static io.netty.handler.codec.http2.Http2CodecUtil.getEmbeddedHttp2Exception;
|
|
|
|
import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
|
|
|
|
import static io.netty.handler.codec.http2.Http2Error.NO_ERROR;
|
|
|
|
import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
|
|
|
|
import static io.netty.handler.codec.http2.Http2Exception.connectionError;
|
|
|
|
import static io.netty.handler.codec.http2.Http2Exception.isStreamError;
|
|
|
|
import static io.netty.handler.codec.http2.Http2FrameTypes.SETTINGS;
|
|
|
|
import static io.netty.util.CharsetUtil.UTF_8;
|
|
|
|
import static io.netty.util.internal.ObjectUtil.checkNotNull;
|
|
|
|
import static java.lang.Math.min;
|
|
|
|
import static java.lang.String.format;
|
|
|
|
import static java.util.concurrent.TimeUnit.MILLISECONDS;
|
|
|
|
import static java.util.concurrent.TimeUnit.SECONDS;
|
2015-01-20 01:48:11 +01:00
|
|
|
|
|
|
|
/**
|
2015-02-02 19:20:56 +01:00
|
|
|
* Provides the default implementation for processing inbound frame events and delegates to a
|
|
|
|
* {@link Http2FrameListener}
|
2015-01-20 01:48:11 +01:00
|
|
|
* <p>
|
|
|
|
* This class will read HTTP/2 frames and delegate the events to a {@link Http2FrameListener}
|
|
|
|
* <p>
|
2015-02-02 19:20:56 +01:00
|
|
|
* This interface enforces inbound flow control functionality through
|
|
|
|
* {@link Http2LocalFlowController}
|
2015-01-20 01:48:11 +01:00
|
|
|
*/
|
|
|
|
public class Http2ConnectionHandler extends ByteToMessageDecoder implements Http2LifecycleManager,
|
|
|
|
ChannelOutboundHandler {
|
2015-04-09 21:30:16 +02:00
|
|
|
private static final InternalLogger logger = InternalLoggerFactory.getInstance(Http2ConnectionHandler.class);
|
2015-08-18 19:36:16 +02:00
|
|
|
private static final long DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_MILLIS = MILLISECONDS.convert(30, SECONDS);
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
private final Http2ConnectionDecoder decoder;
|
|
|
|
private final Http2ConnectionEncoder encoder;
|
2015-05-21 21:12:29 +02:00
|
|
|
private final Http2Settings initialSettings;
|
2015-01-20 01:48:11 +01:00
|
|
|
private ChannelFutureListener closeListener;
|
2015-03-23 18:52:11 +01:00
|
|
|
private BaseDecoder byteDecoder;
|
2015-09-27 02:44:11 +02:00
|
|
|
private long gracefulShutdownTimeoutMillis;
|
2015-01-20 01:48:11 +01:00
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Builder which builds {@link Http2ConnectionHandler} objects.
|
|
|
|
*/
|
|
|
|
public static final class Builder extends BuilderBase<Http2ConnectionHandler, Builder> {
|
|
|
|
@Override
|
|
|
|
public Http2ConnectionHandler build0(Http2ConnectionDecoder decoder,
|
|
|
|
Http2ConnectionEncoder encoder) {
|
|
|
|
return new Http2ConnectionHandler(decoder, encoder, initialSettings());
|
|
|
|
}
|
2015-09-03 20:45:38 +02:00
|
|
|
}
|
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Base class for a {@code builder} of any subclass of {@link Http2ConnectionHandler}.
|
|
|
|
* @param <T> The type of handler created by this builder.
|
|
|
|
* @param <B> The concrete type for this builder.
|
|
|
|
*/
|
|
|
|
public abstract static class BuilderBase<T extends Http2ConnectionHandler, B extends BuilderBase<T, B>> {
|
|
|
|
private Http2Settings initialSettings = new Http2Settings();
|
|
|
|
private Http2FrameListener frameListener;
|
|
|
|
private Http2FrameLogger frameLogger;
|
|
|
|
private boolean validateHeaders = true;
|
|
|
|
private boolean server = true;
|
2015-10-14 19:58:54 +02:00
|
|
|
private boolean encoderEnforceMaxConcurrentStreams;
|
2015-09-27 02:44:11 +02:00
|
|
|
private long gracefulShutdownTimeoutMillis = DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT_MILLIS;
|
2015-01-20 01:48:11 +01:00
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Sets the listener for inbound frames.
|
|
|
|
* This listener will only be set if the decoder's listener is {@code null}.
|
|
|
|
*/
|
|
|
|
public B frameListener(Http2FrameListener listener) {
|
|
|
|
frameListener = listener;
|
|
|
|
return thisB();
|
|
|
|
}
|
2015-09-03 20:45:38 +02:00
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Determine if HTTP headers should be validated according to
|
|
|
|
* <a href="https://tools.ietf.org/html/rfc7540#section-8.1.2.6">RFC 7540, 8.1.2.6</a>.
|
|
|
|
*/
|
|
|
|
public B validateHeaders(boolean validate) {
|
|
|
|
validateHeaders = validate;
|
|
|
|
return thisB();
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Get if HTTP headers should be validated according to
|
|
|
|
* <a href="https://tools.ietf.org/html/rfc7540#section-8.1.2.6">RFC 7540, 8.1.2.6</a>.
|
|
|
|
*/
|
|
|
|
public final boolean isValidateHeaders() {
|
|
|
|
return validateHeaders;
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Settings to use for the initial connection settings exchange.
|
|
|
|
*/
|
|
|
|
public B initialSettings(Http2Settings settings) {
|
|
|
|
initialSettings = settings;
|
|
|
|
return thisB();
|
2015-05-21 21:12:29 +02:00
|
|
|
}
|
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Get the settings to use for the initial connection settings exchange.
|
|
|
|
*/
|
|
|
|
public final Http2Settings initialSettings() {
|
|
|
|
return initialSettings;
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Determines if {@link #build()} will to create a {@link Http2Connection} in server mode ({@code true})
|
|
|
|
* or client mode ({@code false}).
|
|
|
|
*/
|
|
|
|
public B server(boolean isServer) {
|
|
|
|
server = isServer;
|
|
|
|
return thisB();
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Set the logger that is used for the encoder and decoder.
|
|
|
|
*/
|
|
|
|
public B frameLogger(Http2FrameLogger logger) {
|
|
|
|
frameLogger = logger;
|
|
|
|
return thisB();
|
|
|
|
}
|
|
|
|
|
|
|
|
public B gracefulShutdownTimeoutMillis(long gracefulShutdownTimeoutMillis) {
|
|
|
|
this.gracefulShutdownTimeoutMillis = gracefulShutdownTimeoutMillis;
|
|
|
|
return thisB();
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
2015-10-14 19:58:54 +02:00
|
|
|
* Determine if the encoder should queue frames if the maximum number of concurrent streams
|
|
|
|
* would otherwise be exceeded.
|
2015-09-27 02:44:11 +02:00
|
|
|
*/
|
|
|
|
public B encoderEnforceMaxConcurrentStreams(boolean encoderEnforceMaxConcurrentStreams) {
|
2015-10-14 19:58:54 +02:00
|
|
|
this.encoderEnforceMaxConcurrentStreams = encoderEnforceMaxConcurrentStreams;
|
2015-09-27 02:44:11 +02:00
|
|
|
return thisB();
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Create a new {@link Http2Connection} and build a new instance.
|
|
|
|
*/
|
|
|
|
public final T build() {
|
|
|
|
return build(new DefaultHttp2Connection(server));
|
|
|
|
}
|
2015-05-21 21:12:29 +02:00
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
/**
|
|
|
|
* Build a new instance with an existing {@link Http2Connection}.
|
|
|
|
* <p>
|
|
|
|
* Methods that will be ignored due to objects already being created:
|
|
|
|
* <ul><li>{@link #server(boolean)}</li></ul>
|
|
|
|
*/
|
|
|
|
public final T build(Http2Connection connection) {
|
|
|
|
Http2FrameReader reader = new DefaultHttp2FrameReader(validateHeaders);
|
|
|
|
Http2FrameWriter writer = new DefaultHttp2FrameWriter();
|
|
|
|
if (frameLogger != null) {
|
|
|
|
reader = new Http2InboundFrameLogger(reader, frameLogger);
|
|
|
|
writer = new Http2OutboundFrameLogger(writer, frameLogger);
|
|
|
|
}
|
|
|
|
Http2ConnectionEncoder encoder = new DefaultHttp2ConnectionEncoder(connection, writer);
|
2015-10-14 19:58:54 +02:00
|
|
|
if (encoderEnforceMaxConcurrentStreams) {
|
|
|
|
if (connection.isServer()) {
|
2015-10-23 23:03:08 +02:00
|
|
|
encoder.close();
|
|
|
|
reader.close();
|
2015-10-14 19:58:54 +02:00
|
|
|
throw new IllegalArgumentException(
|
|
|
|
"encoderEnforceMaxConcurrentStreams: " + encoderEnforceMaxConcurrentStreams +
|
|
|
|
" not supported for server");
|
|
|
|
}
|
|
|
|
encoder = new StreamBufferingEncoder(encoder);
|
2015-09-27 02:44:11 +02:00
|
|
|
}
|
|
|
|
Http2ConnectionDecoder decoder = new DefaultHttp2ConnectionDecoder(connection, encoder, reader);
|
|
|
|
return build(decoder, encoder);
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Build a new instance with an existing {@link Http2ConnectionDecoder} and {@link Http2ConnectionEncoder}.
|
|
|
|
* <p>
|
|
|
|
* Methods that will be ignored due to objects already being created:
|
2015-10-14 19:58:54 +02:00
|
|
|
* <ul><li>{@link #server(boolean)}</li><li>
|
2015-09-27 02:44:11 +02:00
|
|
|
* {@link #frameLogger(Http2FrameLogger)}</li><li>{@link #encoderEnforceMaxConcurrentStreams(boolean)}</li><li>
|
2015-10-14 19:58:54 +02:00
|
|
|
* {@link #encoderEnforceMaxConcurrentStreams(boolean)} (int)}</li></ul>
|
2015-09-27 02:44:11 +02:00
|
|
|
*/
|
|
|
|
public final T build(Http2ConnectionDecoder decoder, Http2ConnectionEncoder encoder) {
|
2015-10-23 23:03:08 +02:00
|
|
|
final T handler;
|
|
|
|
try {
|
|
|
|
// Call the abstract build method
|
|
|
|
handler = build0(decoder, encoder);
|
|
|
|
} catch (RuntimeException e) {
|
|
|
|
encoder.close();
|
|
|
|
decoder.close();
|
|
|
|
throw e;
|
|
|
|
}
|
2015-09-27 02:44:11 +02:00
|
|
|
|
|
|
|
// Setup post build options
|
|
|
|
handler.gracefulShutdownTimeoutMillis(gracefulShutdownTimeoutMillis);
|
|
|
|
if (handler.decoder().frameListener() == null) {
|
|
|
|
handler.decoder().frameListener(frameListener);
|
|
|
|
}
|
|
|
|
return handler;
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Sub classes should override this to instantiate the concrete type.
|
|
|
|
* <p>
|
|
|
|
* The return of this method will be subject to the following:
|
|
|
|
* <ul><li>{@link #frameListener(Http2FrameListener)} will be set if not already set in the decoder</li><li>
|
|
|
|
* {@link #gracefulShutdownTimeoutMillis(long)} will be set</li></ul>
|
|
|
|
*/
|
|
|
|
protected abstract T build0(Http2ConnectionDecoder decoder, Http2ConnectionEncoder encoder);
|
|
|
|
|
|
|
|
@SuppressWarnings("unchecked")
|
|
|
|
protected B thisB() {
|
|
|
|
return (B) this;
|
|
|
|
}
|
2015-05-21 21:12:29 +02:00
|
|
|
}
|
|
|
|
|
2015-09-27 02:44:11 +02:00
|
|
|
protected Http2ConnectionHandler(Http2ConnectionDecoder decoder, Http2ConnectionEncoder encoder,
|
|
|
|
Http2Settings initialSettings) {
|
|
|
|
this.initialSettings = checkNotNull(initialSettings, "initialSettings");
|
2015-03-27 23:37:20 +01:00
|
|
|
this.decoder = checkNotNull(decoder, "decoder");
|
|
|
|
this.encoder = checkNotNull(encoder, "encoder");
|
2015-01-20 01:48:11 +01:00
|
|
|
if (encoder.connection() != decoder.connection()) {
|
|
|
|
throw new IllegalArgumentException("Encoder and Decoder do not share the same connection object");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-08-18 19:36:16 +02:00
|
|
|
/**
|
|
|
|
* Get the amount of time (in milliseconds) this endpoint will wait for all streams to be closed before closing
|
|
|
|
* the connection during the graceful shutdown process.
|
|
|
|
*/
|
|
|
|
public long gracefulShutdownTimeoutMillis() {
|
|
|
|
return gracefulShutdownTimeoutMillis;
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Set the amount of time (in milliseconds) this endpoint will wait for all streams to be closed before closing
|
|
|
|
* the connection during the graceful shutdown process.
|
|
|
|
* @param gracefulShutdownTimeoutMillis the amount of time (in milliseconds) this endpoint will wait for all
|
|
|
|
* streams to be closed before closing the connection during the graceful shutdown process.
|
|
|
|
*/
|
|
|
|
public void gracefulShutdownTimeoutMillis(long gracefulShutdownTimeoutMillis) {
|
|
|
|
if (gracefulShutdownTimeoutMillis < 0) {
|
|
|
|
throw new IllegalArgumentException("gracefulShutdownTimeoutMillis: " + gracefulShutdownTimeoutMillis +
|
|
|
|
" (expected: >= 0)");
|
|
|
|
}
|
|
|
|
this.gracefulShutdownTimeoutMillis = gracefulShutdownTimeoutMillis;
|
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
public Http2Connection connection() {
|
|
|
|
return encoder.connection();
|
|
|
|
}
|
|
|
|
|
|
|
|
public Http2ConnectionDecoder decoder() {
|
|
|
|
return decoder;
|
|
|
|
}
|
|
|
|
|
|
|
|
public Http2ConnectionEncoder encoder() {
|
|
|
|
return encoder;
|
|
|
|
}
|
|
|
|
|
2015-03-23 18:52:11 +01:00
|
|
|
private boolean prefaceSent() {
|
|
|
|
return byteDecoder != null && byteDecoder.prefaceSent();
|
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
/**
|
|
|
|
* Handles the client-side (cleartext) upgrade from HTTP to HTTP/2.
|
|
|
|
* Reserves local stream 1 for the HTTP/2 response.
|
|
|
|
*/
|
|
|
|
public void onHttpClientUpgrade() throws Http2Exception {
|
|
|
|
if (connection().isServer()) {
|
|
|
|
throw connectionError(PROTOCOL_ERROR, "Client-side HTTP upgrade requested for a server");
|
|
|
|
}
|
2015-03-23 18:52:11 +01:00
|
|
|
if (prefaceSent() || decoder.prefaceReceived()) {
|
2015-01-20 01:48:11 +01:00
|
|
|
throw connectionError(PROTOCOL_ERROR, "HTTP upgrade must occur before HTTP/2 preface is sent or received");
|
|
|
|
}
|
|
|
|
|
|
|
|
// Create a local stream used for the HTTP cleartext upgrade.
|
2015-04-22 23:35:31 +02:00
|
|
|
connection().local().createStream(HTTP_UPGRADE_STREAM_ID, true);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Handles the server-side (cleartext) upgrade from HTTP to HTTP/2.
|
|
|
|
* @param settings the settings for the remote endpoint.
|
|
|
|
*/
|
|
|
|
public void onHttpServerUpgrade(Http2Settings settings) throws Http2Exception {
|
|
|
|
if (!connection().isServer()) {
|
|
|
|
throw connectionError(PROTOCOL_ERROR, "Server-side HTTP upgrade requested for a client");
|
|
|
|
}
|
2015-03-23 18:52:11 +01:00
|
|
|
if (prefaceSent() || decoder.prefaceReceived()) {
|
2015-01-20 01:48:11 +01:00
|
|
|
throw connectionError(PROTOCOL_ERROR, "HTTP upgrade must occur before HTTP/2 preface is sent or received");
|
|
|
|
}
|
|
|
|
|
|
|
|
// Apply the settings but no ACK is necessary.
|
|
|
|
encoder.remoteSettings(settings);
|
|
|
|
|
|
|
|
// Create a stream in the half-closed state.
|
2015-04-22 23:35:31 +02:00
|
|
|
connection().remote().createStream(HTTP_UPGRADE_STREAM_ID, true);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
2015-06-04 20:55:18 +02:00
|
|
|
@Override
|
|
|
|
public void flush(ChannelHandlerContext ctx) throws Http2Exception {
|
|
|
|
// Trigger pending writes in the remote flow controller.
|
2015-08-03 21:46:29 +02:00
|
|
|
encoder.flowController().writePendingBytes();
|
2015-06-04 20:55:18 +02:00
|
|
|
try {
|
2015-06-20 01:08:37 +02:00
|
|
|
ctx.flush();
|
2015-06-04 20:55:18 +02:00
|
|
|
} catch (Throwable t) {
|
|
|
|
throw new Http2Exception(INTERNAL_ERROR, "Error flushing" , t);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-03-23 18:52:11 +01:00
|
|
|
private abstract class BaseDecoder {
|
|
|
|
public abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
|
|
|
|
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { }
|
|
|
|
public void channelActive(ChannelHandlerContext ctx) throws Exception { }
|
|
|
|
|
|
|
|
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
2015-07-08 20:38:22 +02:00
|
|
|
// Connection has terminated, close the encoder and decoder.
|
|
|
|
encoder().close();
|
|
|
|
decoder().close();
|
|
|
|
|
|
|
|
final Http2Connection connection = connection();
|
|
|
|
// Check if there are streams to avoid the overhead of creating the ChannelFuture.
|
|
|
|
if (connection.numActiveStreams() > 0) {
|
|
|
|
final ChannelFuture future = ctx.newSucceededFuture();
|
|
|
|
connection.forEachActiveStream(new Http2StreamVisitor() {
|
|
|
|
@Override
|
|
|
|
public boolean visit(Http2Stream stream) throws Http2Exception {
|
|
|
|
closeStream(stream, future);
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
});
|
2015-03-23 18:52:11 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Determine if the HTTP/2 connection preface been sent.
|
|
|
|
*/
|
|
|
|
public boolean prefaceSent() {
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private final class PrefaceDecoder extends BaseDecoder {
|
|
|
|
private ByteBuf clientPrefaceString;
|
|
|
|
private boolean prefaceSent;
|
|
|
|
|
|
|
|
public PrefaceDecoder(ChannelHandlerContext ctx) {
|
|
|
|
clientPrefaceString = clientPrefaceString(encoder.connection());
|
|
|
|
// This handler was just added to the context. In case it was handled after
|
|
|
|
// the connection became active, send the connection preface now.
|
|
|
|
sendPreface(ctx);
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public boolean prefaceSent() {
|
|
|
|
return prefaceSent;
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
|
|
|
|
try {
|
2015-06-17 00:11:28 +02:00
|
|
|
if (readClientPrefaceString(in) && verifyFirstFrameIsSettings(in)) {
|
2015-03-23 18:52:11 +01:00
|
|
|
// After the preface is read, it is time to hand over control to the post initialized decoder.
|
2015-06-17 00:11:28 +02:00
|
|
|
byteDecoder = new FrameDecoder();
|
|
|
|
byteDecoder.decode(ctx, in, out);
|
2015-03-23 18:52:11 +01:00
|
|
|
}
|
|
|
|
} catch (Throwable e) {
|
2015-09-16 00:33:17 +02:00
|
|
|
onError(ctx, e);
|
2015-03-23 18:52:11 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
// The channel just became active - send the connection preface to the remote endpoint.
|
|
|
|
sendPreface(ctx);
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
cleanup();
|
|
|
|
super.channelInactive(ctx);
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Releases the {@code clientPrefaceString}. Any active streams will be left in the open.
|
|
|
|
*/
|
|
|
|
@Override
|
|
|
|
public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
cleanup();
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Releases the {@code clientPrefaceString}. Any active streams will be left in the open.
|
|
|
|
*/
|
|
|
|
private void cleanup() {
|
|
|
|
if (clientPrefaceString != null) {
|
|
|
|
clientPrefaceString.release();
|
|
|
|
clientPrefaceString = null;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Decodes the client connection preface string from the input buffer.
|
|
|
|
*
|
|
|
|
* @return {@code true} if processing of the client preface string is complete. Since client preface strings can
|
|
|
|
* only be received by servers, returns true immediately for client endpoints.
|
|
|
|
*/
|
|
|
|
private boolean readClientPrefaceString(ByteBuf in) throws Http2Exception {
|
|
|
|
if (clientPrefaceString == null) {
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
|
|
|
int prefaceRemaining = clientPrefaceString.readableBytes();
|
2015-06-17 00:11:28 +02:00
|
|
|
int bytesRead = min(in.readableBytes(), prefaceRemaining);
|
2015-03-23 18:52:11 +01:00
|
|
|
|
|
|
|
// If the input so far doesn't match the preface, break the connection.
|
|
|
|
if (bytesRead == 0 || !ByteBufUtil.equals(in, in.readerIndex(),
|
|
|
|
clientPrefaceString, clientPrefaceString.readerIndex(), bytesRead)) {
|
2015-06-17 00:11:28 +02:00
|
|
|
String receivedBytes = hexDump(in, in.readerIndex(),
|
|
|
|
min(in.readableBytes(), clientPrefaceString.readableBytes()));
|
|
|
|
throw connectionError(PROTOCOL_ERROR, "HTTP/2 client preface string missing or corrupt. " +
|
|
|
|
"Hex dump for received bytes: %s", receivedBytes);
|
2015-03-23 18:52:11 +01:00
|
|
|
}
|
|
|
|
in.skipBytes(bytesRead);
|
|
|
|
clientPrefaceString.skipBytes(bytesRead);
|
|
|
|
|
|
|
|
if (!clientPrefaceString.isReadable()) {
|
|
|
|
// Entire preface has been read.
|
|
|
|
clientPrefaceString.release();
|
|
|
|
clientPrefaceString = null;
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
|
2015-06-17 00:11:28 +02:00
|
|
|
/**
|
|
|
|
* Peeks at that the next frame in the buffer and verifies that it is a {@code SETTINGS} frame.
|
|
|
|
*
|
|
|
|
* @param in the inbound buffer.
|
|
|
|
* @return {@code} true if the next frame is a {@code SETTINGS} frame, {@code false} if more
|
|
|
|
* data is required before we can determine the next frame type.
|
|
|
|
* @throws Http2Exception thrown if the next frame is NOT a {@code SETTINGS} frame.
|
|
|
|
*/
|
|
|
|
private boolean verifyFirstFrameIsSettings(ByteBuf in) throws Http2Exception {
|
|
|
|
if (in.readableBytes() < 4) {
|
|
|
|
// Need more data before we can see the frame type for the first frame.
|
|
|
|
return false;
|
|
|
|
}
|
|
|
|
|
|
|
|
byte frameType = in.getByte(in.readerIndex() + 3);
|
|
|
|
if (frameType != SETTINGS) {
|
|
|
|
throw connectionError(PROTOCOL_ERROR, "First received frame was not SETTINGS. " +
|
|
|
|
"Hex dump for first 4 bytes: %s", hexDump(in, in.readerIndex(), 4));
|
|
|
|
}
|
|
|
|
return true;
|
|
|
|
}
|
|
|
|
|
2015-03-23 18:52:11 +01:00
|
|
|
/**
|
|
|
|
* Sends the HTTP/2 connection preface upon establishment of the connection, if not already sent.
|
|
|
|
*/
|
|
|
|
private void sendPreface(ChannelHandlerContext ctx) {
|
|
|
|
if (prefaceSent || !ctx.channel().isActive()) {
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
prefaceSent = true;
|
|
|
|
|
|
|
|
if (!connection().isServer()) {
|
|
|
|
// Clients must send the preface string as the first bytes on the connection.
|
|
|
|
ctx.write(connectionPrefaceBuf()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
|
|
|
|
}
|
|
|
|
|
|
|
|
// Both client and server must send their initial settings.
|
2015-09-27 02:44:11 +02:00
|
|
|
encoder.writeSettings(ctx, initialSettings, ctx.newPromise()).addListener(
|
2015-03-23 18:52:11 +01:00
|
|
|
ChannelFutureListener.CLOSE_ON_FAILURE);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
private final class FrameDecoder extends BaseDecoder {
|
|
|
|
@Override
|
|
|
|
public void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
|
|
|
|
try {
|
|
|
|
decoder.decodeFrame(ctx, in, out);
|
|
|
|
} catch (Throwable e) {
|
2015-09-16 00:33:17 +02:00
|
|
|
onError(ctx, e);
|
2015-03-23 18:52:11 +01:00
|
|
|
}
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
|
2015-06-30 19:10:17 +02:00
|
|
|
// Initialize the encoder, decoder, flow controllers, and internal state.
|
2015-03-27 23:37:20 +01:00
|
|
|
encoder.lifecycleManager(this);
|
|
|
|
decoder.lifecycleManager(this);
|
2015-06-30 19:10:17 +02:00
|
|
|
encoder.flowController().channelHandlerContext(ctx);
|
|
|
|
decoder.flowController().channelHandlerContext(ctx);
|
2015-03-23 18:52:11 +01:00
|
|
|
byteDecoder = new PrefaceDecoder(ctx);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
|
2015-03-23 18:52:11 +01:00
|
|
|
if (byteDecoder != null) {
|
|
|
|
byteDecoder.handlerRemoved(ctx);
|
|
|
|
byteDecoder = null;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
if (byteDecoder == null) {
|
|
|
|
byteDecoder = new PrefaceDecoder(ctx);
|
2015-03-18 22:09:19 +01:00
|
|
|
}
|
2015-03-23 18:52:11 +01:00
|
|
|
byteDecoder.channelActive(ctx);
|
2015-04-29 01:27:54 +02:00
|
|
|
super.channelActive(ctx);
|
2015-03-23 18:52:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
2015-08-21 04:23:37 +02:00
|
|
|
// Call super class first, as this may result in decode being called.
|
|
|
|
super.channelInactive(ctx);
|
2015-03-23 18:52:11 +01:00
|
|
|
if (byteDecoder != null) {
|
|
|
|
byteDecoder.channelInactive(ctx);
|
|
|
|
byteDecoder = null;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-06-30 19:10:17 +02:00
|
|
|
@Override
|
|
|
|
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
// Writability is expected to change while we are writing. We cannot allow this event to trigger reentering
|
|
|
|
// the allocation and write loop. Reentering the event loop will lead to over or illegal allocation.
|
2015-08-03 21:46:29 +02:00
|
|
|
try {
|
|
|
|
if (ctx.channel().isWritable()) {
|
|
|
|
flush(ctx);
|
|
|
|
}
|
2015-09-26 00:49:05 +02:00
|
|
|
encoder.flowController().channelWritabilityChanged();
|
2015-08-03 21:46:29 +02:00
|
|
|
} finally {
|
|
|
|
super.channelWritabilityChanged(ctx);
|
2015-06-30 19:10:17 +02:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-03-23 18:52:11 +01:00
|
|
|
@Override
|
|
|
|
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
|
|
|
|
byteDecoder.decode(ctx, in, out);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
|
|
|
|
ctx.bind(localAddress, promise);
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
|
|
|
|
ChannelPromise promise) throws Exception {
|
|
|
|
ctx.connect(remoteAddress, localAddress, promise);
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
|
|
|
|
ctx.disconnect(promise);
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
|
|
|
|
// Avoid NotYetConnectedException
|
|
|
|
if (!ctx.channel().isActive()) {
|
|
|
|
ctx.close(promise);
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
2015-04-02 23:39:46 +02:00
|
|
|
ChannelFuture future = goAway(ctx, null);
|
2015-04-23 23:23:23 +02:00
|
|
|
ctx.flush();
|
2015-08-18 19:36:16 +02:00
|
|
|
doGracefulShutdown(ctx, future, promise);
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
|
2015-08-18 19:36:16 +02:00
|
|
|
private void doGracefulShutdown(ChannelHandlerContext ctx, ChannelFuture future, ChannelPromise promise) {
|
2015-01-20 01:48:11 +01:00
|
|
|
// If there are no active streams, close immediately after the send is complete.
|
|
|
|
// Otherwise wait until all streams are inactive.
|
2015-05-06 21:04:55 +02:00
|
|
|
if (isGracefulShutdownComplete()) {
|
2015-01-20 01:48:11 +01:00
|
|
|
future.addListener(new ClosingChannelFutureListener(ctx, promise));
|
|
|
|
} else {
|
2015-08-18 19:36:16 +02:00
|
|
|
closeListener = new ClosingChannelFutureListener(ctx, promise,
|
|
|
|
gracefulShutdownTimeoutMillis, MILLISECONDS);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-04-29 07:51:32 +02:00
|
|
|
@Override
|
|
|
|
public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
|
|
|
|
ctx.deregister(promise);
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void read(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
ctx.read();
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
|
|
|
|
ctx.write(msg, promise);
|
|
|
|
}
|
|
|
|
|
2015-04-23 23:23:23 +02:00
|
|
|
@Override
|
|
|
|
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
|
|
|
|
// Trigger flush after read on the assumption that flush is cheap if there is nothing to write and that
|
|
|
|
// for flow-control the read may release window that causes data to be written that can now be flushed.
|
2015-07-07 17:14:12 +02:00
|
|
|
try {
|
|
|
|
flush(ctx);
|
|
|
|
} finally {
|
|
|
|
super.channelReadComplete(ctx);
|
|
|
|
}
|
2015-04-23 23:23:23 +02:00
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
/**
|
|
|
|
* Handles {@link Http2Exception} objects that were thrown from other handlers. Ignores all other exceptions.
|
|
|
|
*/
|
|
|
|
@Override
|
|
|
|
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
|
|
|
if (getEmbeddedHttp2Exception(cause) != null) {
|
|
|
|
// Some exception in the causality chain is an Http2Exception - handle it.
|
2015-09-16 00:33:17 +02:00
|
|
|
onError(ctx, cause);
|
2015-01-20 01:48:11 +01:00
|
|
|
} else {
|
|
|
|
super.exceptionCaught(ctx, cause);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Closes the local side of the given stream. If this causes the stream to be closed, adds a
|
|
|
|
* hook to close the channel after the given future completes.
|
|
|
|
*
|
|
|
|
* @param stream the stream to be half closed.
|
|
|
|
* @param future If closing, the future after which to close the channel.
|
|
|
|
*/
|
|
|
|
@Override
|
2015-04-02 23:39:46 +02:00
|
|
|
public void closeStreamLocal(Http2Stream stream, ChannelFuture future) {
|
2015-01-20 01:48:11 +01:00
|
|
|
switch (stream.state()) {
|
|
|
|
case HALF_CLOSED_LOCAL:
|
|
|
|
case OPEN:
|
|
|
|
stream.closeLocalSide();
|
|
|
|
break;
|
|
|
|
default:
|
|
|
|
closeStream(stream, future);
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Closes the remote side of the given stream. If this causes the stream to be closed, adds a
|
|
|
|
* hook to close the channel after the given future completes.
|
|
|
|
*
|
|
|
|
* @param stream the stream to be half closed.
|
|
|
|
* @param future If closing, the future after which to close the channel.
|
|
|
|
*/
|
|
|
|
@Override
|
2015-04-02 23:39:46 +02:00
|
|
|
public void closeStreamRemote(Http2Stream stream, ChannelFuture future) {
|
2015-01-20 01:48:11 +01:00
|
|
|
switch (stream.state()) {
|
|
|
|
case HALF_CLOSED_REMOTE:
|
|
|
|
case OPEN:
|
|
|
|
stream.closeRemoteSide();
|
|
|
|
break;
|
|
|
|
default:
|
|
|
|
closeStream(stream, future);
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2015-02-02 19:20:56 +01:00
|
|
|
public void closeStream(final Http2Stream stream, ChannelFuture future) {
|
2015-01-20 01:48:11 +01:00
|
|
|
stream.close();
|
|
|
|
|
2015-05-30 01:27:59 +02:00
|
|
|
if (future.isDone()) {
|
|
|
|
checkCloseConnection(future);
|
|
|
|
} else {
|
|
|
|
future.addListener(new ChannelFutureListener() {
|
|
|
|
@Override
|
|
|
|
public void operationComplete(ChannelFuture future) throws Exception {
|
|
|
|
checkCloseConnection(future);
|
2015-05-06 21:04:55 +02:00
|
|
|
}
|
2015-05-30 01:27:59 +02:00
|
|
|
});
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Central handler for all exceptions caught during HTTP/2 processing.
|
|
|
|
*/
|
|
|
|
@Override
|
2015-09-16 00:33:17 +02:00
|
|
|
public void onError(ChannelHandlerContext ctx, Throwable cause) {
|
2015-01-20 01:48:11 +01:00
|
|
|
Http2Exception embedded = getEmbeddedHttp2Exception(cause);
|
|
|
|
if (isStreamError(embedded)) {
|
|
|
|
onStreamError(ctx, cause, (StreamException) embedded);
|
|
|
|
} else if (embedded instanceof CompositeStreamException) {
|
|
|
|
CompositeStreamException compositException = (CompositeStreamException) embedded;
|
|
|
|
for (StreamException streamException : compositException) {
|
|
|
|
onStreamError(ctx, cause, streamException);
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
onConnectionError(ctx, cause, embedded);
|
|
|
|
}
|
2015-04-23 23:23:23 +02:00
|
|
|
ctx.flush();
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
2015-05-06 21:04:55 +02:00
|
|
|
/**
|
|
|
|
* Called by the graceful shutdown logic to determine when it is safe to close the connection. Returns {@code true}
|
|
|
|
* if the graceful shutdown has completed and the connection can be safely closed. This implementation just
|
|
|
|
* guarantees that there are no active streams. Subclasses may override to provide additional checks.
|
|
|
|
*/
|
|
|
|
protected boolean isGracefulShutdownComplete() {
|
|
|
|
return connection().numActiveStreams() == 0;
|
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
/**
|
|
|
|
* Handler for a connection error. Sends a GO_AWAY frame to the remote endpoint. Once all
|
|
|
|
* streams are closed, the connection is shut down.
|
|
|
|
*
|
|
|
|
* @param ctx the channel context
|
|
|
|
* @param cause the exception that was caught
|
|
|
|
* @param http2Ex the {@link Http2Exception} that is embedded in the causality chain. This may
|
|
|
|
* be {@code null} if it's an unknown exception.
|
|
|
|
*/
|
|
|
|
protected void onConnectionError(ChannelHandlerContext ctx, Throwable cause, Http2Exception http2Ex) {
|
|
|
|
if (http2Ex == null) {
|
|
|
|
http2Ex = new Http2Exception(INTERNAL_ERROR, cause.getMessage(), cause);
|
|
|
|
}
|
2015-08-18 19:36:16 +02:00
|
|
|
|
|
|
|
ChannelPromise promise = ctx.newPromise();
|
|
|
|
ChannelFuture future = goAway(ctx, http2Ex);
|
|
|
|
switch (http2Ex.shutdownHint()) {
|
|
|
|
case GRACEFUL_SHUTDOWN:
|
|
|
|
doGracefulShutdown(ctx, future, promise);
|
|
|
|
break;
|
|
|
|
default:
|
|
|
|
future.addListener(new ClosingChannelFutureListener(ctx, promise));
|
|
|
|
break;
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Handler for a stream error. Sends a {@code RST_STREAM} frame to the remote endpoint and closes the
|
|
|
|
* stream.
|
|
|
|
*
|
|
|
|
* @param ctx the channel context
|
|
|
|
* @param cause the exception that was caught
|
|
|
|
* @param http2Ex the {@link StreamException} that is embedded in the causality chain.
|
|
|
|
*/
|
|
|
|
protected void onStreamError(ChannelHandlerContext ctx, Throwable cause, StreamException http2Ex) {
|
2015-04-02 23:39:46 +02:00
|
|
|
resetStream(ctx, http2Ex.streamId(), http2Ex.error().code(), ctx.newPromise());
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
protected Http2FrameWriter frameWriter() {
|
|
|
|
return encoder().frameWriter();
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2015-04-02 23:39:46 +02:00
|
|
|
public ChannelFuture resetStream(final ChannelHandlerContext ctx, int streamId, long errorCode,
|
2015-03-14 20:58:30 +01:00
|
|
|
final ChannelPromise promise) {
|
|
|
|
final Http2Stream stream = connection().stream(streamId);
|
|
|
|
if (stream == null || stream.isResetSent()) {
|
|
|
|
// Don't write a RST_STREAM frame if we are not aware of the stream, or if we have already written one.
|
|
|
|
return promise.setSuccess();
|
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
ChannelFuture future = frameWriter().writeRstStream(ctx, streamId, errorCode, promise);
|
|
|
|
|
2015-03-14 20:58:30 +01:00
|
|
|
// Synchronously set the resetSent flag to prevent any subsequent calls
|
|
|
|
// from resulting in multiple reset frames being sent.
|
|
|
|
stream.resetSent();
|
|
|
|
|
|
|
|
future.addListener(new ChannelFutureListener() {
|
|
|
|
@Override
|
|
|
|
public void operationComplete(ChannelFuture future) throws Exception {
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
closeStream(stream, promise);
|
|
|
|
} else {
|
|
|
|
// The connection will be closed and so no need to change the resetSent flag to false.
|
|
|
|
onConnectionError(ctx, future.cause(), null);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
});
|
2015-01-20 01:48:11 +01:00
|
|
|
|
|
|
|
return future;
|
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
2015-03-20 02:36:24 +01:00
|
|
|
public ChannelFuture goAway(final ChannelHandlerContext ctx, final int lastStreamId, final long errorCode,
|
|
|
|
final ByteBuf debugData, ChannelPromise promise) {
|
|
|
|
try {
|
|
|
|
final Http2Connection connection = connection();
|
2015-05-03 17:49:49 +02:00
|
|
|
if (connection.goAwaySent() && lastStreamId > connection.remote().lastStreamKnownByPeer()) {
|
2015-03-20 02:36:24 +01:00
|
|
|
throw connectionError(PROTOCOL_ERROR, "Last stream identifier must not increase between " +
|
|
|
|
"sending multiple GOAWAY frames (was '%d', is '%d').",
|
2015-05-03 17:49:49 +02:00
|
|
|
connection.remote().lastStreamKnownByPeer(),
|
2015-03-20 02:36:24 +01:00
|
|
|
lastStreamId);
|
|
|
|
}
|
|
|
|
connection.goAwaySent(lastStreamId, errorCode, debugData);
|
|
|
|
|
2015-05-11 21:10:23 +02:00
|
|
|
// Need to retain before we write the buffer because if we do it after the refCnt could already be 0 and
|
|
|
|
// result in an IllegalRefCountException.
|
|
|
|
debugData.retain();
|
2015-03-20 02:36:24 +01:00
|
|
|
ChannelFuture future = frameWriter().writeGoAway(ctx, lastStreamId, errorCode, debugData, promise);
|
|
|
|
|
2015-05-11 21:10:23 +02:00
|
|
|
if (future.isDone()) {
|
|
|
|
processGoAwayWriteResult(ctx, lastStreamId, errorCode, debugData, future);
|
|
|
|
} else {
|
|
|
|
future.addListener(new ChannelFutureListener() {
|
|
|
|
@Override
|
|
|
|
public void operationComplete(ChannelFuture future) throws Exception {
|
|
|
|
processGoAwayWriteResult(ctx, lastStreamId, errorCode, debugData, future);
|
2015-03-20 02:36:24 +01:00
|
|
|
}
|
2015-05-11 21:10:23 +02:00
|
|
|
});
|
|
|
|
}
|
2015-03-20 02:36:24 +01:00
|
|
|
|
|
|
|
return future;
|
2015-05-11 21:10:23 +02:00
|
|
|
} catch (Throwable cause) { // Make sure to catch Throwable because we are doing a retain() in this method.
|
2015-01-20 01:48:11 +01:00
|
|
|
debugData.release();
|
2015-05-11 21:10:23 +02:00
|
|
|
return promise.setFailure(cause);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-05-30 01:27:59 +02:00
|
|
|
/**
|
|
|
|
* Closes the connection if the graceful shutdown process has completed.
|
|
|
|
* @param future Represents the status that will be passed to the {@link #closeListener}.
|
|
|
|
*/
|
|
|
|
private void checkCloseConnection(ChannelFuture future) {
|
|
|
|
// If this connection is closing and the graceful shutdown has completed, close the connection
|
|
|
|
// once this operation completes.
|
|
|
|
if (closeListener != null && isGracefulShutdownComplete()) {
|
|
|
|
ChannelFutureListener closeListener = Http2ConnectionHandler.this.closeListener;
|
|
|
|
// This method could be called multiple times
|
|
|
|
// and we don't want to notify the closeListener multiple times.
|
|
|
|
Http2ConnectionHandler.this.closeListener = null;
|
|
|
|
try {
|
|
|
|
closeListener.operationComplete(future);
|
|
|
|
} catch (Exception e) {
|
|
|
|
throw new IllegalStateException("Close listener threw an unexpected exception", e);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
/**
|
2015-04-23 23:23:23 +02:00
|
|
|
* Close the remote endpoint with with a {@code GO_AWAY} frame. Does <strong>not</strong> flush
|
|
|
|
* immediately, this is the responsibility of the caller.
|
2015-01-20 01:48:11 +01:00
|
|
|
*/
|
2015-04-02 23:39:46 +02:00
|
|
|
private ChannelFuture goAway(ChannelHandlerContext ctx, Http2Exception cause) {
|
2015-01-20 01:48:11 +01:00
|
|
|
long errorCode = cause != null ? cause.error().code() : NO_ERROR.code();
|
|
|
|
ByteBuf debugData = Http2CodecUtil.toByteBuf(ctx, cause);
|
2015-03-20 02:36:24 +01:00
|
|
|
int lastKnownStream = connection().remote().lastStreamCreated();
|
2015-04-02 23:39:46 +02:00
|
|
|
return goAway(ctx, lastKnownStream, errorCode, debugData, ctx.newPromise());
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
/**
|
|
|
|
* Returns the client preface string if this is a client connection, otherwise returns {@code null}.
|
|
|
|
*/
|
|
|
|
private static ByteBuf clientPrefaceString(Http2Connection connection) {
|
|
|
|
return connection.isServer() ? connectionPrefaceBuf() : null;
|
|
|
|
}
|
|
|
|
|
2015-05-11 21:10:23 +02:00
|
|
|
private static void processGoAwayWriteResult(final ChannelHandlerContext ctx, final int lastStreamId,
|
|
|
|
final long errorCode, final ByteBuf debugData, ChannelFuture future) {
|
|
|
|
try {
|
|
|
|
if (future.isSuccess()) {
|
|
|
|
if (errorCode != NO_ERROR.code()) {
|
|
|
|
if (logger.isDebugEnabled()) {
|
|
|
|
logger.debug(
|
|
|
|
format("Sent GOAWAY: lastStreamId '%d', errorCode '%d', " +
|
|
|
|
"debugData '%s'. Forcing shutdown of the connection.",
|
|
|
|
lastStreamId, errorCode, debugData.toString(UTF_8)),
|
|
|
|
future.cause());
|
|
|
|
}
|
|
|
|
ctx.close();
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
if (logger.isErrorEnabled()) {
|
|
|
|
logger.error(
|
|
|
|
format("Sending GOAWAY failed: lastStreamId '%d', errorCode '%d', " +
|
|
|
|
"debugData '%s'. Forcing shutdown of the connection.",
|
|
|
|
lastStreamId, errorCode, debugData.toString(UTF_8)), future.cause());
|
|
|
|
}
|
|
|
|
ctx.close();
|
|
|
|
}
|
|
|
|
} finally {
|
|
|
|
// We're done with the debug data now.
|
|
|
|
debugData.release();
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2015-01-20 01:48:11 +01:00
|
|
|
/**
|
|
|
|
* Closes the channel when the future completes.
|
|
|
|
*/
|
|
|
|
private static final class ClosingChannelFutureListener implements ChannelFutureListener {
|
|
|
|
private final ChannelHandlerContext ctx;
|
|
|
|
private final ChannelPromise promise;
|
2015-08-18 19:36:16 +02:00
|
|
|
private final ScheduledFuture<?> timeoutTask;
|
2015-01-20 01:48:11 +01:00
|
|
|
|
|
|
|
ClosingChannelFutureListener(ChannelHandlerContext ctx, ChannelPromise promise) {
|
|
|
|
this.ctx = ctx;
|
|
|
|
this.promise = promise;
|
2015-08-18 19:36:16 +02:00
|
|
|
timeoutTask = null;
|
|
|
|
}
|
|
|
|
|
|
|
|
ClosingChannelFutureListener(final ChannelHandlerContext ctx, final ChannelPromise promise,
|
|
|
|
long timeout, TimeUnit unit) {
|
|
|
|
this.ctx = ctx;
|
|
|
|
this.promise = promise;
|
|
|
|
timeoutTask = ctx.executor().schedule(new OneTimeTask() {
|
|
|
|
@Override
|
|
|
|
public void run() {
|
|
|
|
ctx.close(promise);
|
|
|
|
}
|
|
|
|
}, timeout, unit);
|
2015-01-20 01:48:11 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
|
|
public void operationComplete(ChannelFuture sentGoAwayFuture) throws Exception {
|
2015-08-18 19:36:16 +02:00
|
|
|
if (timeoutTask != null) {
|
|
|
|
timeoutTask.cancel(false);
|
|
|
|
}
|
2015-01-20 01:48:11 +01:00
|
|
|
ctx.close(promise);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|