nestReentrancy() — netty Function Reference

Architecture documentation for the nestReentrancy() function in ReentrantChannelTest.java from the netty codebase.

Function java Buffer Telemetry

Entity Profile

Buffer→ Telemetry→ nestReentrancy() — netty Function Reference

Dependency Diagram

graph TD
  da3878d1_9916_f6ab_77a8_b57486aa58a3["nestReentrancy()"]
  13a11eea_27b7_44b5_a4d5_69eb21dd0e09["ReentrantChannelTest"]
  da3878d1_9916_f6ab_77a8_b57486aa58a3 -->|defined in| 13a11eea_27b7_44b5_a4d5_69eb21dd0e09
  style da3878d1_9916_f6ab_77a8_b57486aa58a3 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

transport/src/test/java/io/netty/channel/ReentrantChannelTest.java lines 292–398

    @Test
    public void nestReentrancy() throws Exception {
        EventLoopGroup group = new MultiThreadIoEventLoopGroup(1, LocalIoHandler.newFactory());
        try {
            LocalAddress addr = new LocalAddress("nestReentrancy");

            BlockingQueue<Object> received = new LinkedBlockingQueue<>();

            Channel server = new ServerBootstrap()
                    .group(group)
                    .channel(LocalServerChannel.class)
                    .childHandler(new ChannelInitializer<Channel>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            ch.config().setAutoRead(false);
                            // first handler splits input by \n and into strings
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                    String string = ((ByteBuf) msg).toString(StandardCharsets.UTF_8);
                                    ((ByteBuf) msg).release();
                                    for (String s : string.split("\n")) {
                                        ctx.fireChannelRead(s);
                                    }
                                }
                            });
                            // second handler buffers messages, sends them on in channelReadComplete, and acts as flow
                            // control
                            ch.pipeline().addLast(new ChannelDuplexHandler() {
                                final Queue<Object> queue = new ArrayDeque<>();
                                boolean demand = true;

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

                                @Override
                                public void channelReadComplete(ChannelHandlerContext ctx) {
                                    while (demand) {
                                        Object item = queue.poll();
                                        if (item == null) {
                                            break;
                                        }
                                        demand = false;
                                        ctx.fireChannelRead(item);
                                    }
                                    ctx.fireChannelReadComplete();
                                }

                                @Override
                                public void read(ChannelHandlerContext ctx) {
                                    Object item = queue.poll();
                                    if (item != null) {
                                        ctx.fireChannelRead(item);
                                    } else {
                                        demand = true;
                                        ctx.read();
                                    }
                                }

                                @Override
                                public void handlerAdded(ChannelHandlerContext ctx) {
                                    ctx.read();
                                }
                            });
                            // third handler saves incoming packets so that we can test their order
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                    received.add(msg);
                                    ctx.fireChannelRead(msg);
                                }
                            });
                            // final handler relieves backpressure, triggering reentrant channelReads
                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                                @Override
                                public void channelRead(ChannelHandlerContext ctx, Object msg) {
                                    ctx.read();
                                }