1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
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
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
181
182 DefaultHttp2FrameStream newStream() {
183 return new DefaultHttp2FrameStream();
184 }
185
186
187
188
189
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
210
211
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
223
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
233
234 Http2Stream connectionStream = connection.connectionStream();
235 Http2LocalFlowController localFlowController = connection.local().flowController();
236 final int delta = initialFlowControlWindowSize - localFlowController.initialWindowSize(connectionStream);
237
238 if (delta > 0) {
239
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
248 }
249
250
251
252
253
254 @Override
255 public final void userEventTriggered(final ChannelHandlerContext ctx, final Object evt) throws Exception {
256 if (evt == Http2ConnectionPrefaceAndSettingsFrameWrittenEvent.INSTANCE) {
257
258 tryExpandConnectionFlowControlWindow(connection());
259
260
261
262
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
276
277 return;
278 }
279 if (stream.getProperty(streamKey) == null) {
280
281
282
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
301 }
302
303
304
305
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
319
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
334
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
349
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
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
380
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
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
423
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
452
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
472
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
482
483
484
485
486 Object old = frameStreamToInitializeMap.put(streamId, http2FrameStream);
487
488
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
547
548
549
550 ctx.fireExceptionCaught(cause);
551 }
552 super.onConnectionError(ctx, outbound, cause, http2Ex);
553 }
554
555
556
557
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
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
575 super.onStreamError(ctx, outbound, cause, streamException);
576 return;
577 }
578
579 if (!outbound) {
580
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
588
589
590
591
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
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
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
668 data.release();
669 throw e;
670 }
671 dataframe.stream(stream);
672 onHttp2Frame(ctx, dataframe);
673
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
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
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
762
763
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 }