/*
    * 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:
    *
    * https://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;
  import io.netty.handler.codec.http.HttpHeaderNames;
  import io.netty.handler.codec.http.HttpMethod;
  import io.netty.handler.codec.http.HttpStatusClass;
  import io.netty.handler.codec.http.HttpUtil;
  import io.netty.handler.codec.http2.Http2Connection.Endpoint;
  import io.netty.handler.codec.http2.Http2Headers.PseudoHeaderName;
  import io.netty.util.internal.logging.InternalLogger;
  import io.netty.util.internal.logging.InternalLoggerFactory;
  
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map.Entry;
  
  import static io.netty.handler.codec.http.HttpStatusClass.INFORMATIONAL;
  import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
  import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
  import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
  import static io.netty.handler.codec.http2.Http2Error.STREAM_CLOSED;
  import static io.netty.handler.codec.http2.Http2Exception.connectionError;
  import static io.netty.handler.codec.http2.Http2Exception.streamError;
  import static io.netty.handler.codec.http2.Http2PromisedRequestVerifier.ALWAYS_VERIFY;
  import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
  import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
  import static io.netty.util.internal.ObjectUtil.checkNotNull;
  import static java.lang.Integer.MAX_VALUE;
  import static java.lang.Math.min;
  
  /**
   * Provides the default implementation for processing inbound frame events and delegates to a
   * {@link Http2FrameListener}
   * <p>
   * This class will read HTTP/2 frames and delegate the events to a {@link Http2FrameListener}
   * <p>
   * This interface enforces inbound flow control functionality through
   * {@link Http2LocalFlowController}
   */
  public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
      private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2ConnectionDecoder.class);
      private Http2FrameListener internalFrameListener = new PrefaceFrameListener();
      private final Http2Connection connection;
      private Http2LifecycleManager lifecycleManager;
      private final Http2ConnectionEncoder encoder;
      private final Http2FrameReader frameReader;
      private Http2FrameListener listener;
      private final Http2PromisedRequestVerifier requestVerifier;
      private final Http2SettingsReceivedConsumer settingsReceivedConsumer;
      private final boolean autoAckPing;
      private final Http2Connection.PropertyKey contentLengthKey;
      private final boolean validateHeaders;
      private final boolean validateRequiredPseudoHeaders;
  
      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
                                           Http2ConnectionEncoder encoder,
                                           Http2FrameReader frameReader) {
          this(connection, encoder, frameReader, ALWAYS_VERIFY);
      }
  
      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
                                           Http2ConnectionEncoder encoder,
                                           Http2FrameReader frameReader,
                                           Http2PromisedRequestVerifier requestVerifier) {
          this(connection, encoder, frameReader, requestVerifier, true);
      }
  
      /**
       * Create a new instance.
       * @param connection The {@link Http2Connection} associated with this decoder.
       * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
       * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
       *                    h2 semantics on top of the frames.
       * @param requestVerifier Determines if push promised streams are valid.
       * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
       *  The {@code Http2ConnectionEncoder} is expected to be an instance of {@link Http2SettingsReceivedConsumer} and
       *  will apply the earliest received but not yet ACKed SETTINGS when writing the SETTINGS ACKs.
       * {@code true} to enable automatically applying and sending settings acknowledge frame.
       */
      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
                                           Http2ConnectionEncoder encoder,
                                           Http2FrameReader frameReader,
                                           Http2PromisedRequestVerifier requestVerifier,
                                           boolean autoAckSettings) {
         this(connection, encoder, frameReader, requestVerifier, autoAckSettings, true);
     }
 
     @Deprecated
     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
                                          Http2ConnectionEncoder encoder,
                                          Http2FrameReader frameReader,
                                          Http2PromisedRequestVerifier requestVerifier,
                                          boolean autoAckSettings,
                                          boolean autoAckPing) {
         this(connection, encoder, frameReader, requestVerifier, autoAckSettings, autoAckPing, true);
     }
 
     /**
      * Create a new instance.
      * @param connection The {@link Http2Connection} associated with this decoder.
      * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
      * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
      *                    h2 semantics on top of the frames.
      * @param requestVerifier Determines if push promised streams are valid.
      * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
      *                        The {@code Http2ConnectionEncoder} is expected to be an instance of
      *                        {@link Http2SettingsReceivedConsumer} and will apply the earliest received but not yet
      *                        ACKed SETTINGS when writing the SETTINGS ACKs. {@code true} to enable automatically
      *                        applying and sending settings acknowledge frame.
      * @param autoAckPing {@code false} to disable automatically sending ping acknowledge frame. {@code true} to enable
      *                    automatically sending ping ack frame.
      */
     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
                                          Http2ConnectionEncoder encoder,
                                          Http2FrameReader frameReader,
                                          Http2PromisedRequestVerifier requestVerifier,
                                          boolean autoAckSettings,
                                          boolean autoAckPing,
                                          boolean validateHeaders) {
         this(connection, encoder, frameReader, requestVerifier, autoAckSettings, autoAckPing, validateHeaders, false);
     }
 
     /**
      * Create a new instance.
      * @param connection The {@link Http2Connection} associated with this decoder.
      * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
      * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
      *                    h2 semantics on top of the frames.
      * @param requestVerifier Determines if push promised streams are valid.
      * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
      *                        The {@code Http2ConnectionEncoder} is expected to be an instance of
      *                        {@link Http2SettingsReceivedConsumer} and will apply the earliest received but not yet
      *                        ACKed SETTINGS when writing the SETTINGS ACKs. {@code true} to enable automatically
      *                        applying and sending settings acknowledge frame.
      * @param autoAckPing {@code false} to disable automatically sending ping acknowledge frame. {@code true} to enable
      *                    automatically sending ping ack frame.
      * @param validateHeaders {@code true} to validate headers according to
      *                        <a href="https://tools.ietf.org/html/rfc7540#section-8.1.2.6">RFC 7540, 8.1.2.6</a>.
      * @param validateRequiredPseudoHeaders {@code true} to reject request/response HEADERS that omit a mandatory
      *        pseudo-header field, according to
      *        <a href="https://www.rfc-editor.org/rfc/rfc9113.html#section-8.3">RFC 9113, 8.3</a>.
      */
     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
                                          Http2ConnectionEncoder encoder,
                                          Http2FrameReader frameReader,
                                          Http2PromisedRequestVerifier requestVerifier,
                                          boolean autoAckSettings,
                                          boolean autoAckPing,
                                          boolean validateHeaders,
                                          boolean validateRequiredPseudoHeaders) {
         this.validateHeaders = validateHeaders;
         this.validateRequiredPseudoHeaders = validateRequiredPseudoHeaders;
         this.autoAckPing = autoAckPing;
         if (autoAckSettings) {
             settingsReceivedConsumer = null;
         } else {
             if (!(encoder instanceof Http2SettingsReceivedConsumer)) {
                 throw new IllegalArgumentException("disabling autoAckSettings requires the encoder to be a " +
                         Http2SettingsReceivedConsumer.class);
             }
             settingsReceivedConsumer = (Http2SettingsReceivedConsumer) encoder;
         }
         this.connection = checkNotNull(connection, "connection");
         contentLengthKey = this.connection.newKey();
         this.frameReader = checkNotNull(frameReader, "frameReader");
         this.encoder = checkNotNull(encoder, "encoder");
         this.requestVerifier = checkNotNull(requestVerifier, "requestVerifier");
         if (connection.local().flowController() == null) {
             connection.local().flowController(new DefaultHttp2LocalFlowController(connection));
         }
         connection.local().flowController().frameWriter(encoder.frameWriter());
     }
 
     @Override
     public void lifecycleManager(Http2LifecycleManager lifecycleManager) {
         this.lifecycleManager = checkNotNull(lifecycleManager, "lifecycleManager");
     }
 
     @Override
     public Http2Connection connection() {
         return connection;
     }
 
     @Override
     public final Http2LocalFlowController flowController() {
         return connection.local().flowController();
     }
 
     @Override
     public void frameListener(Http2FrameListener listener) {
         this.listener = checkNotNull(listener, "listener");
     }
 
     @Override
     public Http2FrameListener frameListener() {
         return listener;
     }
 
     @Override
     public boolean prefaceReceived() {
         return FrameReadListener.class == internalFrameListener.getClass();
     }
 
     @Override
     public void decodeFrame(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Http2Exception {
         frameReader.readFrame(ctx, in, internalFrameListener);
     }
 
     @Override
     public Http2Settings localSettings() {
         Http2Settings settings = new Http2Settings();
         Http2FrameReader.Configuration config = frameReader.configuration();
         Http2HeadersDecoder.Configuration headersConfig = config.headersConfiguration();
         Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
         settings.initialWindowSize(flowController().initialWindowSize());
         settings.maxConcurrentStreams(connection.remote().maxActiveStreams());
         settings.headerTableSize(headersConfig.maxHeaderTableSize());
         settings.maxFrameSize(frameSizePolicy.maxFrameSize());
         settings.maxHeaderListSize(headersConfig.maxHeaderListSize());
         if (!connection.isServer()) {
             // Only set the pushEnabled flag if this is a client endpoint.
             settings.pushEnabled(connection.local().allowPushTo());
         }
         return settings;
     }
 
     @Override
     public void close() {
         frameReader.close();
     }
 
     /**
      * Calculate the threshold in bytes which should trigger a {@code GO_AWAY} if a set of headers exceeds this amount.
      * @param maxHeaderListSize
      *      <a href="https://tools.ietf.org/html/rfc7540#section-6.5.2">SETTINGS_MAX_HEADER_LIST_SIZE</a> for the local
      *      endpoint.
      * @return the threshold in bytes which should trigger a {@code GO_AWAY} if a set of headers exceeds this amount.
      */
     protected long calculateMaxHeaderListSizeGoAway(long maxHeaderListSize) {
         return Http2CodecUtil.calculateMaxHeaderListSizeGoAway(maxHeaderListSize);
     }
 
     private int unconsumedBytes(Http2Stream stream) {
         return flowController().unconsumedBytes(stream);
     }
 
     void onGoAwayRead0(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
             throws Http2Exception {
         listener.onGoAwayRead(ctx, lastStreamId, errorCode, debugData);
         connection.goAwayReceived(lastStreamId, errorCode, debugData);
     }
 
     // See https://tools.ietf.org/html/rfc7540#section-8.1.2.6
     private void verifyContentLength(Http2Stream stream, int data, boolean isEnd) throws Http2Exception {
         ContentLength contentLength = stream.getProperty(contentLengthKey);
         if (contentLength != null) {
             try {
                 contentLength.increaseReceivedBytes(connection.isServer(), stream.id(), data, isEnd);
             } finally {
                 if (isEnd) {
                     stream.removeProperty(contentLengthKey);
                 }
             }
         }
     }
 
     /**
      * Validates that an initial request or response HEADERS frame carries the mandatory pseudo-header fields,
      * as required by <a href="https://www.rfc-editor.org/rfc/rfc9113.html#section-8.3">RFC 9113, 8.3</a>.
      * Trailers and informational (1xx) responses are handled by the caller and do not reach this method.
      */
     private static void validateRequiredPseudoHeaders(boolean server, int streamId, Http2Headers headers)
             throws Http2Exception {
         if (server) {
             // Request pseudo-header fields (RFC 9113, 8.3.1).
             CharSequence method = headers.method();
             if (method == null) {
                 throw streamError(streamId, PROTOCOL_ERROR,
                         "Request is missing mandatory :method pseudo-header field.");
             }
             // CONNECT (RFC 9113, 8.5) omits :scheme/:path and carries :authority; extended CONNECT (RFC 8441),
             // identified by :protocol, follows the regular :scheme/:path rules.
             if (HttpMethod.CONNECT.asciiName().contentEquals(method) &&
                     !headers.contains(PseudoHeaderName.PROTOCOL.value())) {
                 if (headers.authority() == null) {
                     throw streamError(streamId, PROTOCOL_ERROR,
                             "CONNECT request is missing mandatory :authority pseudo-header field.");
                 }
             } else {
                 if (headers.scheme() == null) {
                     throw streamError(streamId, PROTOCOL_ERROR,
                             "Request is missing mandatory :scheme pseudo-header field.");
                 }
                 CharSequence path = headers.path();
                 if (path == null || path.length() == 0) {
                     throw streamError(streamId, PROTOCOL_ERROR,
                             "Request is missing mandatory :path pseudo-header field.");
                 }
             }
         } else {
             // Response pseudo-header fields (RFC 9113, 8.3.2).
             if (headers.status() == null) {
                 throw streamError(streamId, PROTOCOL_ERROR,
                         "Response is missing mandatory :status pseudo-header field.");
             }
         }
     }
 
     /**
      * Handles all inbound frames from the network.
      */
     private final class FrameReadListener implements Http2FrameListener {
         @Override
         public int onDataRead(final ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
                               boolean endOfStream) throws Http2Exception {
             Http2Stream stream = connection.stream(streamId);
             Http2LocalFlowController flowController = flowController();
             int readable = data.readableBytes();
             int bytesToReturn = readable + padding;
 
             final boolean shouldIgnore;
             try {
                 shouldIgnore = shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "DATA");
             } catch (Http2Exception e) {
                 // Ignoring this frame. We still need to count the frame towards the connection flow control
                 // window, but we immediately mark all bytes as consumed.
                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
                 flowController.consumeBytes(stream, bytesToReturn);
                 throw e;
             } catch (Throwable t) {
                 throw connectionError(INTERNAL_ERROR, t, "Unhandled error on data stream id %d", streamId);
             }
 
             if (shouldIgnore) {
                 // Ignoring this frame. We still need to count the frame towards the connection flow control
                 // window, but we immediately mark all bytes as consumed.
                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
                 flowController.consumeBytes(stream, bytesToReturn);
 
                 // Verify that the stream may have existed after we apply flow control.
                 verifyStreamMayHaveExisted(streamId, endOfStream, "DATA");
 
                 // All bytes have been consumed.
                 return bytesToReturn;
             }
             Http2Exception error = null;
             switch (stream.state()) {
                 case OPEN:
                 case HALF_CLOSED_LOCAL:
                     break;
                 case HALF_CLOSED_REMOTE:
                 case CLOSED:
                     error = streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
                         stream.id(), stream.state());
                     break;
                 default:
                     error = streamError(stream.id(), PROTOCOL_ERROR,
                         "Stream %d in unexpected state: %s", stream.id(), stream.state());
                     break;
             }
 
             int unconsumedBytes = unconsumedBytes(stream);
             try {
                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
                 // Update the unconsumed bytes after flow control is applied.
                 unconsumedBytes = unconsumedBytes(stream);
 
                 // If the stream is in an invalid state to receive the frame, throw the error.
                 if (error != null) {
                     throw error;
                 }
 
                 verifyContentLength(stream, readable, endOfStream);
 
                 // Call back the application and retrieve the number of bytes that have been
                 // immediately processed.
                 bytesToReturn = listener.onDataRead(ctx, streamId, data, padding, endOfStream);
 
                 if (endOfStream) {
                     lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
                 }
 
                 return bytesToReturn;
             } catch (Http2Exception e) {
                 // If an exception happened during delivery, the listener may have returned part
                 // of the bytes before the error occurred. If that's the case, subtract that from
                 // the total processed bytes so that we don't return too many bytes.
                 int delta = unconsumedBytes - unconsumedBytes(stream);
                 bytesToReturn -= delta;
                 throw e;
             } catch (RuntimeException e) {
                 // If an exception happened during delivery, the listener may have returned part
                 // of the bytes before the error occurred. If that's the case, subtract that from
                 // the total processed bytes so that we don't return too many bytes.
                 int delta = unconsumedBytes - unconsumedBytes(stream);
                 bytesToReturn -= delta;
                 throw e;
             } finally {
                 // If appropriate, return the processed bytes to the flow controller.
                 flowController.consumeBytes(stream, bytesToReturn);
             }
         }
 
         @Override
         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
                 boolean endOfStream) throws Http2Exception {
             onHeadersRead(ctx, streamId, headers, 0, DEFAULT_PRIORITY_WEIGHT, false, padding, endOfStream);
         }
 
         @Override
         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
                 short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
             Http2Stream stream = connection.stream(streamId);
             boolean allowHalfClosedRemote = false;
             boolean isTrailers = false;
             if (stream == null && !connection.streamMayHaveExisted(streamId)) {
                 stream = connection.remote().createStream(streamId, endOfStream);
                 // Allow the state to be HALF_CLOSE_REMOTE if we're creating it in that state.
                 allowHalfClosedRemote = stream.state() == HALF_CLOSED_REMOTE;
             } else if (stream != null) {
                 isTrailers = stream.isHeadersReceived();
             }
 
             if (shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "HEADERS")) {
                 return;
             }
 
             boolean isInformational = !connection.isServer() &&
                     HttpStatusClass.valueOf(headers.status()) == INFORMATIONAL;
             if ((isInformational || !endOfStream) && stream.isHeadersReceived() || stream.isTrailersReceived()) {
                 throw streamError(streamId, PROTOCOL_ERROR,
                                   "Stream %d received too many headers EOS: %s state: %s",
                                   streamId, endOfStream, stream.state());
             }
 
             switch (stream.state()) {
                 case RESERVED_REMOTE:
                     stream.open(endOfStream);
                     break;
                 case OPEN:
                 case HALF_CLOSED_LOCAL:
                     // Allowed to receive headers in these states.
                     break;
                 case HALF_CLOSED_REMOTE:
                     if (!allowHalfClosedRemote) {
                         throw streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
                                 stream.id(), stream.state());
                     }
                     break;
                 case CLOSED:
                     throw streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
                             stream.id(), stream.state());
                 default:
                     // Connection error.
                     throw connectionError(PROTOCOL_ERROR, "Stream %d in unexpected state: %s", stream.id(),
                             stream.state());
             }
 
             if (!isTrailers) {
                 if (validateRequiredPseudoHeaders && !isInformational) {
                     // Reject initial request/response HEADERS that omit a mandatory pseudo-header (RFC 9113, 8.3).
                     validateRequiredPseudoHeaders(connection.isServer(), stream.id(), headers);
                 }
                 // extract the content-length header
                 List<? extends CharSequence> contentLength = headers.getAll(HttpHeaderNames.CONTENT_LENGTH);
                 if (contentLength != null && !contentLength.isEmpty()) {
                     try {
                         long cLength = HttpUtil.normalizeAndGetContentLength(contentLength, false, true);
                         if (cLength != -1) {
                             headers.setLong(HttpHeaderNames.CONTENT_LENGTH, cLength);
                             stream.setProperty(contentLengthKey, new ContentLength(cLength));
                         }
                     } catch (IllegalArgumentException e) {
                         throw streamError(stream.id(), PROTOCOL_ERROR, e,
                                 "Multiple content-length headers received");
                     }
                 }
                 // Use size() instead of isEmpty() for backward compatibility with grpc-java prior to 1.59.1,
                 // see https://github.com/grpc/grpc-java/issues/10665
             } else if (validateHeaders && headers.size() > 0) {
                 // Need to check trailers don't contain pseudo headers. According to RFC 9113
                 // Trailers MUST NOT include pseudo-header fields (Section 8.3).
                 for (Iterator<Entry<CharSequence, CharSequence>> iterator =
                     headers.iterator(); iterator.hasNext();) {
                     CharSequence name = iterator.next().getKey();
                     if (Http2Headers.PseudoHeaderName.hasPseudoHeaderFormat(name)) {
                         throw streamError(stream.id(), PROTOCOL_ERROR,
                                 "Found invalid Pseudo-Header in trailers: %s", name);
                     }
                 }
             }
 
             stream.headersReceived(isInformational);
             verifyContentLength(stream, 0, endOfStream);
             encoder.flowController().updateDependencyTree(streamId, streamDependency, weight, exclusive);
             listener.onHeadersRead(ctx, streamId, headers, streamDependency,
                     weight, exclusive, padding, endOfStream);
             // If the headers completes this stream, close it.
             if (endOfStream) {
                 lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
             }
         }
 
         @Override
         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
                 boolean exclusive) throws Http2Exception {
             encoder.flowController().updateDependencyTree(streamId, streamDependency, weight, exclusive);
 
             listener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
         }
 
         @Override
         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
             Http2Stream stream = connection.stream(streamId);
             if (stream == null) {
                 verifyStreamMayHaveExisted(streamId, false, "RST_STREAM");
                 return;
             }
 
             switch(stream.state()) {
             case IDLE:
                 throw connectionError(PROTOCOL_ERROR, "RST_STREAM received for IDLE stream %d", streamId);
             case CLOSED:
                 return; // RST_STREAM frames must be ignored for closed streams.
             default:
                 break;
             }
 
             listener.onRstStreamRead(ctx, streamId, errorCode);
 
             lifecycleManager.closeStream(stream, ctx.newSucceededFuture());
         }
 
         @Override
         public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
             // Apply oldest outstanding local settings here. This is a synchronization point between endpoints.
             Http2Settings settings = encoder.pollSentSettings();
 
             if (settings != null) {
                 applyLocalSettings(settings);
             }
 
             listener.onSettingsAckRead(ctx);
         }
 
         /**
          * Applies settings sent from the local endpoint.
          * <p>
          * This method is only called after the local settings have been acknowledged from the remote endpoint.
          */
         private void applyLocalSettings(Http2Settings settings) throws Http2Exception {
             Boolean pushEnabled = settings.pushEnabled();
             final Http2FrameReader.Configuration config = frameReader.configuration();
             final Http2HeadersDecoder.Configuration headerConfig = config.headersConfiguration();
             final Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
             if (pushEnabled != null) {
                 if (connection.isServer()) {
                     throw connectionError(PROTOCOL_ERROR, "Server sending SETTINGS frame with ENABLE_PUSH specified");
                 }
                 connection.local().allowPushTo(pushEnabled);
             }
 
             Long maxConcurrentStreams = settings.maxConcurrentStreams();
             if (maxConcurrentStreams != null) {
                 connection.remote().maxActiveStreams((int) min(maxConcurrentStreams, MAX_VALUE));
             }
 
             Long headerTableSize = settings.headerTableSize();
             if (headerTableSize != null) {
                 headerConfig.maxHeaderTableSize(headerTableSize);
             }
 
             Long maxHeaderListSize = settings.maxHeaderListSize();
             if (maxHeaderListSize != null) {
                 headerConfig.maxHeaderListSize(maxHeaderListSize, calculateMaxHeaderListSizeGoAway(maxHeaderListSize));
             }
 
             Integer maxFrameSize = settings.maxFrameSize();
             if (maxFrameSize != null) {
                 frameSizePolicy.maxFrameSize(maxFrameSize);
             }
 
             Integer initialWindowSize = settings.initialWindowSize();
             if (initialWindowSize != null) {
                 flowController().initialWindowSize(initialWindowSize);
             }
         }
 
         @Override
         public void onSettingsRead(final ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
             if (settingsReceivedConsumer == null) {
                 // Acknowledge receipt of the settings. We should do this before we process the settings to ensure our
                 // remote peer applies these settings before any subsequent frames that we may send which depend upon
                 // these new settings. See https://github.com/netty/netty/issues/6520.
                 encoder.writeSettingsAck(ctx, ctx.newPromise());
 
                 encoder.remoteSettings(settings);
             } else {
                 settingsReceivedConsumer.consumeReceivedSettings(settings);
             }
 
             listener.onSettingsRead(ctx, settings);
         }
 
         @Override
         public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
             if (autoAckPing) {
                 // Send an ack back to the remote client.
                 encoder.writePing(ctx, true, data, ctx.newPromise());
             }
             listener.onPingRead(ctx, data);
         }
 
         @Override
         public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
             listener.onPingAckRead(ctx, data);
         }
 
         @Override
         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
                 Http2Headers headers, int padding) throws Http2Exception {
             // A client cannot push.
             if (connection().isServer()) {
                 throw connectionError(PROTOCOL_ERROR, "A client cannot push.");
             }
 
             Http2Stream parentStream = connection.stream(streamId);
 
             if (shouldIgnoreHeadersOrDataFrame(ctx, streamId, parentStream, false, "PUSH_PROMISE")) {
                 return;
             }
 
             switch (parentStream.state()) {
               case OPEN:
               case HALF_CLOSED_LOCAL:
                   // Allowed to receive push promise in these states.
                   break;
               default:
                   // Connection error.
                   throw connectionError(PROTOCOL_ERROR,
                       "Stream %d in unexpected state for receiving push promise: %s",
                       parentStream.id(), parentStream.state());
             }
 
             if (!requestVerifier.isAuthoritative(ctx, headers)) {
                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
                         "Promised request on stream %d for promised stream %d is not authoritative",
                         streamId, promisedStreamId);
             }
             if (!requestVerifier.isCacheable(headers)) {
                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
                         "Promised request on stream %d for promised stream %d is not known to be cacheable",
                         streamId, promisedStreamId);
             }
             if (!requestVerifier.isSafe(headers)) {
                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
                         "Promised request on stream %d for promised stream %d is not known to be safe",
                         streamId, promisedStreamId);
             }
 
             // Reserve the push stream based with a priority based on the current stream's priority.
             connection.remote().reservePushStream(promisedStreamId, parentStream);
 
             listener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
         }
 
         @Override
         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
                 throws Http2Exception {
             onGoAwayRead0(ctx, lastStreamId, errorCode, debugData);
         }
 
         @Override
         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
                 throws Http2Exception {
             Http2Stream stream = connection.stream(streamId);
             if (stream == null || stream.state() == CLOSED || streamCreatedAfterGoAwaySent(streamId)) {
                 // Ignore this frame.
                 verifyStreamMayHaveExisted(streamId, false, "WINDOW_UPDATE");
                 return;
             }
 
             // Update the outbound flow control window.
             encoder.flowController().incrementWindowSize(stream, windowSizeIncrement);
 
             listener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
         }
 
         @Override
         public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
                 ByteBuf payload) throws Http2Exception {
             Http2Stream stream = connection.stream(streamId);
             if (stream == null) {
                 return;
             }
 
             listener.onUnknownFrame(ctx, frameType, streamId, flags, payload);
         }
 
         /**
          * Helper method to determine if a frame that has the semantics of headers or data should be ignored for the
          * {@code stream} (which may be {@code null}) associated with {@code streamId}.
          */
         private boolean shouldIgnoreHeadersOrDataFrame(ChannelHandlerContext ctx, int streamId, Http2Stream stream,
                 boolean endOfStream, String frameName) throws Http2Exception {
             if (stream == null) {
                 if (streamCreatedAfterGoAwaySent(streamId)) {
                     logger.info("{} ignoring {} frame for stream {}. Stream sent after GOAWAY sent",
                             ctx.channel(), frameName, streamId);
                     return true;
                 }
 
                 // Make sure it's not an out-of-order frame, like a rogue DATA frame, for a stream that could
                 // never have existed.
                 verifyStreamMayHaveExisted(streamId, endOfStream, frameName);
 
                 // Its possible that this frame would result in stream ID out of order creation (PROTOCOL ERROR) and its
                 // also possible that this frame is received on a CLOSED stream (STREAM_CLOSED after a RST_STREAM is
                 // sent). We don't have enough information to know for sure, so we choose the lesser of the two errors.
                 throw streamError(streamId, STREAM_CLOSED, "Received %s frame for an unknown stream %d",
                                   frameName, streamId);
             }
             if (stream.isResetSent() || streamCreatedAfterGoAwaySent(streamId)) {
                 // If we have sent a reset stream it is assumed the stream will be closed after the write completes.
                 // If we have not sent a reset, but the stream was created after a GoAway this is not supported by
                 // DefaultHttp2Connection and if a custom Http2Connection is used it is assumed the lifetime is managed
                 // elsewhere so we don't close the stream or otherwise modify the stream's state.
 
                 if (logger.isInfoEnabled()) {
                     logger.info("{} ignoring {} frame for stream {}", ctx.channel(), frameName,
                             stream.isResetSent() ? "RST_STREAM sent." :
                                     "Stream created after GOAWAY sent. Last known stream by peer " +
                                      connection.remote().lastStreamKnownByPeer());
                 }
 
                 return true;
             }
             return false;
         }
 
         /**
          * Helper method for determining whether or not to ignore inbound frames. A stream is considered to be created
          * after a {@code GOAWAY} is sent if the following conditions hold:
          * <p/>
          * <ul>
          *     <li>A {@code GOAWAY} must have been sent by the local endpoint</li>
          *     <li>The {@code streamId} must identify a legitimate stream id for the remote endpoint to be creating</li>
          *     <li>{@code streamId} is greater than the Last Known Stream ID which was sent by the local endpoint
          *     in the last {@code GOAWAY} frame</li>
          * </ul>
          * <p/>
          */
         private boolean streamCreatedAfterGoAwaySent(int streamId) {
             Endpoint<?> remote = connection.remote();
             return connection.goAwaySent() && remote.isValidStreamId(streamId) &&
                     streamId > remote.lastStreamKnownByPeer();
         }
 
         private void verifyStreamMayHaveExisted(int streamId, boolean endOfStream, String frameName)
                 throws Http2Exception {
             if (!connection.streamMayHaveExisted(streamId)) {
                 throw connectionError(PROTOCOL_ERROR,
                         "Stream %d does not exist for inbound frame %s, endOfStream = %b",
                         streamId, frameName, endOfStream);
             }
         }
     }
 
     private final class PrefaceFrameListener implements Http2FrameListener {
         /**
          * Verifies that the HTTP/2 connection preface has been received from the remote endpoint.
          * It is possible that the current call to
          * {@link Http2FrameReader#readFrame(ChannelHandlerContext, ByteBuf, Http2FrameListener)} will have multiple
          * frames to dispatch. So it may be OK for this class to get legitimate frames for the first readFrame.
          */
         private void verifyPrefaceReceived() throws Http2Exception {
             if (!prefaceReceived()) {
                 throw connectionError(PROTOCOL_ERROR, "Received non-SETTINGS as first frame.");
             }
         }
 
         @Override
         public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
                 throws Http2Exception {
             verifyPrefaceReceived();
             return internalFrameListener.onDataRead(ctx, streamId, data, padding, endOfStream);
         }
 
         @Override
         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
                 boolean endOfStream) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onHeadersRead(ctx, streamId, headers, padding, endOfStream);
         }
 
         @Override
         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
                 short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onHeadersRead(ctx, streamId, headers, streamDependency, weight,
                     exclusive, padding, endOfStream);
         }
 
         @Override
         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
                 boolean exclusive) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
         }
 
         @Override
         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onRstStreamRead(ctx, streamId, errorCode);
         }
 
         @Override
         public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onSettingsAckRead(ctx);
         }
 
         @Override
         public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
             // The first settings should change the internalFrameListener to the "real" listener
             // that expects the preface to be verified.
             if (!prefaceReceived()) {
                 internalFrameListener = new FrameReadListener();
             }
             internalFrameListener.onSettingsRead(ctx, settings);
         }
 
         @Override
         public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onPingRead(ctx, data);
         }
 
         @Override
         public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onPingAckRead(ctx, data);
         }
 
         @Override
         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
                 Http2Headers headers, int padding) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
         }
 
         @Override
         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
                 throws Http2Exception {
             onGoAwayRead0(ctx, lastStreamId, errorCode, debugData);
         }
 
         @Override
         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
                 throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
         }
 
         @Override
         public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
                 ByteBuf payload) throws Http2Exception {
             verifyPrefaceReceived();
             internalFrameListener.onUnknownFrame(ctx, frameType, streamId, flags, payload);
         }
     }
 
     private static final class ContentLength {
         private final long expected;
         private long seen;
 
         ContentLength(long expected) {
             this.expected = expected;
         }
 
         void increaseReceivedBytes(boolean server, int streamId, int bytes, boolean isEnd) throws Http2Exception {
             seen += bytes;
             // Check for overflow
             if (seen < 0) {
                 throw streamError(streamId, PROTOCOL_ERROR,
                         "Received amount of data did overflow and so not match content-length header %d", expected);
             }
             // Check if we received more data then what was advertised via the content-length header.
             if (seen > expected) {
                 throw streamError(streamId, PROTOCOL_ERROR,
                         "Received amount of data %d does not match content-length header %d", seen, expected);
             }
 
             if (isEnd) {
                 if (seen == 0 && !server) {
                     // This may be a response to a HEAD request, let's just allow it.
                     return;
                 }
 
                 // Check that we really saw what was told via the content-length header.
                 if (expected > seen) {
                     throw streamError(streamId, PROTOCOL_ERROR,
                             "Received amount of data %d does not match content-length header %d", seen, expected);
                 }
             }
         }
     }
 }