1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.handler.codec.http3;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.Unpooled;
20 import io.netty.channel.ChannelHandler;
21 import io.netty.channel.ChannelHandlerContext;
22 import io.netty.channel.ChannelOutboundHandler;
23 import io.netty.channel.ChannelPromise;
24 import io.netty.channel.PendingWriteQueue;
25 import io.netty.handler.codec.ByteToMessageDecoder;
26 import io.netty.handler.codec.quic.QuicStreamChannel;
27 import io.netty.handler.codec.quic.QuicStreamFrame;
28 import io.netty.util.ReferenceCountUtil;
29 import io.netty.util.concurrent.Future;
30 import io.netty.util.concurrent.GenericFutureListener;
31 import org.jetbrains.annotations.Nullable;
32
33 import java.net.SocketAddress;
34 import java.util.List;
35 import java.util.Map;
36 import java.util.function.BiFunction;
37
38 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_CANCEL_PUSH_FRAME_MAX_LEN;
39 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_CANCEL_PUSH_FRAME_TYPE;
40 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_DATA_FRAME_TYPE;
41 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_GO_AWAY_FRAME_MAX_LEN;
42 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_GO_AWAY_FRAME_TYPE;
43 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_HEADERS_FRAME_TYPE;
44 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_MAX_PUSH_ID_FRAME_MAX_LEN;
45 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_MAX_PUSH_ID_FRAME_TYPE;
46 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_PUSH_PROMISE_FRAME_TYPE;
47 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_SETTINGS_FRAME_MAX_LEN;
48 import static io.netty.handler.codec.http3.Http3CodecUtils.HTTP3_SETTINGS_FRAME_TYPE;
49 import static io.netty.handler.codec.http3.Http3CodecUtils.numBytesForVariableLengthInteger;
50 import static io.netty.handler.codec.http3.Http3CodecUtils.readVariableLengthInteger;
51 import static io.netty.handler.codec.http3.Http3CodecUtils.writeVariableLengthInteger;
52 import static io.netty.util.internal.ObjectUtil.checkNotNull;
53 import static io.netty.util.internal.ObjectUtil.checkPositive;
54 import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
55
56
57
58
59 final class Http3FrameCodec extends ByteToMessageDecoder implements ChannelOutboundHandler {
60 private final Http3FrameTypeValidator validator;
61 private final long maxHeaderListSize;
62 private final int maxUnknownFramePayloadLength;
63 private final QpackDecoder qpackDecoder;
64 private final QpackEncoder qpackEncoder;
65 private final Http3RequestStreamCodecState encodeState;
66 private final Http3RequestStreamCodecState decodeState;
67 private final Http3Settings.NonStandardHttp3SettingsValidator nonStandardSettingsValidator;
68 private boolean firstFrame = true;
69 private boolean error;
70 private long type = -1;
71 private int payLoadLength = -1;
72 private QpackAttributes qpackAttributes;
73 private ReadResumptionListener readResumptionListener;
74 private WriteResumptionListener writeResumptionListener;
75
76 static Http3FrameCodecFactory newFactory(QpackDecoder qpackDecoder,
77 long maxHeaderListSize, int maxUnknownFramePayloadLength,
78 QpackEncoder qpackEncoder) {
79 checkNotNull(qpackEncoder, "qpackEncoder");
80 checkNotNull(qpackDecoder, "qpackDecoder");
81 checkPositive(maxHeaderListSize, "maxHeaderListSize");
82 checkPositive(maxUnknownFramePayloadLength, "maxUnknownFramePayloadLength");
83
84
85 return (validator, encodeState, decodeState,
86 nonStandardSettingsValidator) -> new Http3FrameCodec(validator, qpackDecoder,
87 maxHeaderListSize, maxUnknownFramePayloadLength, qpackEncoder,
88 encodeState, decodeState, nonStandardSettingsValidator);
89 }
90
91 Http3FrameCodec(Http3FrameTypeValidator validator, QpackDecoder qpackDecoder,
92 long maxHeaderListSize, int maxUnknownFramePayloadLength,
93 QpackEncoder qpackEncoder, Http3RequestStreamCodecState encodeState,
94 Http3RequestStreamCodecState decodeState,
95 Http3Settings.NonStandardHttp3SettingsValidator nonStandardSettingsValidator) {
96 this.validator = checkNotNull(validator, "validator");
97 this.qpackDecoder = checkNotNull(qpackDecoder, "qpackDecoder");
98 this.maxHeaderListSize = checkPositive(maxHeaderListSize, "maxHeaderListSize");
99 this.maxUnknownFramePayloadLength = checkPositive(maxUnknownFramePayloadLength, "maxUnknownFramePayloadLength");
100 this.qpackEncoder = checkNotNull(qpackEncoder, "qpackEncoder");
101 this.encodeState = checkNotNull(encodeState, "encodeState");
102 this.decodeState = checkNotNull(decodeState, "decodeState");
103 this.nonStandardSettingsValidator = nonStandardSettingsValidator;
104 }
105
106 @Override
107 public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
108 qpackAttributes = Http3.getQpackAttributes(ctx.channel().parent());
109 assert qpackAttributes != null;
110
111 initReadResumptionListenerIfRequired(ctx);
112 super.handlerAdded(ctx);
113 }
114
115 @Override
116 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
117 if (writeResumptionListener != null) {
118 writeResumptionListener.drain();
119 }
120 super.channelInactive(ctx);
121 }
122
123 @Override
124 protected void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
125 if (writeResumptionListener != null) {
126
127 writeResumptionListener.drain();
128 }
129 super.handlerRemoved0(ctx);
130 }
131
132 @Override
133 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
134 ByteBuf buffer;
135 if (msg instanceof QuicStreamFrame) {
136 QuicStreamFrame streamFrame = (QuicStreamFrame) msg;
137 buffer = streamFrame.content().retain();
138 streamFrame.release();
139 } else {
140 buffer = (ByteBuf) msg;
141 }
142 super.channelRead(ctx, buffer);
143 }
144
145 @Override
146 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
147 assert readResumptionListener != null;
148 if (readResumptionListener.readCompleted()) {
149 super.channelReadComplete(ctx);
150 }
151 }
152
153 private void connectionError(ChannelHandlerContext ctx, Http3ErrorCode code, String msg, boolean fireException) {
154 error = true;
155 Http3CodecUtils.connectionError(ctx, code, msg, fireException);
156 }
157
158 private void connectionError(ChannelHandlerContext ctx, Http3Exception exception, boolean fireException) {
159 error = true;
160 Http3CodecUtils.connectionError(ctx, exception, fireException);
161 }
162
163 @Override
164 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
165 assert readResumptionListener != null;
166 if (!in.isReadable() || readResumptionListener.isSuspended()) {
167 return;
168 }
169 if (error) {
170 in.skipBytes(in.readableBytes());
171 return;
172 }
173 if (type == -1) {
174 int typeLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
175 if (in.readableBytes() < typeLen) {
176 return;
177 }
178 long localType = readVariableLengthInteger(in, typeLen);
179 if (Http3CodecUtils.isReservedHttp2FrameType(localType)) {
180
181 connectionError(ctx, Http3ErrorCode.H3_FRAME_UNEXPECTED,
182 "Reserved type for HTTP/2 received.", true);
183 return;
184 }
185 try {
186
187 validator.validate(localType, firstFrame);
188 } catch (Http3Exception e) {
189 connectionError(ctx, e, true);
190 return;
191 }
192 type = localType;
193 firstFrame = false;
194 if (!in.isReadable()) {
195 return;
196 }
197 }
198 if (payLoadLength == -1) {
199 int payloadLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
200 assert payloadLen <= 8;
201 if (in.readableBytes() < payloadLen) {
202 return;
203 }
204 long len = readVariableLengthInteger(in, payloadLen);
205 if (len > Integer.MAX_VALUE) {
206 connectionError(ctx, Http3ErrorCode.H3_EXCESSIVE_LOAD,
207 "Received an invalid frame len.", true);
208 return;
209 }
210 payLoadLength = (int) len;
211 }
212 int read = decodeFrame(ctx, type, payLoadLength, in, out);
213 if (read >= 0) {
214 if (read == payLoadLength) {
215 type = -1;
216 payLoadLength = -1;
217 } else {
218 payLoadLength -= read;
219 }
220 }
221 }
222
223 private static int skipBytes(ByteBuf in, int payLoadLength) {
224 int length = Math.min(in.readableBytes(), payLoadLength);
225 in.skipBytes(length);
226 return length;
227 }
228
229 private int decodeFrame(ChannelHandlerContext ctx, long longType, int payLoadLength, ByteBuf in, List<Object> out) {
230 if (longType > Integer.MAX_VALUE && !Http3CodecUtils.isReservedFrameType(longType)) {
231 return skipBytes(in, payLoadLength);
232 }
233 int type = (int) longType;
234
235 switch (type) {
236 case HTTP3_DATA_FRAME_TYPE:
237
238
239 int readable = in.readableBytes();
240 if (readable == 0 && payLoadLength > 0) {
241 return 0;
242 }
243 int length = Math.min(readable, payLoadLength);
244 out.add(new DefaultHttp3DataFrame(in.readRetainedSlice(length)));
245 return length;
246 case HTTP3_HEADERS_FRAME_TYPE:
247
248
249 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength,
250
251
252
253 maxHeaderListSize, Http3ErrorCode.H3_EXCESSIVE_LOAD)) {
254 return 0;
255 }
256 assert qpackAttributes != null;
257 if (!qpackAttributes.dynamicTableDisabled() && !qpackAttributes.decoderStreamAvailable()) {
258 assert readResumptionListener != null;
259 readResumptionListener.suspended();
260 return 0;
261 }
262
263 Http3HeadersFrame headersFrame = new DefaultHttp3HeadersFrame();
264 if (decodeHeaders(ctx, headersFrame.headers(), in, payLoadLength, decodeState.receivedFinalHeaders())) {
265 out.add(headersFrame);
266 return payLoadLength;
267 }
268 return -1;
269 case HTTP3_CANCEL_PUSH_FRAME_TYPE:
270
271
272 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength,
273 HTTP3_CANCEL_PUSH_FRAME_MAX_LEN, Http3ErrorCode.H3_FRAME_ERROR)) {
274 return 0;
275 }
276 int pushIdLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
277 out.add(new DefaultHttp3CancelPushFrame(readVariableLengthInteger(in, pushIdLen)));
278 return payLoadLength;
279 case HTTP3_SETTINGS_FRAME_TYPE:
280
281
282
283
284 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength, HTTP3_SETTINGS_FRAME_MAX_LEN,
285 Http3ErrorCode.H3_EXCESSIVE_LOAD)) {
286 return 0;
287 }
288 Http3SettingsFrame settingsFrame = decodeSettings(ctx, in, payLoadLength);
289 if (settingsFrame != null) {
290 out.add(settingsFrame);
291 }
292 return payLoadLength;
293 case HTTP3_PUSH_PROMISE_FRAME_TYPE:
294
295
296 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength,
297
298
299
300 Math.max(maxHeaderListSize, maxHeaderListSize + 8), Http3ErrorCode.H3_EXCESSIVE_LOAD)) {
301 return 0;
302 }
303
304 assert qpackAttributes != null;
305 if (!qpackAttributes.dynamicTableDisabled() && !qpackAttributes.decoderStreamAvailable()) {
306 assert readResumptionListener != null;
307 readResumptionListener.suspended();
308 return 0;
309 }
310 int readerIdx = in.readerIndex();
311 int pushPromiseIdLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
312 Http3PushPromiseFrame pushPromiseFrame = new DefaultHttp3PushPromiseFrame(
313 readVariableLengthInteger(in, pushPromiseIdLen));
314 if (decodeHeaders(ctx, pushPromiseFrame.headers(), in, payLoadLength - pushPromiseIdLen, false)) {
315 out.add(pushPromiseFrame);
316 return payLoadLength;
317 }
318 in.readerIndex(readerIdx);
319 return -1;
320 case HTTP3_GO_AWAY_FRAME_TYPE:
321
322
323 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength,
324 HTTP3_GO_AWAY_FRAME_MAX_LEN, Http3ErrorCode.H3_FRAME_ERROR)) {
325 return 0;
326 }
327 int idLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
328 out.add(new DefaultHttp3GoAwayFrame(readVariableLengthInteger(in, idLen)));
329 return payLoadLength;
330 case HTTP3_MAX_PUSH_ID_FRAME_TYPE:
331
332
333 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength,
334 HTTP3_MAX_PUSH_ID_FRAME_MAX_LEN, Http3ErrorCode.H3_FRAME_ERROR)) {
335 return 0;
336 }
337 int pidLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
338 out.add(new DefaultHttp3MaxPushIdFrame(readVariableLengthInteger(in, pidLen)));
339 return payLoadLength;
340 default:
341 if (!Http3CodecUtils.isReservedFrameType(longType)) {
342 return skipBytes(in, payLoadLength);
343 }
344
345
346 if (!enforceMaxPayloadLength(ctx, in, type, payLoadLength,
347 maxUnknownFramePayloadLength, Http3ErrorCode.H3_EXCESSIVE_LOAD)) {
348 return 0;
349 }
350 out.add(new DefaultHttp3UnknownFrame(longType, in.readRetainedSlice(payLoadLength)));
351 return payLoadLength;
352 }
353 }
354
355 private boolean enforceMaxPayloadLength(
356 ChannelHandlerContext ctx, ByteBuf in, int type, int payLoadLength,
357 long maxPayLoadLength, Http3ErrorCode error) {
358 if (payLoadLength > maxPayLoadLength) {
359 connectionError(ctx, error,
360 "Received an invalid frame len " + payLoadLength + " for frame of type " + type + '.', true);
361 return false;
362 }
363 return in.readableBytes() >= payLoadLength;
364 }
365
366 @Nullable
367 private Http3SettingsFrame decodeSettings(ChannelHandlerContext ctx, ByteBuf in, int payLoadLength) {
368 Http3SettingsFrame settingsFrame = new DefaultHttp3SettingsFrame(
369 new Http3Settings(nonStandardSettingsValidator));
370 while (payLoadLength > 0) {
371 int keyLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
372 long key = readVariableLengthInteger(in, keyLen);
373 if (Http3CodecUtils.isReservedHttp2Setting(key)) {
374
375
376 connectionError(ctx, Http3ErrorCode.H3_SETTINGS_ERROR,
377 "Received a settings key that is reserved for HTTP/2.", true);
378 return null;
379 }
380 payLoadLength -= keyLen;
381 int valueLen = numBytesForVariableLengthInteger(in.getByte(in.readerIndex()));
382 long value = readVariableLengthInteger(in, valueLen);
383 payLoadLength -= valueLen;
384
385 if (settingsFrame.put(key, value) != null) {
386
387
388 connectionError(ctx, Http3ErrorCode.H3_SETTINGS_ERROR,
389 "Received a duplicate settings key.", true);
390 return null;
391 }
392 }
393 return settingsFrame;
394 }
395
396
397
398
399
400
401
402
403
404
405
406
407
408 private boolean decodeHeaders(ChannelHandlerContext ctx, Http3Headers headers, ByteBuf in, int length,
409 boolean trailer) {
410 try {
411 Http3HeadersSink sink = new Http3HeadersSink(headers, maxHeaderListSize, true, trailer);
412 assert qpackAttributes != null;
413 assert readResumptionListener != null;
414 if (qpackDecoder.decode(qpackAttributes,
415 ((QuicStreamChannel) ctx.channel()).streamId(), in, length, sink, readResumptionListener)) {
416
417 sink.finish();
418 return true;
419 }
420 readResumptionListener.suspended();
421 } catch (Http3Exception e) {
422 connectionError(ctx, e.errorCode(), e.getMessage(), true);
423 } catch (QpackException e) {
424
425 connectionError(ctx, Http3ErrorCode.QPACK_DECOMPRESSION_FAILED,
426 "Decompression of header block failed.", true);
427 } catch (Http3HeadersValidationException e) {
428 error = true;
429 ctx.fireExceptionCaught(e);
430
431
432 Http3CodecUtils.streamError(ctx, Http3ErrorCode.H3_MESSAGE_ERROR);
433 }
434 return false;
435 }
436
437 @Override
438 public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
439 assert qpackAttributes != null;
440 if (writeResumptionListener != null) {
441 writeResumptionListener.enqueue(msg, promise);
442 return;
443 }
444
445 if ((msg instanceof Http3HeadersFrame || msg instanceof Http3PushPromiseFrame) &&
446 !qpackAttributes.dynamicTableDisabled() && !qpackAttributes.encoderStreamAvailable()) {
447 writeResumptionListener = WriteResumptionListener.newListener(ctx, this);
448 writeResumptionListener.enqueue(msg, promise);
449 return;
450 }
451
452 write0(ctx, msg, promise);
453 }
454
455 private void write0(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
456 try {
457 if (msg instanceof Http3DataFrame) {
458 writeDataFrame(ctx, (Http3DataFrame) msg, promise);
459 } else if (msg instanceof Http3HeadersFrame) {
460 writeHeadersFrame(ctx, (Http3HeadersFrame) msg, promise);
461 } else if (msg instanceof Http3CancelPushFrame) {
462 writeCancelPushFrame(ctx, (Http3CancelPushFrame) msg, promise);
463 } else if (msg instanceof Http3SettingsFrame) {
464 writeSettingsFrame(ctx, (Http3SettingsFrame) msg, promise);
465 } else if (msg instanceof Http3PushPromiseFrame) {
466 writePushPromiseFrame(ctx, (Http3PushPromiseFrame) msg, promise);
467 } else if (msg instanceof Http3GoAwayFrame) {
468 writeGoAwayFrame(ctx, (Http3GoAwayFrame) msg, promise);
469 } else if (msg instanceof Http3MaxPushIdFrame) {
470 writeMaxPushIdFrame(ctx, (Http3MaxPushIdFrame) msg, promise);
471 } else if (msg instanceof Http3UnknownFrame) {
472 writeUnknownFrame(ctx, (Http3UnknownFrame) msg, promise);
473 } else {
474 unsupported(promise);
475 }
476 } finally {
477 ReferenceCountUtil.release(msg);
478 }
479 }
480
481 private static void writeDataFrame(
482 ChannelHandlerContext ctx, Http3DataFrame frame, ChannelPromise promise) {
483 ByteBuf out = ctx.alloc().directBuffer(16);
484 writeVariableLengthInteger(out, frame.type());
485 writeVariableLengthInteger(out, frame.content().readableBytes());
486 ByteBuf content = frame.content().retain();
487 ctx.write(Unpooled.wrappedUnmodifiableBuffer(out, content), promise);
488 }
489
490 private void writeHeadersFrame(ChannelHandlerContext ctx, Http3HeadersFrame frame, ChannelPromise promise) {
491 assert qpackAttributes != null;
492 final QuicStreamChannel channel = (QuicStreamChannel) ctx.channel();
493 writeDynamicFrame(ctx, frame.type(), frame, (f, out) -> {
494 qpackEncoder.encodeHeaders(qpackAttributes, out, ctx.alloc(), channel.streamId(), f.headers());
495 return true;
496 }, promise);
497 }
498
499 private static void writeCancelPushFrame(
500 ChannelHandlerContext ctx, Http3CancelPushFrame frame, ChannelPromise promise) {
501 writeFrameWithId(ctx, frame.type(), frame.id(), promise);
502 }
503
504 private static void writeSettingsFrame(
505 ChannelHandlerContext ctx, Http3SettingsFrame frame, ChannelPromise promise) {
506 writeDynamicFrame(ctx, frame.type(), frame, (f, out) -> {
507 for (Map.Entry<Long, Long> e : f) {
508 Long key = e.getKey();
509 if (Http3CodecUtils.isReservedHttp2Setting(key)) {
510 Http3Exception exception = new Http3Exception(Http3ErrorCode.H3_SETTINGS_ERROR,
511 "Received a settings key that is reserved for HTTP/2.");
512 promise.setFailure(exception);
513
514 Http3CodecUtils.connectionError(ctx, exception, false);
515 return false;
516 }
517 Long value = e.getValue();
518 int keyLen = numBytesForVariableLengthInteger(key);
519 int valueLen = numBytesForVariableLengthInteger(value);
520 writeVariableLengthInteger(out, key, keyLen);
521 writeVariableLengthInteger(out, value, valueLen);
522 }
523 return true;
524 }, promise);
525 }
526
527 private static <T extends Http3Frame> void writeDynamicFrame(ChannelHandlerContext ctx, long type, T frame,
528 BiFunction<T, ByteBuf, Boolean> writer,
529 ChannelPromise promise) {
530 ByteBuf out = ctx.alloc().directBuffer();
531 int initialWriterIndex = out.writerIndex();
532
533 int payloadStartIndex = initialWriterIndex + 16;
534 out.writerIndex(payloadStartIndex);
535
536 if (writer.apply(frame, out)) {
537 int finalWriterIndex = out.writerIndex();
538 int payloadLength = finalWriterIndex - payloadStartIndex;
539 int len = numBytesForVariableLengthInteger(payloadLength);
540 out.writerIndex(payloadStartIndex - len);
541 writeVariableLengthInteger(out, payloadLength, len);
542
543 int typeLength = numBytesForVariableLengthInteger(type);
544 int startIndex = payloadStartIndex - len - typeLength;
545 out.writerIndex(startIndex);
546 writeVariableLengthInteger(out, type, typeLength);
547
548 out.setIndex(startIndex, finalWriterIndex);
549 ctx.write(out, promise);
550 } else {
551
552 out.release();
553 }
554 }
555
556 private void writePushPromiseFrame(ChannelHandlerContext ctx, Http3PushPromiseFrame frame, ChannelPromise promise) {
557 assert qpackAttributes != null;
558 final QuicStreamChannel channel = (QuicStreamChannel) ctx.channel();
559 writeDynamicFrame(ctx, frame.type(), frame, (f, out) -> {
560 long id = f.id();
561 writeVariableLengthInteger(out, id);
562 qpackEncoder.encodeHeaders(qpackAttributes, out, ctx.alloc(), channel.streamId(), f.headers());
563 return true;
564 }, promise);
565 }
566
567 private static void writeGoAwayFrame(
568 ChannelHandlerContext ctx, Http3GoAwayFrame frame, ChannelPromise promise) {
569 writeFrameWithId(ctx, frame.type(), frame.id(), promise);
570 }
571
572 private static void writeMaxPushIdFrame(
573 ChannelHandlerContext ctx, Http3MaxPushIdFrame frame, ChannelPromise promise) {
574 writeFrameWithId(ctx, frame.type(), frame.id(), promise);
575 }
576
577 private static void writeFrameWithId(ChannelHandlerContext ctx, long type, long id, ChannelPromise promise) {
578 ByteBuf out = ctx.alloc().directBuffer(24);
579 writeVariableLengthInteger(out, type);
580 writeVariableLengthInteger(out, numBytesForVariableLengthInteger(id));
581 writeVariableLengthInteger(out, id);
582 ctx.write(out, promise);
583 }
584
585 private void writeUnknownFrame(
586 ChannelHandlerContext ctx, Http3UnknownFrame frame, ChannelPromise promise) {
587 long type = frame.type();
588 if (Http3CodecUtils.isReservedHttp2FrameType(type)) {
589 Http3Exception exception = new Http3Exception(Http3ErrorCode.H3_FRAME_UNEXPECTED,
590 "Reserved type for HTTP/2 send.");
591 promise.setFailure(exception);
592
593 connectionError(ctx, exception.errorCode(), exception.getMessage(), false);
594 return;
595 }
596 if (!Http3CodecUtils.isReservedFrameType(type)) {
597 Http3Exception exception = new Http3Exception(Http3ErrorCode.H3_FRAME_UNEXPECTED,
598 "Non reserved type for HTTP/3 send.");
599 promise.setFailure(exception);
600 return;
601 }
602 ByteBuf out = ctx.alloc().directBuffer();
603 writeVariableLengthInteger(out, type);
604 writeVariableLengthInteger(out, frame.content().readableBytes());
605 ByteBuf content = frame.content().retain();
606 ctx.write(Unpooled.wrappedUnmodifiableBuffer(out, content), promise);
607 }
608
609 private void initReadResumptionListenerIfRequired(ChannelHandlerContext ctx) {
610 if (readResumptionListener == null) {
611 readResumptionListener = new ReadResumptionListener(ctx, this);
612 }
613 }
614
615 private static void unsupported(ChannelPromise promise) {
616 promise.setFailure(new UnsupportedOperationException());
617 }
618
619 @Override
620 public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) {
621 ctx.bind(localAddress, promise);
622 }
623
624 @Override
625 public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress,
626 SocketAddress localAddress, ChannelPromise promise) {
627 ctx.connect(remoteAddress, localAddress, promise);
628 }
629
630 @Override
631 public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) {
632 ctx.disconnect(promise);
633 }
634
635 @Override
636 public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
637 ctx.close(promise);
638 }
639
640 @Override
641 public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) {
642 ctx.deregister(promise);
643 }
644
645 @Override
646 public void read(ChannelHandlerContext ctx) {
647 assert readResumptionListener != null;
648 if (readResumptionListener.readRequested()) {
649 ctx.read();
650 }
651 }
652
653 @Override
654 public void flush(ChannelHandlerContext ctx) {
655 if (writeResumptionListener != null) {
656 writeResumptionListener.enqueueFlush();
657 } else {
658 ctx.flush();
659 }
660 }
661
662 private static final class ReadResumptionListener
663 implements Runnable, GenericFutureListener<Future<? super QuicStreamChannel>> {
664 private static final int STATE_SUSPENDED = 0b1000_0000;
665 private static final int STATE_READ_PENDING = 0b0100_0000;
666 private static final int STATE_READ_COMPLETE_PENDING = 0b0010_0000;
667
668 private final ChannelHandlerContext ctx;
669 private final Http3FrameCodec codec;
670 private byte state;
671
672 ReadResumptionListener(ChannelHandlerContext ctx, Http3FrameCodec codec) {
673 this.ctx = ctx;
674 this.codec = codec;
675 assert codec.qpackAttributes != null;
676 if (!codec.qpackAttributes.dynamicTableDisabled() && !codec.qpackAttributes.decoderStreamAvailable()) {
677 codec.qpackAttributes.whenDecoderStreamAvailable(this);
678 }
679 }
680
681 void suspended() {
682 assert !codec.qpackAttributes.dynamicTableDisabled();
683 setState(STATE_SUSPENDED);
684 }
685
686 boolean readCompleted() {
687 if (hasState(STATE_SUSPENDED)) {
688 setState(STATE_READ_COMPLETE_PENDING);
689 return false;
690 }
691 return true;
692 }
693
694 boolean readRequested() {
695 if (hasState(STATE_SUSPENDED)) {
696 setState(STATE_READ_PENDING);
697 return false;
698 }
699 return true;
700 }
701
702 boolean isSuspended() {
703 return hasState(STATE_SUSPENDED);
704 }
705
706 @Override
707 public void operationComplete(Future<? super QuicStreamChannel> future) {
708 if (future.isSuccess()) {
709 resume();
710 } else {
711 ctx.fireExceptionCaught(future.cause());
712 }
713 }
714
715 @Override
716 public void run() {
717 resume();
718 }
719
720 private void resume() {
721 unsetState(STATE_SUSPENDED);
722 try {
723 codec.channelRead(ctx, Unpooled.EMPTY_BUFFER);
724 if (hasState(STATE_READ_COMPLETE_PENDING)) {
725 unsetState(STATE_READ_COMPLETE_PENDING);
726 codec.channelReadComplete(ctx);
727 }
728 if (hasState(STATE_READ_PENDING)) {
729 unsetState(STATE_READ_PENDING);
730 codec.read(ctx);
731 }
732 } catch (Exception e) {
733 ctx.fireExceptionCaught(e);
734 }
735 }
736
737 private void setState(int toSet) {
738 state |= toSet;
739 }
740
741 private boolean hasState(int toCheck) {
742 return (state & toCheck) == toCheck;
743 }
744
745 private void unsetState(int toUnset) {
746 state &= ~toUnset;
747 }
748 }
749
750 private static final class WriteResumptionListener
751 implements GenericFutureListener<Future<? super QuicStreamChannel>> {
752 private static final Object FLUSH = new Object();
753 private final PendingWriteQueue queue;
754 private final ChannelHandlerContext ctx;
755 private final Http3FrameCodec codec;
756
757 private WriteResumptionListener(ChannelHandlerContext ctx, Http3FrameCodec codec) {
758 this.ctx = ctx;
759 this.codec = codec;
760 queue = new PendingWriteQueue(ctx);
761 }
762
763 @Override
764 public void operationComplete(Future<? super QuicStreamChannel> future) {
765 drain();
766 }
767
768 void enqueue(Object msg, ChannelPromise promise) {
769 assert ctx.channel().eventLoop().inEventLoop();
770
771 ReferenceCountUtil.touch(msg);
772 queue.add(msg, promise);
773 }
774
775 void enqueueFlush() {
776 assert ctx.channel().eventLoop().inEventLoop();
777 queue.add(FLUSH, ctx.voidPromise());
778 }
779
780 void drain() {
781 assert ctx.channel().eventLoop().inEventLoop();
782 boolean flushSeen = false;
783 try {
784 for (;;) {
785 Object entry = queue.current();
786 if (entry == null) {
787 break;
788 }
789 if (entry == FLUSH) {
790 flushSeen = true;
791 queue.remove().trySuccess();
792 } else {
793
794 codec.write0(ctx, ReferenceCountUtil.retain(entry), queue.remove());
795 }
796 }
797
798
799 codec.writeResumptionListener = null;
800 } finally {
801 if (flushSeen) {
802 codec.flush(ctx);
803 }
804 }
805 }
806
807 static WriteResumptionListener newListener(ChannelHandlerContext ctx, Http3FrameCodec codec) {
808 WriteResumptionListener listener = new WriteResumptionListener(ctx, codec);
809 assert codec.qpackAttributes != null;
810 codec.qpackAttributes.whenEncoderStreamAvailable(listener);
811 return listener;
812 }
813 }
814
815
816
817
818 @FunctionalInterface
819 interface Http3FrameCodecFactory {
820
821
822
823
824
825
826
827
828 ChannelHandler newCodec(Http3FrameTypeValidator validator, Http3RequestStreamCodecState encodeState,
829 Http3RequestStreamCodecState decodeState,
830 Http3Settings.NonStandardHttp3SettingsValidator nonStandardSettingsValidator);
831 }
832 }