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