View Javadoc
1   /*
2    * Copyright 2014 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License, version 2.0 (the
5    * "License"); you may not use this file except in compliance with the License. You may obtain a
6    * copy of the License at:
7    *
8    * https://www.apache.org/licenses/LICENSE-2.0
9    *
10   * Unless required by applicable law or agreed to in writing, software distributed under the License
11   * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
12   * or implied. See the License for the specific language governing permissions and limitations under
13   * the License.
14   */
15  package io.netty.handler.codec.http2;
16  
17  import io.netty.buffer.ByteBuf;
18  import io.netty.channel.ChannelHandlerContext;
19  import io.netty.handler.codec.http.HttpHeaderNames;
20  import io.netty.handler.codec.http.HttpStatusClass;
21  import io.netty.handler.codec.http.HttpUtil;
22  import io.netty.handler.codec.http2.Http2Connection.Endpoint;
23  import io.netty.util.internal.logging.InternalLogger;
24  import io.netty.util.internal.logging.InternalLoggerFactory;
25  
26  import java.util.Iterator;
27  import java.util.List;
28  import java.util.Map.Entry;
29  
30  import static io.netty.handler.codec.http.HttpStatusClass.INFORMATIONAL;
31  import static io.netty.handler.codec.http2.Http2CodecUtil.DEFAULT_PRIORITY_WEIGHT;
32  import static io.netty.handler.codec.http2.Http2Error.INTERNAL_ERROR;
33  import static io.netty.handler.codec.http2.Http2Error.PROTOCOL_ERROR;
34  import static io.netty.handler.codec.http2.Http2Error.STREAM_CLOSED;
35  import static io.netty.handler.codec.http2.Http2Exception.connectionError;
36  import static io.netty.handler.codec.http2.Http2Exception.streamError;
37  import static io.netty.handler.codec.http2.Http2PromisedRequestVerifier.ALWAYS_VERIFY;
38  import static io.netty.handler.codec.http2.Http2Stream.State.CLOSED;
39  import static io.netty.handler.codec.http2.Http2Stream.State.HALF_CLOSED_REMOTE;
40  import static io.netty.util.internal.ObjectUtil.checkNotNull;
41  import static java.lang.Integer.MAX_VALUE;
42  import static java.lang.Math.min;
43  
44  /**
45   * Provides the default implementation for processing inbound frame events and delegates to a
46   * {@link Http2FrameListener}
47   * <p>
48   * This class will read HTTP/2 frames and delegate the events to a {@link Http2FrameListener}
49   * <p>
50   * This interface enforces inbound flow control functionality through
51   * {@link Http2LocalFlowController}
52   */
53  public class DefaultHttp2ConnectionDecoder implements Http2ConnectionDecoder {
54      private static final InternalLogger logger = InternalLoggerFactory.getInstance(DefaultHttp2ConnectionDecoder.class);
55      private Http2FrameListener internalFrameListener = new PrefaceFrameListener();
56      private final Http2Connection connection;
57      private Http2LifecycleManager lifecycleManager;
58      private final Http2ConnectionEncoder encoder;
59      private final Http2FrameReader frameReader;
60      private Http2FrameListener listener;
61      private final Http2PromisedRequestVerifier requestVerifier;
62      private final Http2SettingsReceivedConsumer settingsReceivedConsumer;
63      private final boolean autoAckPing;
64      private final Http2Connection.PropertyKey contentLengthKey;
65      private final boolean validateHeaders;
66  
67      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
68                                           Http2ConnectionEncoder encoder,
69                                           Http2FrameReader frameReader) {
70          this(connection, encoder, frameReader, ALWAYS_VERIFY);
71      }
72  
73      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
74                                           Http2ConnectionEncoder encoder,
75                                           Http2FrameReader frameReader,
76                                           Http2PromisedRequestVerifier requestVerifier) {
77          this(connection, encoder, frameReader, requestVerifier, true);
78      }
79  
80      /**
81       * Create a new instance.
82       * @param connection The {@link Http2Connection} associated with this decoder.
83       * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
84       * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
85       *                    h2 semantics on top of the frames.
86       * @param requestVerifier Determines if push promised streams are valid.
87       * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
88       *  The {@code Http2ConnectionEncoder} is expected to be an instance of {@link Http2SettingsReceivedConsumer} and
89       *  will apply the earliest received but not yet ACKed SETTINGS when writing the SETTINGS ACKs.
90       * {@code true} to enable automatically applying and sending settings acknowledge frame.
91       */
92      public DefaultHttp2ConnectionDecoder(Http2Connection connection,
93                                           Http2ConnectionEncoder encoder,
94                                           Http2FrameReader frameReader,
95                                           Http2PromisedRequestVerifier requestVerifier,
96                                           boolean autoAckSettings) {
97          this(connection, encoder, frameReader, requestVerifier, autoAckSettings, true);
98      }
99  
100     @Deprecated
101     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
102                                          Http2ConnectionEncoder encoder,
103                                          Http2FrameReader frameReader,
104                                          Http2PromisedRequestVerifier requestVerifier,
105                                          boolean autoAckSettings,
106                                          boolean autoAckPing) {
107         this(connection, encoder, frameReader, requestVerifier, autoAckSettings, true, true);
108     }
109 
110     /**
111      * Create a new instance.
112      * @param connection The {@link Http2Connection} associated with this decoder.
113      * @param encoder The {@link Http2ConnectionEncoder} associated with this decoder.
114      * @param frameReader Responsible for reading/parsing the raw frames. As opposed to this object which applies
115      *                    h2 semantics on top of the frames.
116      * @param requestVerifier Determines if push promised streams are valid.
117      * @param autoAckSettings {@code false} to disable automatically applying and sending settings acknowledge frame.
118      *                        The {@code Http2ConnectionEncoder} is expected to be an instance of
119      *                        {@link Http2SettingsReceivedConsumer} and will apply the earliest received but not yet
120      *                        ACKed SETTINGS when writing the SETTINGS ACKs. {@code true} to enable automatically
121      *                        applying and sending settings acknowledge frame.
122      * @param autoAckPing {@code false} to disable automatically sending ping acknowledge frame. {@code true} to enable
123      *                    automatically sending ping ack frame.
124      */
125     public DefaultHttp2ConnectionDecoder(Http2Connection connection,
126                                          Http2ConnectionEncoder encoder,
127                                          Http2FrameReader frameReader,
128                                          Http2PromisedRequestVerifier requestVerifier,
129                                          boolean autoAckSettings,
130                                          boolean autoAckPing,
131                                          boolean validateHeaders) {
132         this.validateHeaders = validateHeaders;
133         this.autoAckPing = autoAckPing;
134         if (autoAckSettings) {
135             settingsReceivedConsumer = null;
136         } else {
137             if (!(encoder instanceof Http2SettingsReceivedConsumer)) {
138                 throw new IllegalArgumentException("disabling autoAckSettings requires the encoder to be a " +
139                         Http2SettingsReceivedConsumer.class);
140             }
141             settingsReceivedConsumer = (Http2SettingsReceivedConsumer) encoder;
142         }
143         this.connection = checkNotNull(connection, "connection");
144         contentLengthKey = this.connection.newKey();
145         this.frameReader = checkNotNull(frameReader, "frameReader");
146         this.encoder = checkNotNull(encoder, "encoder");
147         this.requestVerifier = checkNotNull(requestVerifier, "requestVerifier");
148         if (connection.local().flowController() == null) {
149             connection.local().flowController(new DefaultHttp2LocalFlowController(connection));
150         }
151         connection.local().flowController().frameWriter(encoder.frameWriter());
152     }
153 
154     @Override
155     public void lifecycleManager(Http2LifecycleManager lifecycleManager) {
156         this.lifecycleManager = checkNotNull(lifecycleManager, "lifecycleManager");
157     }
158 
159     @Override
160     public Http2Connection connection() {
161         return connection;
162     }
163 
164     @Override
165     public final Http2LocalFlowController flowController() {
166         return connection.local().flowController();
167     }
168 
169     @Override
170     public void frameListener(Http2FrameListener listener) {
171         this.listener = checkNotNull(listener, "listener");
172     }
173 
174     @Override
175     public Http2FrameListener frameListener() {
176         return listener;
177     }
178 
179     @Override
180     public boolean prefaceReceived() {
181         return FrameReadListener.class == internalFrameListener.getClass();
182     }
183 
184     @Override
185     public void decodeFrame(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Http2Exception {
186         frameReader.readFrame(ctx, in, internalFrameListener);
187     }
188 
189     @Override
190     public Http2Settings localSettings() {
191         Http2Settings settings = new Http2Settings();
192         Http2FrameReader.Configuration config = frameReader.configuration();
193         Http2HeadersDecoder.Configuration headersConfig = config.headersConfiguration();
194         Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
195         settings.initialWindowSize(flowController().initialWindowSize());
196         settings.maxConcurrentStreams(connection.remote().maxActiveStreams());
197         settings.headerTableSize(headersConfig.maxHeaderTableSize());
198         settings.maxFrameSize(frameSizePolicy.maxFrameSize());
199         settings.maxHeaderListSize(headersConfig.maxHeaderListSize());
200         if (!connection.isServer()) {
201             // Only set the pushEnabled flag if this is a client endpoint.
202             settings.pushEnabled(connection.local().allowPushTo());
203         }
204         return settings;
205     }
206 
207     @Override
208     public void close() {
209         frameReader.close();
210     }
211 
212     /**
213      * Calculate the threshold in bytes which should trigger a {@code GO_AWAY} if a set of headers exceeds this amount.
214      * @param maxHeaderListSize
215      *      <a href="https://tools.ietf.org/html/rfc7540#section-6.5.2">SETTINGS_MAX_HEADER_LIST_SIZE</a> for the local
216      *      endpoint.
217      * @return the threshold in bytes which should trigger a {@code GO_AWAY} if a set of headers exceeds this amount.
218      */
219     protected long calculateMaxHeaderListSizeGoAway(long maxHeaderListSize) {
220         return Http2CodecUtil.calculateMaxHeaderListSizeGoAway(maxHeaderListSize);
221     }
222 
223     private int unconsumedBytes(Http2Stream stream) {
224         return flowController().unconsumedBytes(stream);
225     }
226 
227     void onGoAwayRead0(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
228             throws Http2Exception {
229         listener.onGoAwayRead(ctx, lastStreamId, errorCode, debugData);
230         connection.goAwayReceived(lastStreamId, errorCode, debugData);
231     }
232 
233     void onUnknownFrame0(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
234             ByteBuf payload) throws Http2Exception {
235         listener.onUnknownFrame(ctx, frameType, streamId, flags, payload);
236     }
237 
238     // See https://tools.ietf.org/html/rfc7540#section-8.1.2.6
239     private void verifyContentLength(Http2Stream stream, int data, boolean isEnd) throws Http2Exception {
240         ContentLength contentLength = stream.getProperty(contentLengthKey);
241         if (contentLength != null) {
242             try {
243                 contentLength.increaseReceivedBytes(connection.isServer(), stream.id(), data, isEnd);
244             } finally {
245                 if (isEnd) {
246                     stream.removeProperty(contentLengthKey);
247                 }
248             }
249         }
250     }
251 
252     /**
253      * Handles all inbound frames from the network.
254      */
255     private final class FrameReadListener implements Http2FrameListener {
256         @Override
257         public int onDataRead(final ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
258                               boolean endOfStream) throws Http2Exception {
259             Http2Stream stream = connection.stream(streamId);
260             Http2LocalFlowController flowController = flowController();
261             int readable = data.readableBytes();
262             int bytesToReturn = readable + padding;
263 
264             final boolean shouldIgnore;
265             try {
266                 shouldIgnore = shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "DATA");
267             } catch (Http2Exception e) {
268                 // Ignoring this frame. We still need to count the frame towards the connection flow control
269                 // window, but we immediately mark all bytes as consumed.
270                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
271                 flowController.consumeBytes(stream, bytesToReturn);
272                 throw e;
273             } catch (Throwable t) {
274                 throw connectionError(INTERNAL_ERROR, t, "Unhandled error on data stream id %d", streamId);
275             }
276 
277             if (shouldIgnore) {
278                 // Ignoring this frame. We still need to count the frame towards the connection flow control
279                 // window, but we immediately mark all bytes as consumed.
280                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
281                 flowController.consumeBytes(stream, bytesToReturn);
282 
283                 // Verify that the stream may have existed after we apply flow control.
284                 verifyStreamMayHaveExisted(streamId, endOfStream, "DATA");
285 
286                 // All bytes have been consumed.
287                 return bytesToReturn;
288             }
289             Http2Exception error = null;
290             switch (stream.state()) {
291                 case OPEN:
292                 case HALF_CLOSED_LOCAL:
293                     break;
294                 case HALF_CLOSED_REMOTE:
295                 case CLOSED:
296                     error = streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
297                         stream.id(), stream.state());
298                     break;
299                 default:
300                     error = streamError(stream.id(), PROTOCOL_ERROR,
301                         "Stream %d in unexpected state: %s", stream.id(), stream.state());
302                     break;
303             }
304 
305             int unconsumedBytes = unconsumedBytes(stream);
306             try {
307                 flowController.receiveFlowControlledFrame(stream, data, padding, endOfStream);
308                 // Update the unconsumed bytes after flow control is applied.
309                 unconsumedBytes = unconsumedBytes(stream);
310 
311                 // If the stream is in an invalid state to receive the frame, throw the error.
312                 if (error != null) {
313                     throw error;
314                 }
315 
316                 verifyContentLength(stream, readable, endOfStream);
317 
318                 // Call back the application and retrieve the number of bytes that have been
319                 // immediately processed.
320                 bytesToReturn = listener.onDataRead(ctx, streamId, data, padding, endOfStream);
321 
322                 if (endOfStream) {
323                     lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
324                 }
325 
326                 return bytesToReturn;
327             } catch (Http2Exception e) {
328                 // If an exception happened during delivery, the listener may have returned part
329                 // of the bytes before the error occurred. If that's the case, subtract that from
330                 // the total processed bytes so that we don't return too many bytes.
331                 int delta = unconsumedBytes - unconsumedBytes(stream);
332                 bytesToReturn -= delta;
333                 throw e;
334             } catch (RuntimeException e) {
335                 // If an exception happened during delivery, the listener may have returned part
336                 // of the bytes before the error occurred. If that's the case, subtract that from
337                 // the total processed bytes so that we don't return too many bytes.
338                 int delta = unconsumedBytes - unconsumedBytes(stream);
339                 bytesToReturn -= delta;
340                 throw e;
341             } finally {
342                 // If appropriate, return the processed bytes to the flow controller.
343                 flowController.consumeBytes(stream, bytesToReturn);
344             }
345         }
346 
347         @Override
348         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
349                 boolean endOfStream) throws Http2Exception {
350             onHeadersRead(ctx, streamId, headers, 0, DEFAULT_PRIORITY_WEIGHT, false, padding, endOfStream);
351         }
352 
353         @Override
354         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
355                 short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
356             Http2Stream stream = connection.stream(streamId);
357             boolean allowHalfClosedRemote = false;
358             boolean isTrailers = false;
359             if (stream == null && !connection.streamMayHaveExisted(streamId)) {
360                 stream = connection.remote().createStream(streamId, endOfStream);
361                 // Allow the state to be HALF_CLOSE_REMOTE if we're creating it in that state.
362                 allowHalfClosedRemote = stream.state() == HALF_CLOSED_REMOTE;
363             } else if (stream != null) {
364                 isTrailers = stream.isHeadersReceived();
365             }
366 
367             if (shouldIgnoreHeadersOrDataFrame(ctx, streamId, stream, endOfStream, "HEADERS")) {
368                 return;
369             }
370 
371             boolean isInformational = !connection.isServer() &&
372                     HttpStatusClass.valueOf(headers.status()) == INFORMATIONAL;
373             if ((isInformational || !endOfStream) && stream.isHeadersReceived() || stream.isTrailersReceived()) {
374                 throw streamError(streamId, PROTOCOL_ERROR,
375                                   "Stream %d received too many headers EOS: %s state: %s",
376                                   streamId, endOfStream, stream.state());
377             }
378 
379             switch (stream.state()) {
380                 case RESERVED_REMOTE:
381                     stream.open(endOfStream);
382                     break;
383                 case OPEN:
384                 case HALF_CLOSED_LOCAL:
385                     // Allowed to receive headers in these states.
386                     break;
387                 case HALF_CLOSED_REMOTE:
388                     if (!allowHalfClosedRemote) {
389                         throw streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
390                                 stream.id(), stream.state());
391                     }
392                     break;
393                 case CLOSED:
394                     throw streamError(stream.id(), STREAM_CLOSED, "Stream %d in unexpected state: %s",
395                             stream.id(), stream.state());
396                 default:
397                     // Connection error.
398                     throw connectionError(PROTOCOL_ERROR, "Stream %d in unexpected state: %s", stream.id(),
399                             stream.state());
400             }
401 
402             if (!isTrailers) {
403                 // extract the content-length header
404                 List<? extends CharSequence> contentLength = headers.getAll(HttpHeaderNames.CONTENT_LENGTH);
405                 if (contentLength != null && !contentLength.isEmpty()) {
406                     try {
407                         long cLength = HttpUtil.normalizeAndGetContentLength(contentLength, false, true);
408                         if (cLength != -1) {
409                             headers.setLong(HttpHeaderNames.CONTENT_LENGTH, cLength);
410                             stream.setProperty(contentLengthKey, new ContentLength(cLength));
411                         }
412                     } catch (IllegalArgumentException e) {
413                         throw streamError(stream.id(), PROTOCOL_ERROR, e,
414                                 "Multiple content-length headers received");
415                     }
416                 }
417                 // Use size() instead of isEmpty() for backward compatibility with grpc-java prior to 1.59.1,
418                 // see https://github.com/grpc/grpc-java/issues/10665
419             } else if (validateHeaders && headers.size() > 0) {
420                 // Need to check trailers don't contain pseudo headers. According to RFC 9113
421                 // Trailers MUST NOT include pseudo-header fields (Section 8.3).
422                 for (Iterator<Entry<CharSequence, CharSequence>> iterator =
423                     headers.iterator(); iterator.hasNext();) {
424                     CharSequence name = iterator.next().getKey();
425                     if (Http2Headers.PseudoHeaderName.hasPseudoHeaderFormat(name)) {
426                         throw streamError(stream.id(), PROTOCOL_ERROR,
427                                 "Found invalid Pseudo-Header in trailers: %s", name);
428                     }
429                 }
430             }
431 
432             stream.headersReceived(isInformational);
433             verifyContentLength(stream, 0, endOfStream);
434             encoder.flowController().updateDependencyTree(streamId, streamDependency, weight, exclusive);
435             listener.onHeadersRead(ctx, streamId, headers, streamDependency,
436                     weight, exclusive, padding, endOfStream);
437             // If the headers completes this stream, close it.
438             if (endOfStream) {
439                 lifecycleManager.closeStreamRemote(stream, ctx.newSucceededFuture());
440             }
441         }
442 
443         @Override
444         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
445                 boolean exclusive) throws Http2Exception {
446             encoder.flowController().updateDependencyTree(streamId, streamDependency, weight, exclusive);
447 
448             listener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
449         }
450 
451         @Override
452         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
453             Http2Stream stream = connection.stream(streamId);
454             if (stream == null) {
455                 verifyStreamMayHaveExisted(streamId, false, "RST_STREAM");
456                 return;
457             }
458 
459             switch(stream.state()) {
460             case IDLE:
461                 throw connectionError(PROTOCOL_ERROR, "RST_STREAM received for IDLE stream %d", streamId);
462             case CLOSED:
463                 return; // RST_STREAM frames must be ignored for closed streams.
464             default:
465                 break;
466             }
467 
468             listener.onRstStreamRead(ctx, streamId, errorCode);
469 
470             lifecycleManager.closeStream(stream, ctx.newSucceededFuture());
471         }
472 
473         @Override
474         public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
475             // Apply oldest outstanding local settings here. This is a synchronization point between endpoints.
476             Http2Settings settings = encoder.pollSentSettings();
477 
478             if (settings != null) {
479                 applyLocalSettings(settings);
480             }
481 
482             listener.onSettingsAckRead(ctx);
483         }
484 
485         /**
486          * Applies settings sent from the local endpoint.
487          * <p>
488          * This method is only called after the local settings have been acknowledged from the remote endpoint.
489          */
490         private void applyLocalSettings(Http2Settings settings) throws Http2Exception {
491             Boolean pushEnabled = settings.pushEnabled();
492             final Http2FrameReader.Configuration config = frameReader.configuration();
493             final Http2HeadersDecoder.Configuration headerConfig = config.headersConfiguration();
494             final Http2FrameSizePolicy frameSizePolicy = config.frameSizePolicy();
495             if (pushEnabled != null) {
496                 if (connection.isServer()) {
497                     throw connectionError(PROTOCOL_ERROR, "Server sending SETTINGS frame with ENABLE_PUSH specified");
498                 }
499                 connection.local().allowPushTo(pushEnabled);
500             }
501 
502             Long maxConcurrentStreams = settings.maxConcurrentStreams();
503             if (maxConcurrentStreams != null) {
504                 connection.remote().maxActiveStreams((int) min(maxConcurrentStreams, MAX_VALUE));
505             }
506 
507             Long headerTableSize = settings.headerTableSize();
508             if (headerTableSize != null) {
509                 headerConfig.maxHeaderTableSize(headerTableSize);
510             }
511 
512             Long maxHeaderListSize = settings.maxHeaderListSize();
513             if (maxHeaderListSize != null) {
514                 headerConfig.maxHeaderListSize(maxHeaderListSize, calculateMaxHeaderListSizeGoAway(maxHeaderListSize));
515             }
516 
517             Integer maxFrameSize = settings.maxFrameSize();
518             if (maxFrameSize != null) {
519                 frameSizePolicy.maxFrameSize(maxFrameSize);
520             }
521 
522             Integer initialWindowSize = settings.initialWindowSize();
523             if (initialWindowSize != null) {
524                 flowController().initialWindowSize(initialWindowSize);
525             }
526         }
527 
528         @Override
529         public void onSettingsRead(final ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
530             if (settingsReceivedConsumer == null) {
531                 // Acknowledge receipt of the settings. We should do this before we process the settings to ensure our
532                 // remote peer applies these settings before any subsequent frames that we may send which depend upon
533                 // these new settings. See https://github.com/netty/netty/issues/6520.
534                 encoder.writeSettingsAck(ctx, ctx.newPromise());
535 
536                 encoder.remoteSettings(settings);
537             } else {
538                 settingsReceivedConsumer.consumeReceivedSettings(settings);
539             }
540 
541             listener.onSettingsRead(ctx, settings);
542         }
543 
544         @Override
545         public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
546             if (autoAckPing) {
547                 // Send an ack back to the remote client.
548                 encoder.writePing(ctx, true, data, ctx.newPromise());
549             }
550             listener.onPingRead(ctx, data);
551         }
552 
553         @Override
554         public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
555             listener.onPingAckRead(ctx, data);
556         }
557 
558         @Override
559         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
560                 Http2Headers headers, int padding) throws Http2Exception {
561             // A client cannot push.
562             if (connection().isServer()) {
563                 throw connectionError(PROTOCOL_ERROR, "A client cannot push.");
564             }
565 
566             Http2Stream parentStream = connection.stream(streamId);
567 
568             if (shouldIgnoreHeadersOrDataFrame(ctx, streamId, parentStream, false, "PUSH_PROMISE")) {
569                 return;
570             }
571 
572             switch (parentStream.state()) {
573               case OPEN:
574               case HALF_CLOSED_LOCAL:
575                   // Allowed to receive push promise in these states.
576                   break;
577               default:
578                   // Connection error.
579                   throw connectionError(PROTOCOL_ERROR,
580                       "Stream %d in unexpected state for receiving push promise: %s",
581                       parentStream.id(), parentStream.state());
582             }
583 
584             if (!requestVerifier.isAuthoritative(ctx, headers)) {
585                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
586                         "Promised request on stream %d for promised stream %d is not authoritative",
587                         streamId, promisedStreamId);
588             }
589             if (!requestVerifier.isCacheable(headers)) {
590                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
591                         "Promised request on stream %d for promised stream %d is not known to be cacheable",
592                         streamId, promisedStreamId);
593             }
594             if (!requestVerifier.isSafe(headers)) {
595                 throw streamError(promisedStreamId, PROTOCOL_ERROR,
596                         "Promised request on stream %d for promised stream %d is not known to be safe",
597                         streamId, promisedStreamId);
598             }
599 
600             // Reserve the push stream based with a priority based on the current stream's priority.
601             connection.remote().reservePushStream(promisedStreamId, parentStream);
602 
603             listener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
604         }
605 
606         @Override
607         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
608                 throws Http2Exception {
609             onGoAwayRead0(ctx, lastStreamId, errorCode, debugData);
610         }
611 
612         @Override
613         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
614                 throws Http2Exception {
615             Http2Stream stream = connection.stream(streamId);
616             if (stream == null || stream.state() == CLOSED || streamCreatedAfterGoAwaySent(streamId)) {
617                 // Ignore this frame.
618                 verifyStreamMayHaveExisted(streamId, false, "WINDOW_UPDATE");
619                 return;
620             }
621 
622             // Update the outbound flow control window.
623             encoder.flowController().incrementWindowSize(stream, windowSizeIncrement);
624 
625             listener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
626         }
627 
628         @Override
629         public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
630                 ByteBuf payload) throws Http2Exception {
631             onUnknownFrame0(ctx, frameType, streamId, flags, payload);
632         }
633 
634         /**
635          * Helper method to determine if a frame that has the semantics of headers or data should be ignored for the
636          * {@code stream} (which may be {@code null}) associated with {@code streamId}.
637          */
638         private boolean shouldIgnoreHeadersOrDataFrame(ChannelHandlerContext ctx, int streamId, Http2Stream stream,
639                 boolean endOfStream, String frameName) throws Http2Exception {
640             if (stream == null) {
641                 if (streamCreatedAfterGoAwaySent(streamId)) {
642                     logger.info("{} ignoring {} frame for stream {}. Stream sent after GOAWAY sent",
643                             ctx.channel(), frameName, streamId);
644                     return true;
645                 }
646 
647                 // Make sure it's not an out-of-order frame, like a rogue DATA frame, for a stream that could
648                 // never have existed.
649                 verifyStreamMayHaveExisted(streamId, endOfStream, frameName);
650 
651                 // Its possible that this frame would result in stream ID out of order creation (PROTOCOL ERROR) and its
652                 // also possible that this frame is received on a CLOSED stream (STREAM_CLOSED after a RST_STREAM is
653                 // sent). We don't have enough information to know for sure, so we choose the lesser of the two errors.
654                 throw streamError(streamId, STREAM_CLOSED, "Received %s frame for an unknown stream %d",
655                                   frameName, streamId);
656             }
657             if (stream.isResetSent() || streamCreatedAfterGoAwaySent(streamId)) {
658                 // If we have sent a reset stream it is assumed the stream will be closed after the write completes.
659                 // If we have not sent a reset, but the stream was created after a GoAway this is not supported by
660                 // DefaultHttp2Connection and if a custom Http2Connection is used it is assumed the lifetime is managed
661                 // elsewhere so we don't close the stream or otherwise modify the stream's state.
662 
663                 if (logger.isInfoEnabled()) {
664                     logger.info("{} ignoring {} frame for stream {}", ctx.channel(), frameName,
665                             stream.isResetSent() ? "RST_STREAM sent." :
666                                     "Stream created after GOAWAY sent. Last known stream by peer " +
667                                      connection.remote().lastStreamKnownByPeer());
668                 }
669 
670                 return true;
671             }
672             return false;
673         }
674 
675         /**
676          * Helper method for determining whether or not to ignore inbound frames. A stream is considered to be created
677          * after a {@code GOAWAY} is sent if the following conditions hold:
678          * <p/>
679          * <ul>
680          *     <li>A {@code GOAWAY} must have been sent by the local endpoint</li>
681          *     <li>The {@code streamId} must identify a legitimate stream id for the remote endpoint to be creating</li>
682          *     <li>{@code streamId} is greater than the Last Known Stream ID which was sent by the local endpoint
683          *     in the last {@code GOAWAY} frame</li>
684          * </ul>
685          * <p/>
686          */
687         private boolean streamCreatedAfterGoAwaySent(int streamId) {
688             Endpoint<?> remote = connection.remote();
689             return connection.goAwaySent() && remote.isValidStreamId(streamId) &&
690                     streamId > remote.lastStreamKnownByPeer();
691         }
692 
693         private void verifyStreamMayHaveExisted(int streamId, boolean endOfStream, String frameName)
694                 throws Http2Exception {
695             if (!connection.streamMayHaveExisted(streamId)) {
696                 throw connectionError(PROTOCOL_ERROR,
697                         "Stream %d does not exist for inbound frame %s, endOfStream = %b",
698                         streamId, frameName, endOfStream);
699             }
700         }
701     }
702 
703     private final class PrefaceFrameListener implements Http2FrameListener {
704         /**
705          * Verifies that the HTTP/2 connection preface has been received from the remote endpoint.
706          * It is possible that the current call to
707          * {@link Http2FrameReader#readFrame(ChannelHandlerContext, ByteBuf, Http2FrameListener)} will have multiple
708          * frames to dispatch. So it may be OK for this class to get legitimate frames for the first readFrame.
709          */
710         private void verifyPrefaceReceived() throws Http2Exception {
711             if (!prefaceReceived()) {
712                 throw connectionError(PROTOCOL_ERROR, "Received non-SETTINGS as first frame.");
713             }
714         }
715 
716         @Override
717         public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding, boolean endOfStream)
718                 throws Http2Exception {
719             verifyPrefaceReceived();
720             return internalFrameListener.onDataRead(ctx, streamId, data, padding, endOfStream);
721         }
722 
723         @Override
724         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int padding,
725                 boolean endOfStream) throws Http2Exception {
726             verifyPrefaceReceived();
727             internalFrameListener.onHeadersRead(ctx, streamId, headers, padding, endOfStream);
728         }
729 
730         @Override
731         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers, int streamDependency,
732                 short weight, boolean exclusive, int padding, boolean endOfStream) throws Http2Exception {
733             verifyPrefaceReceived();
734             internalFrameListener.onHeadersRead(ctx, streamId, headers, streamDependency, weight,
735                     exclusive, padding, endOfStream);
736         }
737 
738         @Override
739         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency, short weight,
740                 boolean exclusive) throws Http2Exception {
741             verifyPrefaceReceived();
742             internalFrameListener.onPriorityRead(ctx, streamId, streamDependency, weight, exclusive);
743         }
744 
745         @Override
746         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) throws Http2Exception {
747             verifyPrefaceReceived();
748             internalFrameListener.onRstStreamRead(ctx, streamId, errorCode);
749         }
750 
751         @Override
752         public void onSettingsAckRead(ChannelHandlerContext ctx) throws Http2Exception {
753             verifyPrefaceReceived();
754             internalFrameListener.onSettingsAckRead(ctx);
755         }
756 
757         @Override
758         public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) throws Http2Exception {
759             // The first settings should change the internalFrameListener to the "real" listener
760             // that expects the preface to be verified.
761             if (!prefaceReceived()) {
762                 internalFrameListener = new FrameReadListener();
763             }
764             internalFrameListener.onSettingsRead(ctx, settings);
765         }
766 
767         @Override
768         public void onPingRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
769             verifyPrefaceReceived();
770             internalFrameListener.onPingRead(ctx, data);
771         }
772 
773         @Override
774         public void onPingAckRead(ChannelHandlerContext ctx, long data) throws Http2Exception {
775             verifyPrefaceReceived();
776             internalFrameListener.onPingAckRead(ctx, data);
777         }
778 
779         @Override
780         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
781                 Http2Headers headers, int padding) throws Http2Exception {
782             verifyPrefaceReceived();
783             internalFrameListener.onPushPromiseRead(ctx, streamId, promisedStreamId, headers, padding);
784         }
785 
786         @Override
787         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData)
788                 throws Http2Exception {
789             onGoAwayRead0(ctx, lastStreamId, errorCode, debugData);
790         }
791 
792         @Override
793         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement)
794                 throws Http2Exception {
795             verifyPrefaceReceived();
796             internalFrameListener.onWindowUpdateRead(ctx, streamId, windowSizeIncrement);
797         }
798 
799         @Override
800         public void onUnknownFrame(ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags,
801                 ByteBuf payload) throws Http2Exception {
802             onUnknownFrame0(ctx, frameType, streamId, flags, payload);
803         }
804     }
805 
806     private static final class ContentLength {
807         private final long expected;
808         private long seen;
809 
810         ContentLength(long expected) {
811             this.expected = expected;
812         }
813 
814         void increaseReceivedBytes(boolean server, int streamId, int bytes, boolean isEnd) throws Http2Exception {
815             seen += bytes;
816             // Check for overflow
817             if (seen < 0) {
818                 throw streamError(streamId, PROTOCOL_ERROR,
819                         "Received amount of data did overflow and so not match content-length header %d", expected);
820             }
821             // Check if we received more data then what was advertised via the content-length header.
822             if (seen > expected) {
823                 throw streamError(streamId, PROTOCOL_ERROR,
824                         "Received amount of data %d does not match content-length header %d", seen, expected);
825             }
826 
827             if (isEnd) {
828                 if (seen == 0 && !server) {
829                     // This may be a response to a HEAD request, let's just allow it.
830                     return;
831                 }
832 
833                 // Check that we really saw what was told via the content-length header.
834                 if (expected > seen) {
835                     throw streamError(streamId, PROTOCOL_ERROR,
836                             "Received amount of data %d does not match content-length header %d", seen, expected);
837                 }
838             }
839         }
840     }
841 }