View Javadoc
1   /*
2    * Copyright 2016 The Netty Project
3    *
4    * The Netty Project licenses this file to you under the Apache License,
5    * version 2.0 (the "License"); you may not use this file except in compliance
6    * with the License. You may obtain a 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
11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13   * License for the specific language governing permissions and limitations
14   * under the License.
15   */
16  package io.netty.handler.codec.http2;
17  
18  import io.netty.buffer.ByteBuf;
19  import io.netty.channel.Channel;
20  import io.netty.channel.ChannelFuture;
21  import io.netty.channel.ChannelFutureListener;
22  import io.netty.channel.ChannelHandlerContext;
23  import io.netty.channel.ChannelInboundHandler;
24  import io.netty.channel.ChannelPromise;
25  import io.netty.handler.codec.UnsupportedMessageTypeException;
26  import io.netty.handler.codec.http.HttpServerUpgradeHandler.UpgradeEvent;
27  import io.netty.handler.codec.http2.Http2Connection.PropertyKey;
28  import io.netty.handler.codec.http2.Http2Stream.State;
29  import io.netty.handler.codec.http2.StreamBufferingEncoder.Http2ChannelClosedException;
30  import io.netty.handler.codec.http2.StreamBufferingEncoder.Http2GoAwayException;
31  import io.netty.util.ReferenceCountUtil;
32  import io.netty.util.ReferenceCounted;
33  import io.netty.util.collection.IntObjectHashMap;
34  import io.netty.util.collection.IntObjectMap;
35  import io.netty.util.internal.logging.InternalLogger;
36  import io.netty.util.internal.logging.InternalLoggerFactory;
37  
38  import static io.netty.buffer.ByteBufUtil.writeAscii;
39  import static io.netty.handler.codec.http2.Http2CodecUtil.HTTP_UPGRADE_STREAM_ID;
40  import static io.netty.handler.codec.http2.Http2CodecUtil.isStreamIdValid;
41  import static io.netty.handler.codec.http2.Http2Error.NO_ERROR;
42  import static io.netty.util.internal.logging.InternalLogLevel.DEBUG;
43  
44  /**
45   * <p>An HTTP/2 handler that maps HTTP/2 frames to {@link Http2Frame} objects and vice versa. For every incoming HTTP/2
46   * frame, an {@link Http2Frame} object is created and propagated via {@link #channelRead}. Outbound {@link Http2Frame}
47   * objects received via {@link #write} are converted to the HTTP/2 wire format. HTTP/2 frames specific to a stream
48   * implement the {@link Http2StreamFrame} interface. The {@link Http2FrameCodec} is instantiated using the
49   * {@link Http2FrameCodecBuilder}. It's recommended for channel handlers to inherit from the
50   * {@link Http2ChannelDuplexHandler}, as it provides additional functionality like iterating over all active streams or
51   * creating outbound streams.
52   *
53   * <h3>Stream Lifecycle</h3>
54   * <p>
55   * The frame codec delivers and writes frames for active streams. An active stream is closed when either side sends a
56   * {@code RST_STREAM} frame or both sides send a frame with the {@code END_STREAM} flag set. Each
57   * {@link Http2StreamFrame} has a {@link Http2FrameStream} object attached that uniquely identifies a particular stream.
58   *
59   * <p>{@link Http2StreamFrame}s read from the channel always a {@link Http2FrameStream} object set, while when writing a
60   * {@link Http2StreamFrame} the application code needs to set a {@link Http2FrameStream} object using
61   * {@link Http2StreamFrame#stream(Http2FrameStream)}.
62   *
63   * <h3>Flow control</h3>
64   * <p>
65   * The frame codec automatically increments stream and connection flow control windows.
66   *
67   * <p>Incoming flow controlled frames need to be consumed by writing a {@link Http2WindowUpdateFrame} with the consumed
68   * number of bytes and the corresponding stream identifier set to the frame codec.
69   *
70   * <p>The local stream-level flow control window can be changed by writing a {@link Http2SettingsFrame} with the
71   * {@link Http2Settings#initialWindowSize()} set to the targeted value.
72   *
73   * <p>The connection-level flow control window can be changed by writing a {@link Http2WindowUpdateFrame} with the
74   * desired window size <em>increment</em> in bytes and the stream identifier set to {@code 0}. By default the initial
75   * connection-level flow control window is the same as initial stream-level flow control window.
76   *
77   * <h3>New inbound Streams</h3>
78   * <p>
79   * The first frame of an HTTP/2 stream must be an {@link Http2HeadersFrame}, which will have an {@link Http2FrameStream}
80   * object attached.
81   *
82   * <h3>New outbound Streams</h3>
83   * <p>
84   * A outbound HTTP/2 stream can be created by first instantiating a new {@link Http2FrameStream} object via
85   * {@link Http2ChannelDuplexHandler#newStream()}, and then writing a {@link Http2HeadersFrame} object with the stream
86   * attached.
87   *
88   * <pre> {@code
89   *     final Http2Stream2 stream = handler.newStream();
90   *     ctx.write(headersFrame.stream(stream)).addListener(new ChannelFutureListener() {
91   *
92   *         @Override
93   *         public void operationComplete(ChannelFuture f) {
94   *             if (f.isSuccess()) {
95   *                 // Stream is active and stream.id() returns a valid stream identifier.
96   *                 System.out.println("New stream with id " + stream.id() + " created.");
97   *             } else {
98   *                 // Stream failed to become active. Handle error.
99   *                 if (f.cause() instanceof Http2NoMoreStreamIdsException) {
100  *
101  *                 } else if (f.cause() instanceof Http2GoAwayException) {
102  *
103  *                 } else {
104  *
105  *                 }
106  *             }
107  *         }
108  *     }
109  *     }
110  * </pre>
111  *
112  * <p>If a new stream cannot be created due to stream id exhaustion of the endpoint, the {@link ChannelPromise} of the
113  * HEADERS frame will fail with a {@link Http2NoMoreStreamIdsException}.
114  *
115  * <p>The HTTP/2 standard allows for an endpoint to limit the maximum number of concurrently active streams via the
116  * {@code SETTINGS_MAX_CONCURRENT_STREAMS} setting. When this limit is reached, no new streams can be created. However,
117  * the {@link Http2FrameCodec} can be build with
118  * {@link Http2FrameCodecBuilder#encoderEnforceMaxConcurrentStreams(boolean)} enabled, in which case a new stream and
119  * its associated frames will be buffered until either the limit is increased or an active stream is closed. It's,
120  * however, possible that a buffered stream will never become active. That is, the channel might
121  * get closed or a GO_AWAY frame might be received. In the first case, all writes of buffered streams will fail with a
122  * {@link Http2ChannelClosedException}. In the second case, all writes of buffered streams with an identifier less than
123  * the last stream identifier of the GO_AWAY frame will fail with a {@link Http2GoAwayException}.
124  *
125  * <h3>Error Handling</h3>
126  * <p>
127  * Exceptions and errors are propagated via {@link ChannelInboundHandler#exceptionCaught}. Exceptions that apply to
128  * a specific HTTP/2 stream are wrapped in a {@link Http2FrameStreamException} and have the corresponding
129  * {@link Http2FrameStream} object attached.
130  *
131  * <h3>Reference Counting</h3>
132  * <p>
133  * Some {@link Http2StreamFrame}s implement the {@link ReferenceCounted} interface, as they carry
134  * reference counted objects (e.g. {@link ByteBuf}s). The frame codec will call {@link ReferenceCounted#retain()} before
135  * propagating a reference counted object through the pipeline, and thus an application handler needs to release such
136  * an object after having consumed it. For more information on reference counting take a look at
137  * <a href="https://netty.io/wiki/reference-counted-objects.html">Reference counted objects</a>
138  *
139  * <h3>HTTP Upgrade</h3>
140  * <p>
141  * Server-side HTTP to HTTP/2 upgrade is supported in conjunction with {@link Http2ServerUpgradeCodec}; the necessary
142  * HTTP-to-HTTP/2 conversion is performed automatically.
143  */
144 public class Http2FrameCodec extends Http2ConnectionHandler {
145 
146     private static final InternalLogger LOG = InternalLoggerFactory.getInstance(Http2FrameCodec.class);
147 
148     private static final Class<?>[] SUPPORTED_MESSAGES = new Class[] {
149             Http2DataFrame.class, Http2HeadersFrame.class, Http2WindowUpdateFrame.class, Http2ResetFrame.class,
150             Http2PingFrame.class, Http2SettingsFrame.class, Http2SettingsAckFrame.class, Http2GoAwayFrame.class,
151             Http2PushPromiseFrame.class, Http2PriorityFrame.class, Http2UnknownFrame.class };
152 
153     protected final PropertyKey streamKey;
154     private final PropertyKey upgradeKey;
155 
156     private final Integer initialFlowControlWindowSize;
157 
158     ChannelHandlerContext ctx;
159 
160     /**
161      * Number of buffered streams if the {@link StreamBufferingEncoder} is used.
162      **/
163     private int numBufferedStreams;
164     private final IntObjectMap<DefaultHttp2FrameStream> frameStreamToInitializeMap =
165             new IntObjectHashMap<DefaultHttp2FrameStream>(8);
166 
167     protected Http2FrameCodec(Http2ConnectionEncoder encoder, Http2ConnectionDecoder decoder,
168                               Http2Settings initialSettings, boolean decoupleCloseAndGoAway, boolean flushPreface) {
169         super(decoder, encoder, initialSettings, decoupleCloseAndGoAway, flushPreface);
170 
171         decoder.frameListener(new FrameListener());
172         connection().addListener(new ConnectionListener());
173         connection().remote().flowController().listener(new Http2RemoteFlowControllerListener());
174         streamKey = connection().newKey();
175         upgradeKey = connection().newKey();
176         initialFlowControlWindowSize = initialSettings.initialWindowSize();
177     }
178 
179     /**
180      * Creates a new outbound/local stream.
181      */
182     DefaultHttp2FrameStream newStream() {
183         return new DefaultHttp2FrameStream();
184     }
185 
186     /**
187      * Iterates over all active HTTP/2 streams.
188      *
189      * <p>This method must not be called outside of the event loop.
190      */
191     final void forEachActiveStream(final Http2FrameStreamVisitor streamVisitor) throws Http2Exception {
192         assert ctx.executor().inEventLoop();
193         if (connection().numActiveStreams() > 0) {
194             connection().forEachActiveStream(new Http2StreamVisitor() {
195                 @Override
196                 public boolean visit(Http2Stream stream) {
197                     try {
198                         return streamVisitor.visit((Http2FrameStream) stream.getProperty(streamKey));
199                     } catch (Throwable cause) {
200                         onError(ctx, false, cause);
201                         return false;
202                     }
203                 }
204             });
205         }
206     }
207 
208     /**
209      * Retrieve the number of streams currently in the process of being initialized.
210      * <p>
211      * This is package-private for testing only.
212      */
213     int numInitializingStreams() {
214         return frameStreamToInitializeMap.size();
215     }
216 
217     @Override
218     public final void handlerAdded(ChannelHandlerContext ctx) throws Exception {
219         this.ctx = ctx;
220         super.handlerAdded(ctx);
221         handlerAdded0(ctx);
222         // Must be after Http2ConnectionHandler does its initialization in handlerAdded above.
223         // The server will not send a connection preface so we are good to send a window update.
224         Http2Connection connection = connection();
225         if (connection.isServer()) {
226             tryExpandConnectionFlowControlWindow(connection);
227         }
228     }
229 
230     private void tryExpandConnectionFlowControlWindow(Http2Connection connection) throws Http2Exception {
231         if (initialFlowControlWindowSize != null) {
232             // The window size in the settings explicitly excludes the connection window. So we manually manipulate the
233             // connection window to accommodate more concurrent data per connection.
234             Http2Stream connectionStream = connection.connectionStream();
235             Http2LocalFlowController localFlowController = connection.local().flowController();
236             final int delta = initialFlowControlWindowSize - localFlowController.initialWindowSize(connectionStream);
237             // Only increase the connection window, don't decrease it.
238             if (delta > 0) {
239                 // Double the delta just so a single stream can't exhaust the connection window.
240                 localFlowController.incrementWindowSize(connectionStream, Math.max(delta << 1, delta));
241                 flush(ctx);
242             }
243         }
244     }
245 
246     void handlerAdded0(@SuppressWarnings("unsed") ChannelHandlerContext ctx) throws Exception {
247         // sub-class can override this for extra steps that needs to be done when the handler is added.
248     }
249 
250     /**
251      * Handles the cleartext HTTP upgrade event. If an upgrade occurred, sends a simple response via
252      * HTTP/2 on stream 1 (the stream specifically reserved for cleartext HTTP upgrade).
253      */
254     @Override
255     public final void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception {
256         if (evt == Http2ConnectionPrefaceAndSettingsFrameWrittenEvent.INSTANCE) {
257             // The user event implies that we are on the client.
258             tryExpandConnectionFlowControlWindow(connection());
259 
260             // We schedule this on the EventExecutor to allow to have any extra handlers added to the pipeline
261             // before we pass the event to the next handler. This is needed as the event may be called from within
262             // handlerAdded(...) which will be run before other handlers will be added to the pipeline.
263             ctx.executor().execute(new Runnable() {
264                 @Override
265                 public void run() {
266                     ctx.fireUserEventTriggered(evt);
267                 }
268             });
269         } else if (evt instanceof UpgradeEvent) {
270             UpgradeEvent upgrade = (UpgradeEvent) evt;
271             try {
272                 onUpgradeEvent(ctx, upgrade.retain());
273                 Http2Stream stream = connection().stream(HTTP_UPGRADE_STREAM_ID);
274                 if (stream == null) {
275                     // The upgrade failed (e.g. because the channel was already closed), so the stream was never
276                     // created. The failure was already propagated through the pipeline.
277                     return;
278                 }
279                 if (stream.getProperty(streamKey) == null) {
280                     // TODO: improve handler/stream lifecycle so that stream isn't active before handler added.
281                     // The stream was already made active, but ctx may have been null so it wasn't initialized.
282                     // https://github.com/netty/netty/issues/4942
283                     onStreamActive0(stream);
284                 }
285                 upgrade.upgradeRequest().headers().setInt(
286                         HttpConversionUtil.ExtensionHeaderNames.STREAM_ID.text(), HTTP_UPGRADE_STREAM_ID);
287                 stream.setProperty(upgradeKey, true);
288                 InboundHttpToHttp2Adapter.handle(
289                         ctx, connection(), decoder().frameListener(), upgrade.upgradeRequest().retain());
290             } finally {
291                 upgrade.release();
292             }
293         } else {
294             onUserEventTriggered(ctx, evt);
295             ctx.fireUserEventTriggered(evt);
296         }
297     }
298 
299     void onUserEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception {
300         // noop
301     }
302 
303     /**
304      * Processes all {@link Http2Frame}s. {@link Http2StreamFrame}s may only originate in child
305      * streams.
306      */
307     @Override
308     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
309         if (msg instanceof Http2DataFrame) {
310             Http2DataFrame dataFrame = (Http2DataFrame) msg;
311             encoder().writeData(ctx, dataFrame.stream().id(), dataFrame.content(),
312                     dataFrame.padding(), dataFrame.isEndStream(), promise);
313         } else if (msg instanceof Http2HeadersFrame) {
314             writeHeadersFrame(ctx, (Http2HeadersFrame) msg, promise);
315         } else if (msg instanceof Http2WindowUpdateFrame) {
316             Http2WindowUpdateFrame frame = (Http2WindowUpdateFrame) msg;
317             Http2FrameStream frameStream = frame.stream();
318             // It is legit to send a WINDOW_UPDATE frame for the connection stream. The parent channel doesn't attempt
319             // to set the Http2FrameStream so we assume if it is null the WINDOW_UPDATE is for the connection stream.
320             try {
321                 if (frameStream == null) {
322                     increaseInitialConnectionWindow(frame.windowSizeIncrement());
323                 } else {
324                     consumeBytes(frameStream.id(), frame.windowSizeIncrement());
325                 }
326                 promise.setSuccess();
327             } catch (Throwable t) {
328                 promise.setFailure(t);
329             }
330         } else if (msg instanceof Http2ResetFrame) {
331             Http2ResetFrame rstFrame = (Http2ResetFrame) msg;
332             int id = rstFrame.stream().id();
333             // Only ever send a reset frame if stream may have existed before as otherwise we may send a RST on a
334             // stream in an invalid state and cause a connection error.
335             if (connection().streamMayHaveExisted(id)) {
336                 encoder().writeRstStream(ctx, rstFrame.stream().id(), rstFrame.errorCode(), promise);
337             } else {
338                 ReferenceCountUtil.release(rstFrame);
339                 promise.setFailure(Http2Exception.streamError(
340                         rstFrame.stream().id(), Http2Error.PROTOCOL_ERROR, "Stream never existed"));
341             }
342         } else if (msg instanceof Http2PingFrame) {
343             Http2PingFrame frame = (Http2PingFrame) msg;
344             encoder().writePing(ctx, frame.ack(), frame.content(), promise);
345         } else if (msg instanceof Http2SettingsFrame) {
346             encoder().writeSettings(ctx, ((Http2SettingsFrame) msg).settings(), promise);
347         } else if (msg instanceof Http2SettingsAckFrame) {
348             // In the event of manual SETTINGS ACK, it is assumed the encoder will apply the earliest received but not
349             // yet ACKed settings.
350             encoder().writeSettingsAck(ctx, promise);
351         } else if (msg instanceof Http2GoAwayFrame) {
352             writeGoAwayFrame(ctx, (Http2GoAwayFrame) msg, promise);
353         } else if (msg instanceof Http2PushPromiseFrame) {
354             Http2PushPromiseFrame pushPromiseFrame = (Http2PushPromiseFrame) msg;
355             writePushPromise(ctx, pushPromiseFrame, promise);
356         } else if (msg instanceof Http2PriorityFrame) {
357             Http2PriorityFrame priorityFrame = (Http2PriorityFrame) msg;
358             encoder().writePriority(ctx, priorityFrame.stream().id(), priorityFrame.streamDependency(),
359                     priorityFrame.weight(), priorityFrame.exclusive(), promise);
360         } else if (msg instanceof Http2UnknownFrame) {
361             Http2UnknownFrame unknownFrame = (Http2UnknownFrame) msg;
362             encoder().writeFrame(ctx, unknownFrame.frameType(), unknownFrame.stream().id(),
363                     unknownFrame.flags(), unknownFrame.content(), promise);
364         } else if (!(msg instanceof Http2Frame)) {
365             ctx.write(msg, promise);
366         } else {
367             ReferenceCountUtil.release(msg);
368             throw new UnsupportedMessageTypeException(msg, SUPPORTED_MESSAGES);
369         }
370     }
371 
372     private void increaseInitialConnectionWindow(int deltaBytes) throws Http2Exception {
373         // The LocalFlowController is responsible for detecting over/under flow.
374         connection().local().flowController().incrementWindowSize(connection().connectionStream(), deltaBytes);
375     }
376 
377     final boolean consumeBytes(int streamId, int bytes) throws Http2Exception {
378         Http2Stream stream = connection().stream(streamId);
379         // Upgraded requests are ineligible for stream control. We add the null check
380         // in case the stream has been deregistered.
381         if (stream != null && streamId == Http2CodecUtil.HTTP_UPGRADE_STREAM_ID) {
382             Boolean upgraded = stream.getProperty(upgradeKey);
383             if (Boolean.TRUE.equals(upgraded)) {
384                 return false;
385             }
386         }
387 
388         return connection().local().flowController().consumeBytes(stream, bytes);
389     }
390 
391     private void writeGoAwayFrame(ChannelHandlerContext ctx, Http2GoAwayFrame frame, ChannelPromise promise) {
392         if (frame.lastStreamId() > -1) {
393             frame.release();
394             throw new IllegalArgumentException("Last stream id must not be set on GOAWAY frame");
395         }
396 
397         int lastStreamCreated = connection().remote().lastStreamCreated();
398         long lastStreamId = lastStreamCreated + ((long) frame.extraStreamIds()) * 2;
399         // Check if the computation overflowed.
400         if (lastStreamId > Integer.MAX_VALUE) {
401             lastStreamId = Integer.MAX_VALUE;
402         }
403         goAway(ctx, (int) lastStreamId, frame.errorCode(), frame.content(), promise);
404     }
405 
406     private void writeHeadersFrame(final ChannelHandlerContext ctx, Http2HeadersFrame headersFrame,
407                                    ChannelPromise promise) {
408 
409         if (isStreamIdValid(headersFrame.stream().id())) {
410             encoder().writeHeaders(ctx, headersFrame.stream().id(), headersFrame.headers(), headersFrame.padding(),
411                     headersFrame.isEndStream(), promise);
412         } else if (initializeNewStream(ctx, (DefaultHttp2FrameStream) headersFrame.stream(), promise)) {
413             promise = promise.unvoid();
414 
415             final int streamId = headersFrame.stream().id();
416 
417             encoder().writeHeaders(ctx, streamId, headersFrame.headers(), headersFrame.padding(),
418                     headersFrame.isEndStream(), promise);
419 
420             if (!promise.isDone()) {
421                 numBufferedStreams++;
422                 // Clean up the stream being initialized if writing the headers fails and also
423                 // decrement the number of buffered streams.
424                 promise.addListener(new ChannelFutureListener() {
425                     @Override
426                     public void operationComplete(ChannelFuture channelFuture) {
427                         numBufferedStreams--;
428                         handleHeaderFuture(channelFuture, streamId);
429                     }
430                 });
431             } else {
432                 handleHeaderFuture(promise, streamId);
433             }
434         }
435     }
436 
437     private void writePushPromise(final ChannelHandlerContext ctx, Http2PushPromiseFrame pushPromiseFrame,
438                                   final ChannelPromise promise) {
439         if (isStreamIdValid(pushPromiseFrame.pushStream().id())) {
440             encoder().writePushPromise(ctx, pushPromiseFrame.stream().id(), pushPromiseFrame.pushStream().id(),
441                     pushPromiseFrame.http2Headers(), pushPromiseFrame.padding(), promise);
442         } else if (initializeNewStream(ctx, (DefaultHttp2FrameStream) pushPromiseFrame.pushStream(), promise)) {
443             final int streamId = pushPromiseFrame.stream().id();
444             encoder().writePushPromise(ctx, streamId, pushPromiseFrame.pushStream().id(),
445                     pushPromiseFrame.http2Headers(), pushPromiseFrame.padding(), promise);
446 
447             if (promise.isDone()) {
448                 handleHeaderFuture(promise, streamId);
449             } else {
450                 numBufferedStreams++;
451                 // Clean up the stream being initialized if writing the headers fails and also
452                 // decrement the number of buffered streams.
453                 promise.addListener(new ChannelFutureListener() {
454                     @Override
455                     public void operationComplete(ChannelFuture channelFuture) {
456                         numBufferedStreams--;
457                         handleHeaderFuture(channelFuture, streamId);
458                     }
459                 });
460             }
461         }
462     }
463 
464     private boolean initializeNewStream(ChannelHandlerContext ctx, DefaultHttp2FrameStream http2FrameStream,
465                                         ChannelPromise promise) {
466         final Http2Connection connection = connection();
467         final int streamId = connection.local().incrementAndGetNextStreamId();
468         if (streamId < 0) {
469             promise.setFailure(new Http2NoMoreStreamIdsException());
470 
471             // Simulate a GOAWAY being received due to stream exhaustion on this connection. We use the maximum
472             // valid stream ID for the current peer.
473             onHttp2Frame(ctx, new DefaultHttp2GoAwayFrame(connection.isServer() ? Integer.MAX_VALUE :
474                     Integer.MAX_VALUE - 1, NO_ERROR.code(),
475                     writeAscii(ctx.alloc(), "Stream IDs exhausted on local stream creation")));
476 
477             return false;
478         }
479         http2FrameStream.id = streamId;
480 
481         // Use a Map to store all pending streams as we may have multiple. This is needed as if we would store the
482         // stream in a field directly we may override the stored field before onStreamAdded(...) was called
483         // and so not correctly set the property for the buffered stream.
484         //
485         // See https://github.com/netty/netty/issues/8692
486         Object old = frameStreamToInitializeMap.put(streamId, http2FrameStream);
487 
488         // We should not re-use ids.
489         assert old == null;
490         return true;
491     }
492 
493     private void handleHeaderFuture(ChannelFuture channelFuture, int streamId) {
494         if (!channelFuture.isSuccess()) {
495             frameStreamToInitializeMap.remove(streamId);
496         }
497     }
498 
499     private void onStreamActive0(Http2Stream stream) {
500         if (stream.id() != Http2CodecUtil.HTTP_UPGRADE_STREAM_ID &&
501                 connection().local().isValidStreamId(stream.id())) {
502             return;
503         }
504 
505         DefaultHttp2FrameStream stream2 = newStream().setStreamAndProperty(streamKey, stream);
506         onHttp2StreamStateChanged(ctx, stream2);
507     }
508 
509     private final class ConnectionListener extends Http2ConnectionAdapter {
510         @Override
511         public void onStreamAdded(Http2Stream stream) {
512             DefaultHttp2FrameStream frameStream = frameStreamToInitializeMap.remove(stream.id());
513 
514             if (frameStream != null) {
515                 frameStream.setStreamAndProperty(streamKey, stream);
516             }
517         }
518 
519         @Override
520         public void onStreamActive(Http2Stream stream) {
521             onStreamActive0(stream);
522         }
523 
524         @Override
525         public void onStreamClosed(Http2Stream stream) {
526             onHttp2StreamStateChanged0(stream);
527         }
528 
529         @Override
530         public void onStreamHalfClosed(Http2Stream stream) {
531             onHttp2StreamStateChanged0(stream);
532         }
533 
534         private void onHttp2StreamStateChanged0(Http2Stream stream) {
535             DefaultHttp2FrameStream stream2 = stream.getProperty(streamKey);
536             if (stream2 != null) {
537                 onHttp2StreamStateChanged(ctx, stream2);
538             }
539         }
540     }
541 
542     @Override
543     protected void onConnectionError(
544             ChannelHandlerContext ctx, boolean outbound, Throwable cause, Http2Exception http2Ex) {
545         if (!outbound) {
546             // allow the user to handle it first in the pipeline, and then automatically clean up.
547             // If this is not desired behavior the user can override this method.
548             //
549             // We only forward non outbound errors as outbound errors will already be reflected by failing the promise.
550             ctx.fireExceptionCaught(cause);
551         }
552         super.onConnectionError(ctx, outbound, cause, http2Ex);
553     }
554 
555     /**
556      * Exceptions for unknown streams, that is streams that have no {@link Http2FrameStream} object attached
557      * are simply logged and replied to by sending a RST_STREAM frame.
558      */
559     @Override
560     protected final void onStreamError(ChannelHandlerContext ctx, boolean outbound, Throwable cause,
561                                        Http2Exception.StreamException streamException) {
562         int streamId = streamException.streamId();
563         Http2Stream connectionStream = connection().stream(streamId);
564         if (connectionStream == null) {
565             onHttp2UnknownStreamError(ctx, cause, streamException);
566             // Write a RST_STREAM
567             super.onStreamError(ctx, outbound, cause, streamException);
568             return;
569         }
570 
571         Http2FrameStream stream = connectionStream.getProperty(streamKey);
572         if (stream == null) {
573             LOG.warn("{} Stream exception thrown without stream object attached.", ctx.channel(), cause);
574             // Write a RST_STREAM
575             super.onStreamError(ctx, outbound, cause, streamException);
576             return;
577         }
578 
579         if (!outbound) {
580             // We only forward non outbound errors as outbound errors will already be reflected by failing the promise.
581             onHttp2FrameStreamException(ctx, new Http2FrameStreamException(stream, streamException.error(), cause));
582         }
583     }
584 
585     private static void onHttp2UnknownStreamError(@SuppressWarnings("unused") ChannelHandlerContext ctx,
586             Throwable cause, Http2Exception.StreamException streamException) {
587         // We log here for debugging purposes. This exception will be propagated to the upper layers through other ways:
588         // - fireExceptionCaught
589         // - fireUserEventTriggered(Http2ResetFrame), see Http2MultiplexHandler#channelRead(...)
590         // - by failing write promise
591         // Receiver of the error is responsible for correct handling of this exception.
592         LOG.log(DEBUG, "{} Stream exception thrown for unknown stream {}.",
593                 ctx.channel(), streamException.streamId(), cause);
594     }
595 
596     @Override
597     protected final boolean isGracefulShutdownComplete() {
598         return super.isGracefulShutdownComplete() && numBufferedStreams == 0;
599     }
600 
601     private final class FrameListener implements Http2FrameListener {
602 
603         @Override
604         public void onUnknownFrame(
605                 ChannelHandlerContext ctx, byte frameType, int streamId, Http2Flags flags, ByteBuf payload) {
606             if (streamId == 0) {
607                 // Ignore unknown frames on connection stream, for example: HTTP/2 GREASE testing
608                 return;
609             }
610             Http2FrameStream stream = requireStream(streamId);
611             onHttp2Frame(ctx, newHttp2UnknownFrame(frameType, streamId, flags, payload.retain()).stream(stream));
612         }
613 
614         @Override
615         public void onSettingsRead(ChannelHandlerContext ctx, Http2Settings settings) {
616             onHttp2Frame(ctx, new DefaultHttp2SettingsFrame(settings));
617         }
618 
619         @Override
620         public void onPingRead(ChannelHandlerContext ctx, long data) {
621             onHttp2Frame(ctx, new DefaultHttp2PingFrame(data, false));
622         }
623 
624         @Override
625         public void onPingAckRead(ChannelHandlerContext ctx, long data) {
626             onHttp2Frame(ctx, new DefaultHttp2PingFrame(data, true));
627         }
628 
629         @Override
630         public void onRstStreamRead(ChannelHandlerContext ctx, int streamId, long errorCode) {
631             Http2FrameStream stream = requireStream(streamId);
632             onHttp2Frame(ctx, new DefaultHttp2ResetFrame(errorCode).stream(stream));
633         }
634 
635         @Override
636         public void onWindowUpdateRead(ChannelHandlerContext ctx, int streamId, int windowSizeIncrement) {
637             if (streamId == 0) {
638                 // Ignore connection window updates.
639                 return;
640             }
641             Http2FrameStream stream = requireStream(streamId);
642             onHttp2Frame(ctx, new DefaultHttp2WindowUpdateFrame(windowSizeIncrement).stream(stream));
643         }
644 
645         @Override
646         public void onHeadersRead(ChannelHandlerContext ctx, int streamId,
647                                   Http2Headers headers, int streamDependency, short weight, boolean
648                                           exclusive, int padding, boolean endStream) {
649             onHeadersRead(ctx, streamId, headers, padding, endStream);
650         }
651 
652         @Override
653         public void onHeadersRead(ChannelHandlerContext ctx, int streamId, Http2Headers headers,
654                                   int padding, boolean endOfStream) {
655             Http2FrameStream stream = requireStream(streamId);
656             onHttp2Frame(ctx, new DefaultHttp2HeadersFrame(headers, endOfStream, padding).stream(stream));
657         }
658 
659         @Override
660         public int onDataRead(ChannelHandlerContext ctx, int streamId, ByteBuf data, int padding,
661                               boolean endOfStream) {
662             Http2FrameStream stream = requireStream(streamId);
663             final Http2DataFrame dataframe;
664             try {
665                 dataframe = new DefaultHttp2DataFrame(data.retain(), endOfStream, padding);
666             } catch (IllegalArgumentException e) {
667                 // Might be thrown in case of invalid padding / length.
668                 data.release();
669                 throw e;
670             }
671             dataframe.stream(stream);
672             onHttp2Frame(ctx, dataframe);
673             // We return the bytes in consumeBytes() once the stream channel consumed the bytes.
674             return 0;
675         }
676 
677         @Override
678         public void onGoAwayRead(ChannelHandlerContext ctx, int lastStreamId, long errorCode, ByteBuf debugData) {
679             onHttp2Frame(ctx, new DefaultHttp2GoAwayFrame(lastStreamId, errorCode, debugData.retain()));
680         }
681 
682         @Override
683         public void onPriorityRead(ChannelHandlerContext ctx, int streamId, int streamDependency,
684                                    short weight, boolean exclusive) {
685 
686             Http2Stream stream = connection().stream(streamId);
687             if (stream == null) {
688                 // The stream was not opened yet, let's just ignore this for now.
689                 return;
690             }
691             Http2FrameStream frameStream = requireStream(streamId);
692             onHttp2Frame(ctx, new DefaultHttp2PriorityFrame(streamDependency, weight, exclusive)
693                     .stream(frameStream));
694         }
695 
696         @Override
697         public void onSettingsAckRead(ChannelHandlerContext ctx) {
698             onHttp2Frame(ctx, Http2SettingsAckFrame.INSTANCE);
699         }
700 
701         @Override
702         public void onPushPromiseRead(ChannelHandlerContext ctx, int streamId, int promisedStreamId,
703                                       Http2Headers headers, int padding) {
704             Http2FrameStream stream = requireStream(streamId);
705             onHttp2Frame(ctx, new DefaultHttp2PushPromiseFrame(headers, padding, promisedStreamId)
706                     .pushStream(new DefaultHttp2FrameStream()
707                             .setStreamAndProperty(streamKey, connection().stream(promisedStreamId)))
708                     .stream(stream));
709         }
710 
711         private Http2FrameStream requireStream(int streamId) {
712             Http2FrameStream stream = connection().stream(streamId).getProperty(streamKey);
713             if (stream == null) {
714                 throw new IllegalStateException("Stream object required for identifier: " + streamId);
715             }
716             return stream;
717         }
718     }
719 
720     private void onUpgradeEvent(ChannelHandlerContext ctx, UpgradeEvent evt) {
721         ctx.fireUserEventTriggered(evt);
722     }
723 
724     private void onHttp2StreamWritabilityChanged(ChannelHandlerContext ctx, DefaultHttp2FrameStream stream,
725                                                  @SuppressWarnings("unused") boolean writable) {
726         ctx.fireUserEventTriggered(stream.writabilityChanged);
727     }
728 
729     void onHttp2StreamStateChanged(ChannelHandlerContext ctx, DefaultHttp2FrameStream stream) {
730         ctx.fireUserEventTriggered(stream.stateChanged);
731     }
732 
733     void onHttp2Frame(ChannelHandlerContext ctx, Http2Frame frame) {
734         ctx.fireChannelRead(frame);
735     }
736 
737     /**
738      * Create a Http2UnknownFrame. The ownership of the {@link ByteBuf} is transferred.
739      * */
740     protected Http2StreamFrame newHttp2UnknownFrame(byte frameType, int streamId, Http2Flags flags, ByteBuf payload) {
741         return new DefaultHttp2UnknownFrame(frameType, flags, payload);
742     }
743 
744     void onHttp2FrameStreamException(ChannelHandlerContext ctx, Http2FrameStreamException cause) {
745         ctx.fireExceptionCaught(cause);
746     }
747 
748     private final class Http2RemoteFlowControllerListener implements Http2RemoteFlowController.Listener {
749         @Override
750         public void writabilityChanged(Http2Stream stream) {
751             DefaultHttp2FrameStream frameStream = stream.getProperty(streamKey);
752             if (frameStream == null) {
753                 return;
754             }
755             onHttp2StreamWritabilityChanged(
756                     ctx, frameStream, connection().remote().flowController().isWritable(stream));
757         }
758     }
759 
760     /**
761      * {@link Http2FrameStream} implementation.
762      */
763     // TODO(buchgr): Merge Http2FrameStream and Http2Stream.
764     static class DefaultHttp2FrameStream implements Http2FrameStream {
765 
766         private volatile int id = -1;
767         private volatile Http2Stream stream;
768 
769         final Http2FrameStreamEvent stateChanged = Http2FrameStreamEvent.stateChanged(this);
770         final Http2FrameStreamEvent writabilityChanged = Http2FrameStreamEvent.writabilityChanged(this);
771 
772         Channel attachment;
773 
774         DefaultHttp2FrameStream setStreamAndProperty(PropertyKey streamKey, Http2Stream stream) {
775             assert id == -1 || stream.id() == id;
776             this.stream = stream;
777             this.id = stream.id();
778             stream.setProperty(streamKey, this);
779             return this;
780         }
781 
782         @Override
783         public int id() {
784             Http2Stream stream = this.stream;
785             return stream == null ? id : stream.id();
786         }
787 
788         @Override
789         public State state() {
790             Http2Stream stream = this.stream;
791             return stream == null ? State.IDLE : stream.state();
792         }
793 
794         @Override
795         public String toString() {
796             return String.valueOf(id());
797         }
798     }
799 }