LastInboundHandler Class — netty Architecture

Source Code

codec-http2/src/test/java/io/netty/handler/codec/http2/LastInboundHandler.java lines 36–222

public class LastInboundHandler extends ChannelDuplexHandler {
    private final List<Object> queue = new ArrayList<Object>();
    private final Consumer<ChannelHandlerContext> channelReadCompleteConsumer;
    private Throwable lastException;
    private ChannelHandlerContext ctx;
    private boolean channelActive;
    private String writabilityStates = "";

    // TODO(scott): use JDK 8's Consumer
    public interface Consumer<T> {
        void accept(T obj);
    }

    private static final Consumer<Object> NOOP_CONSUMER = new Consumer<Object>() {
        @Override
        public void accept(Object obj) {
        }
    };

    @SuppressWarnings("unchecked")
    public static <T> Consumer<T> noopConsumer() {
        return (Consumer<T>) NOOP_CONSUMER;
    }

    public LastInboundHandler() {
        this(LastInboundHandler.<ChannelHandlerContext>noopConsumer());
    }

    public LastInboundHandler(Consumer<ChannelHandlerContext> channelReadCompleteConsumer) {
        this.channelReadCompleteConsumer = checkNotNull(channelReadCompleteConsumer, "channelReadCompleteConsumer");
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        super.handlerAdded(ctx);
        this.ctx = ctx;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        if (channelActive) {
            throw new IllegalStateException("channelActive may only be fired once.");
        }
        channelActive = true;
        super.channelActive(ctx);
    }

    public boolean isChannelActive() {
        return channelActive;
    }

    public String writabilityStates() {
        return writabilityStates;
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if (!channelActive) {
            throw new IllegalStateException("channelInactive may only be fired once after channelActive.");
        }
        channelActive = false;
        super.channelInactive(ctx);
    }

    @Override
    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        if ("".equals(writabilityStates)) {
            writabilityStates = String.valueOf(ctx.channel().isWritable());
        } else {
            writabilityStates += "," + ctx.channel().isWritable();
        }
        super.channelWritabilityChanged(ctx);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        queue.add(msg);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {