ReuseFdHandler Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/AbstractSocketReuseFdTest.java lines 119–185

    static class ReuseFdHandler extends ChannelInboundHandlerAdapter {
        private static final String EXPECTED_PAYLOAD = "payload";

        private final Promise<Void> donePromise;
        private final AtomicInteger remaining;
        private final boolean client;
        volatile Channel channel;
        final AtomicReference<Throwable> globalException;
        final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
        final StringBuilder received = new StringBuilder();

        ReuseFdHandler(
            boolean client,
            AtomicReference<Throwable> globalException,
            AtomicInteger remaining,
            Promise<Void> donePromise) {
            this.client = client;
            this.globalException = globalException;
            this.remaining = remaining;
            this.donePromise = donePromise;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            channel = ctx.channel();
            if (client) {
                ctx.writeAndFlush(Unpooled.copiedBuffer(EXPECTED_PAYLOAD, CharsetUtil.US_ASCII));
            }
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            if (msg instanceof ByteBuf) {
                ByteBuf buf = (ByteBuf) msg;
                received.append(buf.toString(CharsetUtil.US_ASCII));
                buf.release();

                if (received.toString().equals(EXPECTED_PAYLOAD)) {
                    if (client) {
                        ctx.close();
                    } else {
                        ctx.writeAndFlush(Unpooled.copiedBuffer(EXPECTED_PAYLOAD, CharsetUtil.US_ASCII));
                    }
                }
            }
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            if (exception.compareAndSet(null, cause)) {
                donePromise.tryFailure(new IllegalStateException("exceptionCaught: " + ctx.channel(), cause));
                ctx.close();
            }
            globalException.compareAndSet(null, cause);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) {
            if (remaining.decrementAndGet() == 0) {
                if (received.toString().equals(EXPECTED_PAYLOAD)) {
                    donePromise.setSuccess(null);
                } else {
                    donePromise.tryFailure(new Exception("Unexpected payload:" + received));
                }
            }
        }
    }