ReadResumptionListener Class — netty Architecture

Source Code

codec-http3/src/main/java/io/netty/handler/codec/http3/Http3FrameCodec.java lines 652–738

    private static final class ReadResumptionListener
            implements Runnable, GenericFutureListener<Future<? super QuicStreamChannel>> {
        private static final int STATE_SUSPENDED = 0b1000_0000;
        private static final int STATE_READ_PENDING = 0b0100_0000;
        private static final int STATE_READ_COMPLETE_PENDING = 0b0010_0000;

        private final ChannelHandlerContext ctx;
        private final Http3FrameCodec codec;
        private byte state;

        ReadResumptionListener(ChannelHandlerContext ctx, Http3FrameCodec codec) {
            this.ctx = ctx;
            this.codec = codec;
            assert codec.qpackAttributes != null;
            if (!codec.qpackAttributes.dynamicTableDisabled() && !codec.qpackAttributes.decoderStreamAvailable()) {
                codec.qpackAttributes.whenDecoderStreamAvailable(this);
            }
        }

        void suspended() {
            assert !codec.qpackAttributes.dynamicTableDisabled();
            setState(STATE_SUSPENDED);
        }

        boolean readCompleted() {
            if (hasState(STATE_SUSPENDED)) {
                setState(STATE_READ_COMPLETE_PENDING);
                return false;
            }
            return true;
        }

        boolean readRequested() {
            if (hasState(STATE_SUSPENDED)) {
                setState(STATE_READ_PENDING);
                return false;
            }
            return true;
        }

        boolean isSuspended() {
            return hasState(STATE_SUSPENDED);
        }

        @Override
        public void operationComplete(Future<? super QuicStreamChannel> future) {
            if (future.isSuccess()) {
                resume();
            } else {
                ctx.fireExceptionCaught(future.cause());
            }
        }

        @Override
        public void run() {
            resume();
        }

        private void resume() {
            unsetState(STATE_SUSPENDED);
            try {
                codec.channelRead(ctx, Unpooled.EMPTY_BUFFER);
                if (hasState(STATE_READ_COMPLETE_PENDING)) {
                    unsetState(STATE_READ_COMPLETE_PENDING);
                    codec.channelReadComplete(ctx);
                }
                if (hasState(STATE_READ_PENDING)) {
                    unsetState(STATE_READ_PENDING);
                    codec.read(ctx);
                }
            } catch (Exception e) {
                ctx.fireExceptionCaught(e);
            }
        }

        private void setState(int toSet) {
            state |= toSet;
        }

        private boolean hasState(int toCheck) {
            return (state & toCheck) == toCheck;