WriteResumptionListener Class — netty Architecture

Source Code

codec-http3/src/main/java/io/netty/handler/codec/http3/Http3FrameCodec.java lines 740–803

    private static final class WriteResumptionListener
            implements GenericFutureListener<Future<? super QuicStreamChannel>> {
        private static final Object FLUSH = new Object();
        private final PendingWriteQueue queue;
        private final ChannelHandlerContext ctx;
        private final Http3FrameCodec codec;

        private WriteResumptionListener(ChannelHandlerContext ctx, Http3FrameCodec codec) {
            this.ctx = ctx;
            this.codec = codec;
            queue = new PendingWriteQueue(ctx);
        }

        @Override
        public void operationComplete(Future<? super QuicStreamChannel> future) {
            drain();
        }

        void enqueue(Object msg, ChannelPromise promise) {
            assert ctx.channel().eventLoop().inEventLoop();
            // Touch the message to allow easier debugging of memory leaks
            ReferenceCountUtil.touch(msg);
            queue.add(msg, promise);
        }

        void enqueueFlush() {
            assert ctx.channel().eventLoop().inEventLoop();
            queue.add(FLUSH, ctx.voidPromise());
        }

        void drain() {
            assert ctx.channel().eventLoop().inEventLoop();
            boolean flushSeen = false;
            try {
                for (;;) {
                    Object entry = queue.current();
                    if (entry == null) {
                        break;
                    }
                    if (entry == FLUSH) {
                        flushSeen = true;
                        queue.remove().trySuccess();
                    } else {
                        // Retain the entry as remove() will call release() as well.
                        codec.write0(ctx, ReferenceCountUtil.retain(entry), queue.remove());
                    }
                }
                // indicate that writes do not need to be enqueued. As we are on the eventloop, no other writes can
                // happen while we are draining, hence we would not write out of order.
                codec.writeResumptionListener = null;
            } finally {
                if (flushSeen) {
                    codec.flush(ctx);
                }
            }
        }

        static WriteResumptionListener newListener(ChannelHandlerContext ctx, Http3FrameCodec codec) {
            WriteResumptionListener listener = new WriteResumptionListener(ctx, codec);
            assert codec.qpackAttributes != null;
            codec.qpackAttributes.whenEncoderStreamAvailable(listener);
            return listener;
        }
    }