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